Feature: chunked encoding

README.md

@@ -55,6 +55,5 @@ $loop->run();
 ## TODO
 
 * gzip content encoding
-* chunked transfer encoding
 * keep-alive connections
 * following redirections

src/ChunkedStreamDecoder.php

@@ -0,0 +1,105 @@
+<?php
+
+namespace React\HttpClient;
+
+use Evenement\EventEmitterTrait;
+use React\Stream\DuplexStreamInterface;
+use React\Stream\Util;
+
+class ChunkedStreamDecoder
+{
+    const CRLF = "\r\n";
+
+    use EventEmitterTrait;
+
+    /**
+     * @var string
+     */
+    protected $buffer = '';
+
+    /**
+     * @var int
+     */
+    protected $remainingLength = 0;
+
+    /**
+     * @var bool
+     */
+    protected $nextChunkIsLength = true;
+
+    /**
+     * @var DuplexStreamInterface
+     */
+    protected $stream;
+
+    /**
+     * @param DuplexStreamInterface $stream
+     */
+    public function __construct(DuplexStreamInterface $stream)
+    {
+        $this->stream = $stream;
+        $this->stream->on('data', array($this, 'handleData'));
+        Util::forwardEvents($this->stream, $this, [
+            'error',
+            'end',
+        ]);
+    }
+
+    public function handleData($data)
+    {
+        $this->buffer .= $data;
+
+        do {
+            $this->iterateBuffer();
+        } while (strlen($this->buffer) > 0 && strpos($this->buffer, static::CRLF) !== false);
+    }
+
+    protected function iterateBuffer()
+    {
+        if ($this->nextChunkIsLength) {
+            $this->nextChunkIsLength = false;
+            $this->remainingLength = hexdec(substr($this->buffer, 0, strpos($this->buffer, static::CRLF)));
+            $this->buffer = substr($this->buffer, strpos($this->buffer, static::CRLF) + 2);
+        }
+
+        if ($this->remainingLength > 0) {
+            $chunkLength = $this->getChunkLength();
+            $this->emit('data', array(
+                substr($this->buffer, 0, $chunkLength),
+                $this
+            ));
+            $this->remainingLength -= $chunkLength;
+            $this->buffer = substr($this->buffer, $chunkLength);
+            return;
+        }
+
+        $this->nextChunkIsLength = true;
+        $this->buffer = substr($this->buffer, 2);
+    }
+
+    protected function getChunkLength()
+    {
+        $bufferLength = strlen($this->buffer);
+
+        if ($bufferLength >= $this->remainingLength) {
+            return $this->remainingLength;
+        }
+
+        return $bufferLength;
+    }
+
+    public function end($data = null)
+    {
+        $this->stream->end($data);
+    }
+
+    public function pause()
+    {
+        $this->stream->pause();
+    }
+
+    public function resume()
+    {
+        $this->stream->resume();
+    }
+}

src/Response.php

@@ -34,9 +34,13 @@ public function __construct(DuplexStreamInterface $stream, $protocol, $version,
         $this->reasonPhrase = $reasonPhrase;
         $this->headers = $headers;
 
-        $stream->on('data', array($this, 'handleData'));
-        $stream->on('error', array($this, 'handleError'));
-        $stream->on('end', array($this, 'handleEnd'));
+        if (isset($this->headers['transfer-encoding']) && $this->headers['transfer-encoding'] == 'chunked') {
+            $this->stream = new ChunkedStreamDecoder($stream);
+        }
+
+        $this->stream->on('data', array($this, 'handleData'));
+        $this->stream->on('error', array($this, 'handleError'));
+        $this->stream->on('end', array($this, 'handleEnd'));
     }
 
     public function getProtocol()

tests/DecodeChunkedStreamTest.php

@@ -0,0 +1,40 @@
+<?php
+
+namespace React\Tests\HttpClient;
+
+use React\HttpClient\ChunkedStreamDecoder;
+use React\Stream\ThroughStream;
+
+class DecodeChunkedStreamTest extends TestCase
+{
+    public function provideChunkedEncoding()
+    {
+        return [
+            [["4\r\nWiki\r\n5\r\npedia\r\ne\r\n in\r\n\r\nchunks.\r\n0\r\n\r\n"]],
+            [["4\r\nWiki\r\n", "5\r\npedia\r\ne\r\n in\r\n\r\nchunks.\r\n0\r\n\r\n"]],
+            [["4\r\nWiki\r\n", "5\r\n", "pedia\r\ne\r\n in\r\n\r\nchunks.\r\n0\r\n\r\n"]],
+            [["4\r\nWiki\r\n", "5\r\n", "pedia\r\ne\r\n in\r\n", "\r\nchunks.\r\n0\r\n\r\n"]],
+            [["4\r\n", "Wiki\r\n", "5\r\n", "pedia\r\ne\r\n in\r\n", "\r\nchunks.\r\n0\r\n\r\n"]],
+            [["4\r\n", "Wiki\r\n", "5\r\n", "pedia\r\ne\r\n", " in\r\n", "\r\nchunks.\r\n0\r\n\r\n"]],
+            [["4\r\n", "Wiki\r\n", "5\r\n", "pedia\r\ne\r\n", " in\r\n", "\r\nchunks.\r\n", "0\r\n\r\n"]],
+        ];
+    }
+
+    /**
+     * @test
+     * @dataProvider provideChunkedEncoding
+     */
+    public function testChunkedEncoding(array $strings)
+    {
+        $stream = new ThroughStream();
+        $response = new ChunkedStreamDecoder($stream);
+        $buffer = '';
+        $response->on('data', function ($data) use (&$buffer) {
+            $buffer .= $data;
+        });
+        foreach ($strings as $string) {
+            $stream->write($string);
+        }
+        $this->assertSame("Wikipedia in\r\n\r\nchunks.", $buffer);
+    }
+}

tests/ResponseTest.php

@@ -3,6 +3,7 @@
 namespace React\Tests\HttpClient;
 
 use React\HttpClient\Response;
+use React\Stream\ThroughStream;
 
 class ResponseTest extends TestCase
 {
@@ -73,5 +74,32 @@ public function closedResponseShouldNotBeResumedOrPaused()
         $response->resume();
         $response->pause();
     }
+
+    /** @test */
+    public function chunkedEncodingResponse()
+    {
+        $stream = new ThroughStream();
+        $response = new Response(
+            $stream,
+            'http',
+            '1.0',
+            '200',
+            'ok',
+            [
+                'content-type' => 'text/plain',
+                'transfer-encoding' => 'chunked',
+            ]
+        );
+
+        $buffer = '';
+        $response->on('data', function ($data) use (&$buffer) {
+            $buffer.= $data;
+        });
+        $this->assertSame('', $buffer);
+        $stream->write("4\r\n");
+        $this->assertSame('', $buffer);
+        $stream->write("Wiki\r\n");
+        $this->assertSame('Wiki', $buffer);
+    }
 }