Merge pull request #121 from clue-labs/removereadstream

jsor · web-flow · commit bbf5938885b4 · 2017-11-10T19:13:10.000+01:00
Fix forward compatibility with upcoming EventLoop releases
diff --git a/src/DuplexResourceStream.php b/src/DuplexResourceStream.php
@@ -33,6 +33,7 @@ final class DuplexResourceStream extends EventEmitter implements DuplexStreamInt
     private $readable = true;
     private $writable = true;
     private $closing = false;
+    private $listening = false;
 
     public function __construct($stream, LoopInterface $loop, $readChunkSize = null, WritableStreamInterface $buffer = null)
     {
@@ -100,13 +101,17 @@ public function isWritable()
 
     public function pause()
     {
-        $this->loop->removeReadStream($this->stream);
+        if ($this->listening) {
+            $this->loop->removeReadStream($this->stream);
+            $this->listening = false;
+        }
     }
 
     public function resume()
     {
-        if ($this->readable) {
+        if (!$this->listening && $this->readable) {
             $this->loop->addReadStream($this->stream, array($this, 'handleData'));
+            $this->listening = true;
         }
     }
 
@@ -131,11 +136,13 @@ public function close()
         $this->writable = false;
 
         $this->emit('close');
-        $this->loop->removeStream($this->stream);
+        $this->pause();
         $this->buffer->close();
         $this->removeAllListeners();
 
-        $this->handleClose();
+        if (is_resource($this->stream)) {
+            fclose($this->stream);
+        }
     }
 
     public function end($data = null)
@@ -191,14 +198,6 @@ public function handleData($stream)
         }
     }
 
-    /** @internal */
-    public function handleClose()
-    {
-        if (is_resource($this->stream)) {
-            fclose($this->stream);
-        }
-    }
-
     /**
      * Returns whether this is a pipe resource in a legacy environment
      *
diff --git a/src/ReadableResourceStream.php b/src/ReadableResourceStream.php
@@ -36,6 +36,7 @@ final class ReadableResourceStream extends EventEmitter implements ReadableStrea
     private $bufferSize;
 
     private $closed = false;
+    private $listening = false;
 
     public function __construct($stream, LoopInterface $loop, $readChunkSize = null)
     {
@@ -81,13 +82,17 @@ public function isReadable()
 
     public function pause()
     {
-        $this->loop->removeReadStream($this->stream);
+        if ($this->listening) {
+            $this->loop->removeReadStream($this->stream);
+            $this->listening = false;
+        }
     }
 
     public function resume()
     {
-        if (!$this->closed) {
+        if (!$this->listening && !$this->closed) {
             $this->loop->addReadStream($this->stream, array($this, 'handleData'));
+            $this->listening = true;
         }
     }
 
@@ -105,10 +110,12 @@ public function close()
         $this->closed = true;
 
         $this->emit('close');
-        $this->loop->removeStream($this->stream);
+        $this->pause();
         $this->removeAllListeners();
 
-        $this->handleClose();
+        if (is_resource($this->stream)) {
+            fclose($this->stream);
+        }
     }
 
     /** @internal */
@@ -144,14 +151,6 @@ public function handleData()
         }
     }
 
