Reject request when streaming request body emits error or closes unexpectedly

README.md

@@ -499,6 +499,10 @@ $loop->addTimer(1.0, function () use ($body) {
 $browser->post($url, array('Content-Length' => '11'), $body);
 ```
 
+If the streaming request body emits an `error` event or is explicitly closed
+without emitting a successful `end` event first, the request will automatically
+be closed and rejected.
+
 #### submit()
 
 The `submit($url, array $fields, $headers = array(), $method = 'POST'): PromiseInterface<ResponseInterface>` method can be used to

src/Io/Sender.php

@@ -133,6 +133,19 @@ public function send(RequestInterface $request)
                 // add dummy write to immediately start request even if body does not emit any data yet
                 $body->pipe($requestStream);
                 $requestStream->write('');
+
+                $body->on('close', $close = function () use ($deferred, $requestStream) {
+                    $deferred->reject(new \RuntimeException('Request failed because request body closed unexpectedly'));
+                    $requestStream->close();
+                });
+                $body->on('error', function ($e) use ($deferred, $requestStream, $close, $body) {
+                    $body->removeListener('close', $close);
+                    $deferred->reject(new \RuntimeException('Request failed because request body reported an error', 0, $e));
+                    $requestStream->close();
+                });
+                $body->on('end', function () use ($close, $body) {
+                    $body->removeListener('close', $close);
+                });
             } else {
                 // stream is not readable => end request without body
                 $requestStream->end();

tests/FunctionalBrowserTest.php

@@ -138,7 +138,7 @@ public function testTimeoutDelayedResponseAfterStreamingRequestShouldReject()
     {
         $stream = new ThroughStream();
         $promise = $this->browser->withOptions(array('timeout' => 0.1))->post($this->base . 'delay/10', array(), $stream);
-        $stream->close();
+        $stream->end();
 
         Block\await($promise, $this->loop);
     }

tests/Io/SenderTest.php

@@ -152,6 +152,92 @@ public function testSendPostStreamWillAutomaticallyPipeChunkEncodeBodyForEnd()
         $stream->end();
     }
 
+    public function testSendPostStreamWillRejectWhenRequestBodyEmitsErrorEvent()
+    {
+        $outgoing = $this->getMockBuilder('React\HttpClient\Request')->disableOriginalConstructor()->getMock();
+        $outgoing->expects($this->once())->method('isWritable')->willReturn(true);
+        $outgoing->expects($this->once())->method('write')->with("")->willReturn(false);
+        $outgoing->expects($this->never())->method('end');
+        $outgoing->expects($this->once())->method('close');
+
+        $client = $this->getMockBuilder('React\HttpClient\Client')->disableOriginalConstructor()->getMock();
+        $client->expects($this->once())->method('request')->willReturn($outgoing);
+
+        $sender = new Sender($client, $this->getMockBuilder('Clue\React\Buzz\Message\MessageFactory')->getMock());
+
+        $expected = new \RuntimeException();
+        $stream = new ThroughStream();
+        $request = new Request('POST', 'http://www.google.com/', array(), new ReadableBodyStream($stream));
+        $promise = $sender->send($request);
+
+        $stream->emit('error', array($expected));
+
+        $exception = null;
+        $promise->then(null, function ($e) use (&$exception) {
+            $exception = $e;
+        });
+
+        $this->assertInstanceOf('RuntimeException', $exception);
+        $this->assertEquals('Request failed because request body reported an error', $exception->getMessage());
+        $this->assertSame($expected, $exception->getPrevious());
+    }
+
+    public function testSendPostStreamWillRejectWhenRequestBodyClosesWithoutEnd()
+    {
+        $outgoing = $this->getMockBuilder('React\HttpClient\Request')->disableOriginalConstructor()->getMock();
+        $outgoing->expects($this->once())->method('isWritable')->willReturn(true);
+        $outgoing->expects($this->once())->method('write')->with("")->willReturn(false);
+        $outgoing->expects($this->never())->method('end');
+        $outgoing->expects($this->once())->method('close');
+
+        $client = $this->getMockBuilder('React\HttpClient\Client')->disableOriginalConstructor()->getMock();
+        $client->expects($this->once())->method('request')->willReturn($outgoing);
+
+        $sender = new Sender($client, $this->getMockBuilder('Clue\React\Buzz\Message\MessageFactory')->getMock());
+
+        $stream = new ThroughStream();
+        $request = new Request('POST', 'http://www.google.com/', array(), new ReadableBodyStream($stream));
+        $promise = $sender->send($request);
+
+        $stream->close();
+
+        $exception = null;
+        $promise->then(null, function ($e) use (&$exception) {
+            $exception = $e;
+        });
+
+        $this->assertInstanceOf('RuntimeException', $exception);
+        $this->assertEquals('Request failed because request body closed unexpectedly', $exception->getMessage());
+    }
+
+    public function testSendPostStreamWillNotRejectWhenRequestBodyClosesAfterEnd()
+    {
+        $outgoing = $this->getMockBuilder('React\HttpClient\Request')->disableOriginalConstructor()->getMock();
+        $outgoing->expects($this->once())->method('isWritable')->willReturn(true);
+        $outgoing->expects($this->exactly(2))->method('write')->withConsecutive(array(""), array("0\r\n\r\n"))->willReturn(false);
+        $outgoing->expects($this->once())->method('end');
+        $outgoing->expects($this->never())->method('close');
+
+        $client = $this->getMockBuilder('React\HttpClient\Client')->disableOriginalConstructor()->getMock();
+        $client->expects($this->once())->method('request')->willReturn($outgoing);
+
+        $sender = new Sender($client, $this->getMockBuilder('Clue\React\Buzz\Message\MessageFactory')->getMock());
+
+        $stream = new ThroughStream();
+        $request = new Request('POST', 'http://www.google.com/', array(), new ReadableBodyStream($stream));
+        $promise = $sender->send($request);
+
+        $stream->end();
+        $stream->close();
+
+        $exception = null;
+        $promise->then(null, function ($e) use (&$exception) {
+            $exception = $e;
+        });
+
+        $this->assertNull($exception);
+    }
+
     public function testSendPostStreamWithExplicitContentLengthWillSendHeaderAsIs()
     {
         $client = $this->getMockBuilder('React\HttpClient\Client')->disableOriginalConstructor()->getMock();