[1.x] Ensure connection close handler is cleaned up for each request

src/Io/StreamingServer.php

@@ -157,10 +157,17 @@ public function handleRequest(ConnectionInterface $conn, ServerRequestInterface
         }
 
         // cancel pending promise once connection closes
+        $connectionOnCloseResponseCancelerHandler = function () {};
         if ($response instanceof PromiseInterface && \method_exists($response, 'cancel')) {
-            $conn->on('close', function () use ($response) {
+            $connectionOnCloseResponseCanceler = function () use ($response) {
                 $response->cancel();
-            });
+            };
+            $connectionOnCloseResponseCancelerHandler = function () use ($connectionOnCloseResponseCanceler, $conn) {
+                if ($connectionOnCloseResponseCanceler !== null) {
+                    $conn->removeListener('close', $connectionOnCloseResponseCanceler);
+                }
+            };
+            $conn->on('close', $connectionOnCloseResponseCanceler);
         }
 
         // happy path: response returned, handle and return immediately
@@ -201,7 +208,7 @@ function ($error) use ($that, $conn, $request) {
                 $that->emit('error', array($exception));
                 return $that->writeError($conn, Response::STATUS_INTERNAL_SERVER_ERROR, $request);
             }
-        );
+        )->then($connectionOnCloseResponseCancelerHandler, $connectionOnCloseResponseCancelerHandler);
     }
 
     /** @internal */

tests/Io/StreamingServerTest.php

@@ -25,9 +25,17 @@ class StreamingServerTest extends TestCase
      */
     public function setUpConnectionMockAndSocket()
     {
-        $this->connection = $this->getMockBuilder('React\Socket\Connection')
+        $this->connection = $this->mockConnection();
+
+        $this->socket = new SocketServerStub();
+    }
+
+
+    private function mockConnection(array $additionalMethods = null)
+    {
+        $connection = $this->getMockBuilder('React\Socket\Connection')
             ->disableOriginalConstructor()
-            ->setMethods(
+            ->setMethods(array_merge(
                 array(
                     'write',
                     'end',
@@ -39,14 +47,15 @@ public function setUpConnectionMockAndSocket()
                     'getRemoteAddress',
                     'getLocalAddress',
                     'pipe'
-                )
-            )
+                ),
+                (is_array($additionalMethods) ? $additionalMethods : array())
+            ))
             ->getMock();
 
-        $this->connection->method('isWritable')->willReturn(true);
-        $this->connection->method('isReadable')->willReturn(true);
+        $connection->method('isWritable')->willReturn(true);
+        $connection->method('isReadable')->willReturn(true);
 
-        $this->socket = new SocketServerStub();
+        return $connection;
     }
 
     public function testRequestEventWillNotBeEmittedForIncompleteHeaders()
@@ -3245,6 +3254,25 @@ public function testNewConnectionWillInvokeParserTwiceAfterInvokingRequestHandle
         $this->assertCount(1, $this->connection->listeners('close'));
     }
 
+    public function testCompletingARequestWillRemoveConnectionOnCloseListener()
+    {
+        $connection = $this->mockConnection(array('removeListener'));
+
+        $request = new ServerRequest('GET', 'http://localhost/');
+
+        $server = new StreamingServer(Loop::get(), function () {
+            return \React\Promise\resolve(new Response());
+        });
+
+        $server->listen($this->socket);
+        $this->socket->emit('connection', array($connection));
+
+        $connection->expects($this->once())->method('removeListener');
+
+        // pretend parser just finished parsing
+        $server->handleRequest($connection, $request);
+    }
+
     private function createGetRequest()
     {
         $data = "GET / HTTP/1.1\r\n";