-    /** @internal */
-    public function handleClose()
-    {
-        if (is_resource($this->stream)) {
-            fclose($this->stream);
-        }
-    }
-
     /**
      * Returns whether this is a pipe resource in a legacy environment
      *
diff --git a/tests/DuplexResourceStreamTest.php b/tests/DuplexResourceStreamTest.php
@@ -272,12 +272,85 @@ public function testEndRemovesReadStreamFromLoop()
     {
         $stream = fopen('php://temp', 'r+');
         $loop = $this->createLoopMock();
-        $loop->expects($this->once())->method('removeReadStream');
+        $loop->expects($this->once())->method('addReadStream')->with($stream);
+        $loop->expects($this->once())->method('removeReadStream')->with($stream);
 
         $conn = new DuplexResourceStream($stream, $loop);
         $conn->end('bye');
     }
 
+    /**
+     * @covers React\Stream\DuplexResourceStream::pause
+     */
+    public function testPauseRemovesReadStreamFromLoop()
+    {
+        $stream = fopen('php://temp', 'r+');
+        $loop = $this->createLoopMock();
+        $loop->expects($this->once())->method('addReadStream')->with($stream);
+        $loop->expects($this->once())->method('removeReadStream')->with($stream);
+
+        $conn = new DuplexResourceStream($stream, $loop);
+        $conn->pause();
+        $conn->pause();
+    }
+
+    /**
+     * @covers React\Stream\DuplexResourceStream::pause
+     */
+    public function testResumeDoesAddStreamToLoopOnlyOnce()
+    {
+        $stream = fopen('php://temp', 'r+');
+        $loop = $this->createLoopMock();
+        $loop->expects($this->once())->method('addReadStream')->with($stream);
+
+        $conn = new DuplexResourceStream($stream, $loop);
+        $conn->resume();
+        $conn->resume();
+    }
+
+    /**
+     * @covers React\Stream\DuplexResourceStream::close
+     */
+    public function testCloseRemovesReadStreamFromLoop()
+    {
+        $stream = fopen('php://temp', 'r+');
+        $loop = $this->createLoopMock();
+        $loop->expects($this->once())->method('addReadStream')->with($stream);
+        $loop->expects($this->once())->method('removeReadStream')->with($stream);
+
+        $conn = new DuplexResourceStream($stream, $loop);
+        $conn->close();
+    }
+
+    /**
+     * @covers React\Stream\DuplexResourceStream::close
+     */
+    public function testCloseAfterPauseRemovesReadStreamFromLoopOnlyOnce()
+    {
+        $stream = fopen('php://temp', 'r+');
+        $loop = $this->createLoopMock();
+        $loop->expects($this->once())->method('addReadStream')->with($stream);
+        $loop->expects($this->once())->method('removeReadStream')->with($stream);
+
+        $conn = new DuplexResourceStream($stream, $loop);
+        $conn->pause();
+        $conn->close();
+    }
+
+    /**
+     * @covers React\Stream\DuplexResourceStream::close
+     */
+    public function testResumeAfterCloseDoesAddReadStreamToLoopOnlyOnce()
+    {
+        $stream = fopen('php://temp', 'r+');
+        $loop = $this->createLoopMock();
+        $loop->expects($this->once())->method('addReadStream')->with($stream);
+
+        $conn = new DuplexResourceStream($stream, $loop);
+        $conn->close();
+        $conn->resume();
+    }
+
     public function testEndedStreamsShouldNotWrite()
     {
         $file = tempnam(sys_get_temp_dir(), 'reactphptest_');
diff --git a/tests/ReadableResourceStreamTest.php b/tests/ReadableResourceStreamTest.php
@@ -234,6 +234,78 @@ public function testClosingStreamInDataEventShouldNotTriggerError()
         $conn->handleData($stream);
     }
 
+    /**
+     * @covers React\Stream\ReadableResourceStream::pause
+     */
+    public function testPauseRemovesReadStreamFromLoop()
+    {
+        $stream = fopen('php://temp', 'r+');
+        $loop = $this->createLoopMock();
+        $loop->expects($this->once())->method('addReadStream')->with($stream);
+        $loop->expects($this->once())->method('removeReadStream')->with($stream);
+
+        $conn = new ReadableResourceStream($stream, $loop);
+        $conn->pause();
+        $conn->pause();
+    }
+
+    /**
+     * @covers React\Stream\ReadableResourceStream::pause
+     */
+    public function testResumeDoesAddStreamToLoopOnlyOnce()
+    {
+        $stream = fopen('php://temp', 'r+');
+        $loop = $this->createLoopMock();
+        $loop->expects($this->once())->method('addReadStream')->with($stream);
+
+        $conn = new ReadableResourceStream($stream, $loop);
+        $conn->resume();
+        $conn->resume();
+    }
+
+    /**
+     * @covers React\Stream\ReadableResourceStream::close
+     */
+    public function testCloseRemovesReadStreamFromLoop()
+    {
+        $stream = fopen('php://temp', 'r+');
+        $loop = $this->createLoopMock();
+        $loop->expects($this->once())->method('addReadStream')->with($stream);
+        $loop->expects($this->once())->method('removeReadStream')->with($stream);
+
+        $conn = new ReadableResourceStream($stream, $loop);
+        $conn->close();
+    }
+
+    /**
+     * @covers React\Stream\ReadableResourceStream::close
+     */
+    public function testCloseAfterPauseRemovesReadStreamFromLoopOnce()
+    {
+        $stream = fopen('php://temp', 'r+');
+        $loop = $this->createLoopMock();
+        $loop->expects($this->once())->method('addReadStream')->with($stream);
+        $loop->expects($this->once())->method('removeReadStream')->with($stream);
+
+        $conn = new ReadableResourceStream($stream, $loop);
+        $conn->pause();
+        $conn->close();
+    }
+
+    /**
+     * @covers React\Stream\ReadableResourceStream::close
+     */
+    public function testResumeAfterCloseDoesAddReadStreamToLoopOnlyOnce()
+    {
+        $stream = fopen('php://temp', 'r+');
+        $loop = $this->createLoopMock();
+        $loop->expects($this->once())->method('addReadStream')->with($stream);
+
+        $conn = new ReadableResourceStream($stream, $loop);
+        $conn->close();
+        $conn->resume();
+    }
+
     /**
      * @covers React\Stream\ReadableResourceStream::handleData
      */

Original file line number	Diff line number	Diff line change
`@@ -33,6 +33,7 @@ final class DuplexResourceStream extends EventEmitter implements DuplexStreamInt`
`33`	`33`	`private $readable = true;`
`34`	`34`	`private $writable = true;`
`35`	`35`	`private $closing = false;`
	`36`	`+ private $listening = false;`
