Consistent write() semantics and simplified buffering while connecting

Accordingly, the `drain` event now behaves consistently and we no longer
need the custom `headers-written` event.

Author: clue

README.md

@@ -10,17 +10,18 @@ Requests are prepared using the ``Client#request()`` method. Body can be
 sent with ``Request#write()``. ``Request#end()`` finishes sending the request
 (or sends it at all if no body was written).
 
-Request implements WritableStreamInterface, so a Stream can be piped to
-it. Response implements ReadableStreamInterface.
-
+Request implements WritableStreamInterface, so a Stream can be piped to it.
 Interesting events emitted by Request:
 
 * `response`: The response headers were received from the server and successfully
   parsed. The first argument is a Response instance.
+* `drain`: The outgoing buffer drained and the response is ready to accept more
+  data for the next `write()` call.
 * `error`: An error occurred.
 * `end`: The request is finished. If an error occurred, it is passed as first
   argument. Second and third arguments are the Response and the Request.
 
+Response implements ReadableStreamInterface.
 Interesting events emitted by Response:
 
 * `data`: Passes a chunk of the response body as first argument and a Response

src/Request.php

@@ -6,9 +6,9 @@
 use GuzzleHttp\Psr7 as gPsr;
 use React\Socket\ConnectorInterface;
 use React\Stream\WritableStreamInterface;
+use React\Socket\ConnectionInterface;
 
 /**
- * @event headers-written
  * @event response
  * @event drain
  * @event error
@@ -32,7 +32,7 @@ class Request implements WritableStreamInterface
     private $response;
     private $state = self::STATE_INIT;
 
-    private $pendingWrites = array();
+    private $pendingWrites = '';
 
     public function __construct(ConnectorInterface $connector, RequestData $requestData)
     {
@@ -45,22 +45,19 @@ public function isWritable()
         return self::STATE_END > $this->state;
     }
 
-    public function writeHead()
+    private function writeHead()
     {
-        if (self::STATE_WRITING_HEAD <= $this->state) {
-            throw new \LogicException('Headers already written');
-        }
-
         $this->state = self::STATE_WRITING_HEAD;
 
         $requestData = $this->requestData;
         $streamRef = &$this->stream;
         $stateRef = &$this->state;
+        $pendingWrites = &$this->pendingWrites;
 
         $this
             ->connect()
             ->done(
-                function ($stream) use ($requestData, &$streamRef, &$stateRef) {
+                function (ConnectionInterface $stream) use ($requestData, &$streamRef, &$stateRef, &$pendingWrites) {
                     $streamRef = $stream;
 
                     $stream->on('drain', array($this, 'handleDrain'));
@@ -71,11 +68,18 @@ function ($stream) use ($requestData, &$streamRef, &$stateRef) {
 
                     $headers = (string) $requestData;
 
-                    $stream->write($headers);
+                    $more = $stream->write($headers . $pendingWrites);
 
                     $stateRef = Request::STATE_HEAD_WRITTEN;
 
-                    $this->emit('headers-written', array($this));
+                    // clear pending writes if non-empty
+                    if ($pendingWrites !== '') {
+                        $pendingWrites = '';
+
+                        if ($more) {
+                            $this->emit('drain');
+                        }
+                    }
                 },
                 array($this, 'handleError')
             );
@@ -84,25 +88,16 @@ function ($stream) use ($requestData, &$streamRef, &$stateRef) {
     public function write($data)
     {
         if (!$this->isWritable()) {
-            return;
+            return false;
         }
 
+        // write directly to connection stream if already available
         if (self::STATE_HEAD_WRITTEN <= $this->state) {
             return $this->stream->write($data);
         }
 
-        if (!count($this->pendingWrites)) {
-            $this->on('headers-written', function ($that) {
-                foreach ($that->pendingWrites as $pw) {
-                    $that->write($pw);
-                }
-                $that->pendingWrites = array();
-                $that->emit('drain', array($that));
-            });
-        }
-
-        $this->pendingWrites[] = $data;
-
+        // otherwise buffer and try to establish connection
+        $this->pendingWrites .= $data;
         if (self::STATE_WRITING_HEAD > $this->state) {
             $this->writeHead();
         }
@@ -123,9 +118,10 @@ public function end($data = null)
         }
     }
 
+    /** @internal */
     public function handleDrain()
     {
-        $this->emit('drain', array($this));
+        $this->emit('drain');
     }
 
     public function handleData($data)
@@ -207,6 +203,7 @@ public function close()
         }
 
         $this->state = self::STATE_END;
+        $this->pendingWrites = '';
 
         if ($this->stream) {
             $this->stream->close();

tests/RequestTest.php

@@ -221,7 +221,7 @@ public function requestShouldEmitErrorIfGuzzleParseThrowsException()
 
         $request->on('error', $handler);
 
-        $request->writeHead();
+        $request->end();
         $request->handleData("\r\n\r\n");
     }
 
@@ -253,13 +253,9 @@ public function postRequestShouldSendAPostRequest()
         $this->successfulConnectionMock();
 
         $this->stream
-            ->expects($this->at(5))
-            ->method('write')
-            ->with($this->matchesRegularExpression("#^POST / HTTP/1\.0\r\nHost: www.example.com\r\nUser-Agent:.*\r\n\r\n$#"));
-        $this->stream
-            ->expects($this->at(6))
+            ->expects($this->once())
             ->method('write')
-            ->with($this->identicalTo("some post data"));
+            ->with($this->matchesRegularExpression("#^POST / HTTP/1\.0\r\nHost: www.example.com\r\nUser-Agent:.*\r\n\r\nsome post data$#"));
 
         $factory = $this->createCallableMock();
         $factory->expects($this->once())
