From d5ade6b4776437e96bbb968642fbc5ad89aeb577 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Christian=20L=C3=BCck?= <christian@lueck.tv>
Date: Mon, 15 Aug 2016 14:34:07 +0200
Subject: [PATCH 1/2] Inject Buffer into Stream

---
 src/Stream.php       | 11 +++++++++--
 tests/StreamTest.php | 15 +++++++++++++++
 2 files changed, 24 insertions(+), 2 deletions(-)

diff --git a/src/Stream.php b/src/Stream.php
index 2e6126d..f133cb7 100644
--- a/src/Stream.php
+++ b/src/Stream.php
@@ -33,7 +33,7 @@ class Stream extends EventEmitter implements DuplexStreamInterface
     protected $loop;
     protected $buffer;
 
-    public function __construct($stream, LoopInterface $loop)
+    public function __construct($stream, LoopInterface $loop, WritableStreamInterface $buffer = null)
     {
         $this->stream = $stream;
         if (!is_resource($this->stream) || get_resource_type($this->stream) !== "stream") {
@@ -52,8 +52,12 @@ public function __construct($stream, LoopInterface $loop)
             stream_set_read_buffer($this->stream, 0);
         }
 
+        if ($buffer === null) {
+            $buffer = new Buffer($stream, $loop);
+        }
+
         $this->loop = $loop;
-        $this->buffer = new Buffer($this->stream, $this->loop);
+        $this->buffer = $buffer;
 
         $that = $this;
 
@@ -182,6 +186,9 @@ public function handleClose()
         }
     }
 
+    /**
+     * @return WritableStreamInterface|Buffer
+     */
     public function getBuffer()
     {
         return $this->buffer;
diff --git a/tests/StreamTest.php b/tests/StreamTest.php
index d671432..3ac2d79 100644
--- a/tests/StreamTest.php
+++ b/tests/StreamTest.php
@@ -28,6 +28,21 @@ public function testConstructorThrowsExceptionOnInvalidStream()
         $conn = new Stream('breakme', $loop);
     }
 
+    /**
+     * @covers React\Stream\Stream::__construct
+     */
+    public function testConstructorAcceptsBuffer()
+    {
+        $stream = fopen('php://temp', 'r+');
+        $loop = $this->createLoopMock();
+
+        $buffer = $this->getMock('React\Stream\WritableStreamInterface');
+
+        $conn = new Stream($stream, $loop, $buffer);
+
+        $this->assertSame($buffer, $conn->getBuffer());
+    }
+
     /**
      * @covers React\Stream\Stream::__construct
      * @covers React\Stream\Stream::handleData

From 23d51f08d256565b71f3aaaf8ac55935906ed246 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Christian=20L=C3=BCck?= <christian@lueck.tv>
Date: Mon, 15 Aug 2016 01:01:43 +0200
Subject: [PATCH 2/2] Simplify interaction between Stream and Buffer

---
 src/Stream.php | 7 +++----
 1 file changed, 3 insertions(+), 4 deletions(-)

diff --git a/src/Stream.php b/src/Stream.php
index f133cb7..7379c67 100644
--- a/src/Stream.php
+++ b/src/Stream.php
@@ -63,9 +63,10 @@ public function __construct($stream, LoopInterface $loop, WritableStreamInterfac
 
         $this->buffer->on('error', function ($error) use ($that) {
             $that->emit('error', array($error, $that));
-            $that->close();
         });
 
+        $this->buffer->on('close', array($this, 'close'));
+
         $this->buffer->on('drain', function () use ($that) {
             $that->emit('drain', array($that));
         });
@@ -118,7 +119,7 @@ public function close()
         $this->emit('end', array($this));
         $this->emit('close', array($this));
         $this->loop->removeStream($this->stream);
-        $this->buffer->removeAllListeners();
+        $this->buffer->close();
         $this->removeAllListeners();
 
         $this->handleClose();
@@ -135,8 +136,6 @@ public function end($data = null)
         $this->readable = false;
         $this->writable = false;
 
-        $this->buffer->on('close', array($this, 'close'));
-
         $this->buffer->end($data);
     }