reactphp ^0.7 || ^1.0 || ^1.1 compatibility

CHANGELOG.md

@@ -1,6 +1,12 @@
 CHANGELOG
 =========
 
+* NEXT VERSION
+  * Bump react deps to 1.1.* (@phiamo)
+  * combined the in 0.6.0 existing ReadableStream and WritableStream
+    into the now also final InputStream and OutputStream classes
+  * adapted the cancelTimer API change in Client 
+
 * 0.2.0 (2014-11-29)
 
   * Bump react deps to 0.4.* (@Easen)

composer.json

@@ -5,14 +5,15 @@
     "license": "MIT",
     "require": {
         "php": ">=5.4",
-        "evenement/evenement": "~2.0",
-        "react/socket": "0.4.*",
+        "evenement/evenement": "^3.0 || ^2.0 || ^1.0",
+        "react/socket": "^1.0 || ^0.7",
+        "react/event-loop": "^1.0 || ^0.5",
         "react/promise": "~2.0"
     },
     "autoload": {
         "psr-4": { "React\\Stomp\\": "src" }
     },
     "require-dev": {
-        "phpunit/phpunit": "^7.0 || ^6.4 || ^5.7 || ^4.8.35"
+        "phpunit/phpunit": "^9.0 || ^8.0 || ^7.0 || ^6.4 || ^5.7 || ^4.8.35"
     }
 }

src/Client.php

@@ -50,8 +50,8 @@ public function __construct(LoopInterface $loop, InputStreamInterface $input, Ou
         $this->input->on('frame', array($this, 'handleFrameEvent'));
         $this->input->on('error', array($this, 'handleErrorEvent'));
         $this->input->on('close', array($this, 'handleCloseEvent'));
+        $this->input->on('heart-beat', array($this, 'handleHeartbeatEvent'));
         $this->output = $output;
-
         $this->options = $this->sanatizeOptions($options);
     }
 
@@ -65,22 +65,23 @@ public function connect($timeout = 5)
 
         $deferred = $this->connectDeferred = new Deferred();
         $client = $this;
-
+        $loop = $this->loop;
         $timer = $this->loop->addTimer($timeout, function () use ($client, $deferred) {
             $deferred->reject(new ConnectionException('Connection timeout'));
             $client->resetConnectDeferred();
             $client->setConnectionStatus('not-connected');
         });
 
-        $this->on('connect', function ($client) use ($timer, $deferred) {
-            $timer->cancel();
+        $this->on('connect', function ($client) use ($timer, $deferred, $loop) {
+            $loop->cancelTimer($timer);
             $deferred->resolve($client);
         });
 
         $frame = $this->packageCreator->connect(
             $this->options['vhost'],
             $this->options['login'],
-            $this->options['passcode']
+            $this->options['passcode'],
+            $this->options['heart-beat']
         );
         $this->output->sendFrame($frame);
 
@@ -190,6 +191,11 @@ public function handleCloseEvent()
         $this->emit('close');
     }
 
+    public function handleHeartbeatEvent()
+    {
+        $this->emit('heart-beat');
+    }
+
     public function processFrame(Frame $frame)
     {
         $command = $this->packageProcessor->receiveFrame($frame);
@@ -250,6 +256,7 @@ private function sanatizeOptions($options)
             'vhost'     => isset($options['host']) ? $options['host'] : null,
             'login'     => null,
             'passcode'  => null,
+            'heart-beat'  => null,
         ), $options);
     }
 

src/Client/OutgoingPackageCreator.php

@@ -13,7 +13,7 @@ public function __construct(State $state)
         $this->state = $state;
     }
 
-    public function connect($host, $login = null, $passcode = null)
+    public function connect($host, $login = null, $passcode = null, $heartbeat = null)
     {
         $this->state->startConnecting();
 
@@ -24,6 +24,12 @@ public function connect($host, $login = null, $passcode = null)
                 'passcode'  => (string) $passcode,
             ));
         }
+        if (null !== $heartbeat) {
+            $headers = array_merge($headers, array(
+                'heart-beat' => $heartbeat,
+            ));
+        }
+
         return new Frame('CONNECT', $headers);
     }
 

src/Factory.php

@@ -17,6 +17,7 @@ class Factory
         'vhost'     => '/',
         'login'     => 'guest',
         'passcode'  => 'guest',
+        'heart-beat' => null
     );
 
     private $loop;

src/Io/InputStream.php

@@ -2,8 +2,10 @@
 
 namespace React\Stomp\Io;
 
+use React\Stomp\Protocol\Frame;
 use React\Stomp\Protocol\Parser;
-use React\Stream\WritableStream;
+use Evenement\EventEmitter;
+use React\Stream\WritableStreamInterface;
 
 // $parser = new Parser();
 // $input = new InputStream($parser);
@@ -12,8 +14,9 @@
 // });
 // $conn->pipe($input);
 
