From 602d8ef376da60229eae3a5385c07e5198aa14da Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Christian=20L=C3=BCck?= <christian@lueck.tv>
Date: Fri, 15 Sep 2017 17:39:54 +0200
Subject: [PATCH] Forward compatibility with Stream v1.0 and strict stream
 semantics

---
 composer.json                      | 2 +-
 src/Message/ReadableBodyStream.php | 8 ++++----
 tests/FunctionalBrowserTest.php    | 8 ++++----
 tests/Io/TransactionTest.php       | 4 ++--
 4 files changed, 11 insertions(+), 11 deletions(-)

diff --git a/composer.json b/composer.json
index 86b4d42..e0b96ef 100644
--- a/composer.json
+++ b/composer.json
@@ -20,7 +20,7 @@
         "react/promise": "^2.2.1",
         "react/promise-stream": "^0.1.1",
         "react/socket": "^0.8",
-        "react/stream": "^0.6 || ^0.5 || ^0.4.6",
+        "react/stream": "^1.0 || ^0.7 || ^0.6 || ^0.5 || ^0.4.6",
         "psr/http-message": "^1.0",
         "ringcentral/psr7": "^1.2"
     },
diff --git a/src/Message/ReadableBodyStream.php b/src/Message/ReadableBodyStream.php
index a577788..3610061 100644
--- a/src/Message/ReadableBodyStream.php
+++ b/src/Message/ReadableBodyStream.php
@@ -22,14 +22,14 @@ public function __construct(ReadableStreamInterface $input)
 
         $that = $this;
         $input->on('data', function ($data) use ($that) {
-            $that->emit('data', array($data, $that));
+            $that->emit('data', array($data));
         });
         $input->on('error', function ($error) use ($that) {
-            $that->emit('error', array($error, $that));
+            $that->emit('error', array($error));
             $that->close();
         });
         $input->on('end', function () use ($that) {
-            $that->emit('end', array($that));
+            $that->emit('end');
             $that->close();
         });
         $input->on('close', array($that, 'close'));
@@ -41,7 +41,7 @@ public function close()
             $this->closed = true;
             $this->input->close();
 
-            $this->emit('close', array($this));
+            $this->emit('close');
             $this->removeAllListeners();
         }
     }
diff --git a/tests/FunctionalBrowserTest.php b/tests/FunctionalBrowserTest.php
index c09c39a..e780b1c 100644
--- a/tests/FunctionalBrowserTest.php
+++ b/tests/FunctionalBrowserTest.php
@@ -5,11 +5,11 @@
 use Clue\React\Buzz\Io\Sender;
 use Clue\React\Buzz\Message\ResponseException;
 use Clue\React\Block;
-use React\Stream\ReadableStream;
 use Psr\Http\Message\ServerRequestInterface;
 use React\Http\Response;
 use React\Promise\Stream;
 use React\Socket\Connector;
+use React\Stream\ThroughStream;
 
 class FunctionalBrowserTest extends TestCase
 {
@@ -176,7 +176,7 @@ public function testPostStreamChunked()
 
         $this->base = str_replace('tcp:', 'http:', $socket->getAddress()) . '/';
 
-        $stream = new ReadableStream();
+        $stream = new ThroughStream();
 
         $this->loop->addTimer(0.001, function () use ($stream) {
             $stream->emit('data', array('hello world'));
@@ -195,7 +195,7 @@ public function testPostStreamChunked()
     /** @group online */
     public function testPostStreamKnownLength()
     {
-        $stream = new ReadableStream();
+        $stream = new ThroughStream();
 
         $this->loop->addTimer(0.001, function () use ($stream) {
             $stream->emit('data', array('hello world'));
@@ -211,7 +211,7 @@ public function testPostStreamKnownLength()
     /** @group online */
     public function testPostStreamClosed()
     {
-        $stream = new ReadableStream();
+        $stream = new ThroughStream();
         $stream->close();
 
         $response = Block\await($this->browser->post($this->base . 'post', array(), $stream), $this->loop);
diff --git a/tests/Io/TransactionTest.php b/tests/Io/TransactionTest.php
index 4e7f72d..947afc8 100644
--- a/tests/Io/TransactionTest.php
+++ b/tests/Io/TransactionTest.php
@@ -7,7 +7,7 @@
 use React\Promise;
 use Clue\React\Block;
 use React\EventLoop\Factory;
-use React\Stream\ReadableStream;
+use React\Stream\ThroughStream;
 
 class TransactionTest extends TestCase
 {
@@ -37,7 +37,7 @@ public function testReceivingStreamingBodyWillResolveWithBufferedResponseByDefau
         $messageFactory = new MessageFactory();
         $loop = Factory::create();
 
-        $stream = new ReadableStream();
+        $stream = new ThroughStream();
         $loop->addTimer(0.001, function () use ($stream) {
             $stream->emit('data', array('hello world'));
             $stream->close();