Do not resume() stream if handler explicitly calls pause()

Author: clue

src/Io/PauseBufferStream.php

@@ -30,6 +30,7 @@ class PauseBufferStream extends EventEmitter implements ReadableStreamInterface
     private $endPaused = false;
     private $closePaused = false;
     private $errorPaused = null;
+    private $implicit = false;
 
     public function __construct(ReadableStreamInterface $input)
     {
@@ -41,6 +42,29 @@ public function __construct(ReadableStreamInterface $input)
         $this->input->on('close', array($this, 'handleClose'));
     }
 
+    /**
+     * pause and remember this was not explicitly from user control
+     *
+     * @internal
+     */
+    public function pauseImplicit()
+    {
+        $this->pause();
+        $this->implicit = true;
+    }
+
+    /**
+     * resume only if this was previously paused implicitly and not explicitly from user control
+     *
+     * @internal
+     */
+    public function resumeImplicit()
+    {
+        if ($this->implicit) {
+            $this->resume();
+        }
+    }
+
     public function isReadable()
     {
         return !$this->closed;
@@ -54,6 +78,7 @@ public function pause()
 
         $this->input->pause();
         $this->paused = true;
+        $this->implicit = false;
     }
 
     public function resume()
@@ -63,6 +88,7 @@ public function resume()
         }
 
         $this->paused = false;
+        $this->implicit = false;
 
         if ($this->dataPaused !== '') {
             $this->emit('data', array($this->dataPaused));

src/Middleware/LimitHandlersMiddleware.php

@@ -84,32 +84,33 @@ public function __invoke(ServerRequestInterface $request, $next)
     {
         $body = $request->getBody();
         if ($body instanceof ReadableStreamInterface) {
-            // replace with buffering body to ensure any readable events will be buffered
-            $body = new HttpBodyStream(
-                new PauseBufferStream($body),
-                $body->getSize()
-            );
-
             // pause actual body to stop emitting data until the handler is called
-            $request = $request->withBody($body);
-            $body->pause();
+            $size = $body->getSize();
+            $body = new PauseBufferStream($body);
+            $body->pauseImplicit();
+
+            // replace with buffering body to ensure any readable events will be buffered
+            $request = $request->withBody(new HttpBodyStream(
+                $body,
+                $size
+            ));
         }
+
         $deferred = new Deferred();
         $this->queued->enqueue($deferred);
 
         $this->processQueue();
 
         $that = $this;
         $pending = &$this->pending;
-        return $deferred->promise()->then(function () use ($request, $next, &$pending) {
+        return $deferred->promise()->then(function () use ($request, $next, $body, &$pending) {
             $pending++;
 
             $ret = $next($request);
 
             // resume readable stream and replay buffered events
-            $body = $request->getBody();
-            if ($body instanceof ReadableStreamInterface) {
-                $body->resume();
+            if ($body instanceof PauseBufferStream) {
+                $body->resumeImplicit();
             }
 
             return $ret;

tests/Middleware/LimitHandlersMiddlewareTest.php

@@ -255,7 +255,6 @@ public function testReceivesNextStreamingBodyWithBufferedDataAfterPreviousHandle
             new HttpBodyStream($stream, 10)
         );
 
-        $req = null;
         $once = $this->expectCallableOnceWith('helloworld');
         $middleware($request, function (ServerRequestInterface $request) use ($once) {
             $request->getBody()->on('data', $once);
@@ -266,4 +265,90 @@ public function testReceivesNextStreamingBodyWithBufferedDataAfterPreviousHandle
 
         $deferred->reject(new \RuntimeException());
     }
+
+    public function testReceivesNextStreamingBodyAndDoesNotEmitDataIfExplicitlyClosed()
+    {
+        $deferred = new Deferred();
+        $middleware = new LimitHandlersMiddleware(1);
+        $middleware(new ServerRequest('GET', 'http://example.com/'), function () use ($deferred) {
+            return $deferred->promise();
+        });
+
+        $stream = new ThroughStream();
+        $request = new ServerRequest(
+            'POST',
+            'http://example.com/',
+            array(),
+            new HttpBodyStream($stream, 10)
+        );
+
+        $never = $this->expectCallableNever();
+        $middleware($request, function (ServerRequestInterface $request) use ($never) {
+            $request->getBody()->close();
+            $request->getBody()->on('data', $never);
+        });
+
+        $stream->write('hello');
+        $stream->write('world');
+
+        $deferred->reject(new \RuntimeException());
+    }
+
+    public function testReceivesNextStreamingBodyAndDoesNotEmitDataIfExplicitlyPaused()
+    {
+        $deferred = new Deferred();
+        $middleware = new LimitHandlersMiddleware(1);
+        $middleware(new ServerRequest('GET', 'http://example.com/'), function () use ($deferred) {
+            return $deferred->promise();
+        });
+
+        $stream = new ThroughStream();
+        $request = new ServerRequest(
+            'POST',
+            'http://example.com/',
+            array(),
+            new HttpBodyStream($stream, 10)
+        );
+
+        $never = $this->expectCallableNever();
+        $middleware($request, function (ServerRequestInterface $request) use ($never) {
+            $request->getBody()->pause();
+            $request->getBody()->on('data', $never);
+        });
+
+        $stream->write('hello');
+        $stream->write('world');
+
+        $deferred->reject(new \RuntimeException());
+    }
+
+    public function testReceivesNextStreamingBodyAndDoesEmitDataImmediatelyIfExplicitlyResumed()
+    {
+        $deferred = new Deferred();
+        $middleware = new LimitHandlersMiddleware(1);
+        $middleware(new ServerRequest('GET', 'http://example.com/'), function () use ($deferred) {
+            return $deferred->promise();
+        });
+
+        $stream = new ThroughStream();
+        $request = new ServerRequest(
+            'POST',
+            'http://example.com/',
+            array(),
+            new HttpBodyStream($stream, 10)
+        );
+
+        $once = $this->expectCallableOnceWith('helloworld');
+        $never = $this->expectCallableNever();
+        $middleware($request, function (ServerRequestInterface $request) use ($once, $never) {
+            $request->getBody()->on('data', $once);
+            $request->getBody()->resume();
+            $request->getBody()->on('data', $never);
+        });
+
+        $stream->write('hello');
+        $stream->write('world');
+
+        $deferred->reject(new \RuntimeException());
+    }
 }