-class InputStream extends WritableStream implements InputStreamInterface
+final class InputStream extends EventEmitter implements WritableStreamInterface, InputStreamInterface
 {
+    protected $closed = false;
     private $buffer = '';
     private $parser;
 
@@ -24,6 +27,11 @@ public function __construct(Parser $parser)
 
     public function write($data)
     {
+        if ($data === "\x0a") {
+            $this->emit('heart-beat', [new Frame('MESSAGE\nHEART-BEAT')]);
+            $data = '';
+        }
+
         $data = $this->buffer.$data;
         list($frames, $data) = $this->parser->parse($data);
         $this->buffer = $data;
@@ -32,4 +40,30 @@ public function write($data)
             $this->emit('frame', array($frame));
         }
     }
+
+    public function end($data = null)
+    {
+        if (null !== $data) {
+            $this->write($data);
+        }
+
+        $this->close();
+    }
+
+    public function isWritable()
+    {
+        return !$this->closed;
+    }
+
+    public function close()
+    {
+        if ($this->closed) {
+            return;
+        }
+
+        $this->closed = true;
+        $this->emit('close');
+        $this->removeAllListeners();
+    }
+
 }

src/Io/OutputStream.php

@@ -4,14 +4,19 @@
 
 use React\EventLoop\LoopInterface;
 use React\Stomp\Protocol\Frame;
-use React\Stream\ReadableStream;
+use React\Stream\ReadableStreamInterface;
+use React\Stream\Util;
+use React\Stream\WritableStreamInterface;
+use Evenement\EventEmitter;
 
 // $output = new OutputStream();
 // $output->pipe($conn);
 // $output->sendFrame($frame);
 
-class OutputStream extends ReadableStream implements OutputStreamInterface
+final class OutputStream extends EventEmitter implements ReadableStreamInterface, OutputStreamInterface
 {
+    protected $closed = false;
+
     private $loop;
     private $paused = false;
     private $bufferedFrames = array();
@@ -32,6 +37,28 @@ public function sendFrame(Frame $frame)
         $this->emit('data', array($data));
     }
 
+
+
+    public function sendBufferedFrames()
+    {
+        if ($this->paused) {
+            return;
+        }
+
+        while ($frame = array_shift($this->bufferedFrames)) {
+            $this->sendFrame($frame);
+
+            if ($this->paused) {
+                return;
+            }
+        }
+    }
+    public function isReadable()
+    {
+        return !$this->closed;
+    }
+
+
     public function pause()
     {
         $this->paused = true;
@@ -44,18 +71,19 @@ public function resume()
         $this->loop->addTimer(0.001, array($this, 'sendBufferedFrames'));
     }
 
-    public function sendBufferedFrames()
+    public function pipe(WritableStreamInterface $dest, array $options = array())
     {
-        if ($this->paused) {
+        return Util::pipe($this, $dest, $options);
+    }
+
+    public function close()
+    {
+        if ($this->closed) {
             return;
         }
 
-        while ($frame = array_shift($this->bufferedFrames)) {
-            $this->sendFrame($frame);
-
-            if ($this->paused) {
-                return;
-            }
-        }
+        $this->closed = true;
+        $this->emit('close');
+        $this->removeAllListeners();
     }
 }

tests/React/Tests/Stomp/ClientTest.php

@@ -215,7 +215,9 @@ public function itShouldThrowAnExceptionOnConnectedFrameOutsideWindow()
         $errors = array();
 
         $client = new Client($loop, $input, $output, array('vhost' => 'localhost'));
-        $client->on('error', function ($error) use (&$errors) { $errors[] = $error; });
+        $client->on('error', function ($error) use (&$errors) {
+            $errors[] = $error;
+        });
 
         $frame = new Frame('CONNECTED', array('session' => '1234', 'server' => 'React/alpha'));
         $input->emit('frame', array($frame));
@@ -537,10 +539,10 @@ public function messagesShouldGetRoutedToSubscriptions()
     }
 
     /**
-    * @test
-    * @depends messagesShouldGetRoutedToSubscriptions
-    * @doesNotPerformAssertions
-    */
+     * @test
+     * @depends messagesShouldGetRoutedToSubscriptions
+     * @doesNotPerformAssertions
+     */
     public function callbackShouldNotBeCalledAfterUnsubscribe($data)
     {
         list($input, $output, $client, $capturedFrame) = $data;
@@ -708,7 +710,12 @@ private function createLoopMock()
 
     private function createLoopMockWithConnectionTimer()
     {
+        $timer = $this->createMock('React\EventLoop\TimerInterface');
+
         $loop = $this->createLoopMock();
+        $loop->expects($this->once())
+            ->method('cancelTimer')
+            ->with($timer);
 
         $timer = $this->getMockBuilder('React\EventLoop\Timer\TimerInterface')->getMock();
         $timer->expects($this->once())

tests/React/Tests/Stomp/Io/OutputStreamTest.php

@@ -7,7 +7,7 @@
 
 class OutputStreamTest extends TestCase
 {
-    public function setUp()
+    public function setUp(): void
     {
         $this->loop = $this->getMockBuilder('React\EventLoop\LoopInterface')->getMock();
         $this->loop

tests/bootstrap.php

@@ -1,5 +1,5 @@
 <?php
 
-$loader = require __DIR__.'/../vendor/autoload.php';
+$loader = require __DIR__ . '/../vendor/autoload.php';
 $loader->add('React\Tests', __DIR__);
 $loader->add('React\Functional', __DIR__);