@@ -285,17 +281,13 @@ public function writeWithAPostRequestShouldSendToTheStream()
         $this->stream
             ->expects($this->at(5))
             ->method('write')
-            ->with($this->matchesRegularExpression("#^POST / HTTP/1\.0\r\nHost: www.example.com\r\nUser-Agent:.*\r\n\r\n$#"));
+            ->with($this->matchesRegularExpression("#^POST / HTTP/1\.0\r\nHost: www.example.com\r\nUser-Agent:.*\r\n\r\nsome$#"));
         $this->stream
             ->expects($this->at(6))
             ->method('write')
-            ->with($this->identicalTo("some"));
-        $this->stream
-            ->expects($this->at(7))
-            ->method('write')
             ->with($this->identicalTo("post"));
         $this->stream
-            ->expects($this->at(8))
+            ->expects($this->at(7))
             ->method('write')
             ->with($this->identicalTo("data"));
 
@@ -326,17 +318,56 @@ public function writeWithAPostRequestShouldSendBodyAfterHeadersAndEmitDrainEvent
         $this->stream
             ->expects($this->at(5))
             ->method('write')
-            ->with($this->matchesRegularExpression("#^POST / HTTP/1\.0\r\nHost: www.example.com\r\nUser-Agent:.*\r\n\r\n$#"));
+            ->with($this->matchesRegularExpression("#^POST / HTTP/1\.0\r\nHost: www.example.com\r\nUser-Agent:.*\r\n\r\nsomepost$#"))
+            ->willReturn(true);
         $this->stream
             ->expects($this->at(6))
             ->method('write')
-            ->with($this->identicalTo("some"));
+            ->with($this->identicalTo("data"));
+
+        $factory = $this->createCallableMock();
+        $factory->expects($this->once())
+            ->method('__invoke')
+            ->will($this->returnValue($this->response));
+
+        $request->setResponseFactory($factory);
+
+        $this->assertFalse($request->write("some"));
+        $this->assertFalse($request->write("post"));
+
+        $request->on('drain', $this->expectCallableOnce());
+        $request->once('drain', function () use ($request) {
+            $request->write("data");
+            $request->end();
+        });
+
+        $resolveConnection();
+
+        $request->handleData("HTTP/1.0 200 OK\r\n");
+        $request->handleData("Content-Type: text/plain\r\n");
+        $request->handleData("\r\nbody");
+    }
+
+    /** @test */
+    public function writeWithAPostRequestShouldForwardDrainEventIfFirstChunkExceedsBuffer()
+    {
+        $requestData = new RequestData('POST', 'http://www.example.com');
+        $request = new Request($this->connector, $requestData);
+
+        $this->stream = $this->getMockBuilder('React\Socket\Connection')
+            ->disableOriginalConstructor()
+            ->setMethods(array('write'))
+            ->getMock();
+
+        $resolveConnection = $this->successfulAsyncConnectionMock();
+
         $this->stream
-            ->expects($this->at(7))
+            ->expects($this->at(0))
             ->method('write')
-            ->with($this->identicalTo("post"));
+            ->with($this->matchesRegularExpression("#^POST / HTTP/1\.0\r\nHost: www.example.com\r\nUser-Agent:.*\r\n\r\nsomepost$#"))
+            ->willReturn(false);
         $this->stream
-            ->expects($this->at(8))
+            ->expects($this->at(1))
             ->method('write')
             ->with($this->identicalTo("data"));
 
@@ -350,12 +381,14 @@ public function writeWithAPostRequestShouldSendBodyAfterHeadersAndEmitDrainEvent
         $this->assertFalse($request->write("some"));
         $this->assertFalse($request->write("post"));
 
+        $request->on('drain', $this->expectCallableOnce());
         $request->once('drain', function () use ($request) {
             $request->write("data");
             $request->end();
         });
 
         $resolveConnection();
+        $this->stream->emit('drain');
 
         $request->handleData("HTTP/1.0 200 OK\r\n");
         $request->handleData("Content-Type: text/plain\r\n");
@@ -373,17 +406,13 @@ public function pipeShouldPipeDataIntoTheRequestBody()
         $this->stream
             ->expects($this->at(5))
             ->method('write')
-            ->with($this->matchesRegularExpression("#^POST / HTTP/1\.0\r\nHost: www.example.com\r\nUser-Agent:.*\r\n\r\n$#"));
+            ->with($this->matchesRegularExpression("#^POST / HTTP/1\.0\r\nHost: www.example.com\r\nUser-Agent:.*\r\n\r\nsome$#"));
         $this->stream
             ->expects($this->at(6))
             ->method('write')
-            ->with($this->identicalTo("some"));
-        $this->stream
-            ->expects($this->at(7))
-            ->method('write')
             ->with($this->identicalTo("post"));
         $this->stream
-            ->expects($this->at(8))
+            ->expects($this->at(7))
             ->method('write')
             ->with($this->identicalTo("data"));
 
@@ -424,6 +453,32 @@ public function endShouldOnlyAcceptScalars()
         $request->end(array());
     }
 
+    /**
+     * @test
+     */
+    public function closeShouldEmitCloseEvent()
+    {
+        $requestData = new RequestData('POST', 'http://www.example.com');
+        $request = new Request($this->connector, $requestData);
+
+        $request->on('close', $this->expectCallableOnce());
+        $request->close();
+    }
+
+    /**
+     * @test
+     */
+    public function writeAfterCloseReturnsFalse()
+    {
+        $requestData = new RequestData('POST', 'http://www.example.com');
+        $request = new Request($this->connector, $requestData);
+
+        $request->close();
+
+        $this->assertFalse($request->isWritable());
+        $this->assertFalse($request->write('nope'));
+    }
+
     /** @test */
     public function requestShouldRelayErrorEventsFromResponse()
     {