`36`	`37`
`37`	`38`	`public function __construct($stream, LoopInterface $loop, $readChunkSize = null, WritableStreamInterface $buffer = null)`
`38`	`39`	`{`
`@@ -100,13 +101,17 @@ public function isWritable()`
`100`	`101`
`101`	`102`	`public function pause()`
`102`	`103`	`{`
`103`		`- $this->loop->removeReadStream($this->stream);`
	`104`	`+ if ($this->listening) {`
	`105`	`+ $this->loop->removeReadStream($this->stream);`
	`106`	`+ $this->listening = false;`
	`107`	`+ }`
`104`	`108`	`}`
`105`	`109`
`106`	`110`	`public function resume()`
`107`	`111`	`{`
`108`		`- if ($this->readable) {`
	`112`	`+ if (!$this->listening && $this->readable) {`
`109`	`113`	`$this->loop->addReadStream($this->stream, array($this, 'handleData'));`
	`114`	`+ $this->listening = true;`
`110`	`115`	`}`
`111`	`116`	`}`
`112`	`117`
`@@ -131,11 +136,13 @@ public function close()`
`131`	`136`	`$this->writable = false;`
`132`	`137`
`133`	`138`	`$this->emit('close');`
`134`		`- $this->loop->removeStream($this->stream);`
	`139`	`+ $this->pause();`
`135`	`140`	`$this->buffer->close();`
`136`	`141`	`$this->removeAllListeners();`
`137`	`142`
`138`		`- $this->handleClose();`
	`143`	`+ if (is_resource($this->stream)) {`
	`144`	`+ fclose($this->stream);`
	`145`	`+ }`
`139`	`146`	`}`
`140`	`147`
`141`	`148`	`public function end($data = null)`
`@@ -191,14 +198,6 @@ public function handleData($stream)`
`191`	`198`	`}`
`192`	`199`	`}`
`193`	`200`
`194`		`- /** @internal */`
`195`		`- public function handleClose()`
`196`		`- {`
`197`		`- if (is_resource($this->stream)) {`
`198`		`- fclose($this->stream);`
`199`		`- }`
`200`		`- }`
`201`		`-`
`202`	`201`	`/**`
`203`	`202`	`* Returns whether this is a pipe resource in a legacy environment`
`204`	`203`	`*`
Original file line number	Diff line number	Diff line change
`@@ -36,6 +36,7 @@ final class ReadableResourceStream extends EventEmitter implements ReadableStrea`
`36`	`36`	`private $bufferSize;`
`37`	`37`
`38`	`38`	`private $closed = false;`
	`39`	`+ private $listening = false;`
`39`	`40`
`40`	`41`	`public function __construct($stream, LoopInterface $loop, $readChunkSize = null)`
`41`	`42`	`{`
`@@ -81,13 +82,17 @@ public function isReadable()`
`81`	`82`
`82`	`83`	`public function pause()`
`83`	`84`	`{`
`84`		`- $this->loop->removeReadStream($this->stream);`
	`85`	`+ if ($this->listening) {`
	`86`	`+ $this->loop->removeReadStream($this->stream);`
	`87`	`+ $this->listening = false;`
	`88`	`+ }`
`85`	`89`	`}`
`86`	`90`
`87`	`91`	`public function resume()`
`88`	`92`	`{`
`89`		`- if (!$this->closed) {`
	`93`	`+ if (!$this->listening && !$this->closed) {`
`90`	`94`	`$this->loop->addReadStream($this->stream, array($this, 'handleData'));`
	`95`	`+ $this->listening = true;`
`91`	`96`	`}`
`92`	`97`	`}`
`93`	`98`
`@@ -105,10 +110,12 @@ public function close()`
`105`	`110`	`$this->closed = true;`
`106`	`111`
`107`	`112`	`$this->emit('close');`
`108`		`- $this->loop->removeStream($this->stream);`
	`113`	`+ $this->pause();`
`109`	`114`	`$this->removeAllListeners();`
`110`	`115`
`111`		`- $this->handleClose();`
	`116`	`+ if (is_resource($this->stream)) {`
	`117`	`+ fclose($this->stream);`
	`118`	`+ }`
`112`	`119`	`}`
`113`	`120`
`114`	`121`	`/** @internal */`
`@@ -144,14 +151,6 @@ public function handleData()`
`144`	`151`	`}`
`145`	`152`	`}`
`146`	`153`
`147`		`- /** @internal */`
`148`		`- public function handleClose()`
`149`		`- {`
`150`		`- if (is_resource($this->stream)) {`
`151`		`- fclose($this->stream);`
`152`		`- }`
`153`		`- }`
`154`		`-`
`155`	`154`	`/**`
`156`	`155`	`* Returns whether this is a pipe resource in a legacy environment`
`157`	`156`	`*`