Forward close event only once

src/Io/ConnectionUpcaster.php

@@ -22,6 +22,7 @@ public function __construct(DuplexStreamInterface $stream)
         $this->stream = $stream;
 
         Util::forwardEvents($stream, $this, array('data', 'end', 'close', 'error', 'drain'));
+        $this->stream->on('close', array($this, 'close'));
     }
 
     public function isReadable()

src/Message/ReadableBodyStream.php

@@ -2,10 +2,9 @@
 
 namespace Clue\React\Buzz\Message;
 
+use Evenement\EventEmitter;
 use Psr\Http\Message\StreamInterface;
-use React\Stream\ReadableStream;
 use React\Stream\ReadableStreamInterface;
-use Evenement\EventEmitter;
 use React\Stream\Util;
 use React\Stream\WritableStreamInterface;
 
@@ -14,6 +13,9 @@
  */
 class ReadableBodyStream extends EventEmitter implements ReadableStreamInterface, StreamInterface
 {
+    private $input;
+    private $closed = false;
+
     public function __construct(ReadableStreamInterface $input)
     {
         $this->input = $input;
@@ -24,19 +26,24 @@ public function __construct(ReadableStreamInterface $input)
         });
         $input->on('error', function ($error) use ($that) {
             $that->emit('error', array($error, $that));
+            $that->close();
         });
         $input->on('end', function () use ($that) {
             $that->emit('end', array($that));
-            $that->emit('close', array($that));
-        });
-        $input->on('close', function () use ($that) {
-            $that->emit('close', array($that));
+            $that->close();
         });
+        $input->on('close', array($that, 'close'));
     }
 
     public function close()
     {
-        $this->input->close();
+        if (!$this->closed) {
+            $this->closed = true;
+            $this->input->close();
+
+            $this->emit('close', array($this));
+            $this->removeAllListeners();
+        }
     }
 
     public function isReadable()

tests/FunctionalBrowserTest.php

@@ -154,6 +154,7 @@ public function testPostString()
         $this->assertEquals('hello world', $data['data']);
     }
 
+    /** @group online */
     public function testPostStreamChunked()
     {
         // httpbin used to support `Transfer-Encoding: chunked` for requests,

tests/Message/ReadableBodyStreamTest.php

@@ -1,6 +1,7 @@
 <?php
 
 use Clue\React\Buzz\Message\ReadableBodyStream;
+use React\Stream\ThroughStream;
 
 class ReadableBodyStreamTest extends TestCase
 {
@@ -34,6 +35,54 @@ public function testCloseWillCloseInputStream()
         $this->stream->close();
     }
 
+    public function testCloseWillEmitCloseEvent()
+    {
+        $this->input = new ThroughStream();
+        $this->stream = new ReadableBodyStream($this->input);
+
+        $called = 0;
+        $this->stream->on('close', function () use (&$called) {
+            ++$called;
+        });
+
+        $this->stream->close();
+        $this->stream->close();
+
+        $this->assertEquals(1, $called);
+    }
+
+    public function testCloseInputWillEmitCloseEvent()
+    {
+        $this->input = new ThroughStream();
+        $this->stream = new ReadableBodyStream($this->input);
+
+        $called = 0;
+        $this->stream->on('close', function () use (&$called) {
+            ++$called;
+        });
+
+        $this->input->close();
+        $this->input->close();
+
+        $this->assertEquals(1, $called);
+    }
+
+    public function testEndInputWillEmitCloseEvent()
+    {
+        $this->input = new ThroughStream();
+        $this->stream = new ReadableBodyStream($this->input);
+
+        $called = 0;
+        $this->stream->on('close', function () use (&$called) {
+            ++$called;
+        });
+
+        $this->input->end();
+        $this->input->end();
+
+        $this->assertEquals(1, $called);
+    }
+
     public function testPauseWillPauseInputStream()
     {
         $this->input->expects($this->once())->method('pause');