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,186 @@
+<?php
+
+namespace React\HttpClient;
+
+use Evenement\EventEmitterTrait;
+use Exception;
+use React\Stream\ReadableStreamInterface;
+use React\Stream\Util;
+use React\Stream\WritableStreamInterface;
+
+class ChunkedStreamDecoder implements ReadableStreamInterface
+{
+    const CRLF = "\r\n";
+
+    use EventEmitterTrait;
+
+    /**
+     * @var string
+     */
+    protected $buffer = '';
+
+    /**
+     * @var int
+     */
+    protected $remainingLength = 0;
+
+    /**
+     * @var bool
+     */
+    protected $nextChunkIsLength = true;
+
+    /**
+     * @var ReadableStreamInterface
+     */
+    protected $stream;
+
+    /**
+     * @var bool
+     */
+    protected $closed = false;
+
+    /**
+     * @param ReadableStreamInterface $stream
+     */
+    public function __construct(ReadableStreamInterface $stream)
+    {
+        $this->stream = $stream;
+        $this->stream->on('data', array($this, 'handleData'));
+        $this->stream->on('end',  array($this, 'handleEnd'));
+        Util::forwardEvents($this->stream, $this, [
+            'error',
+        ]);
+    }
+
+    /** @internal */
+    public function handleData($data)
+    {
+        $this->buffer .= $data;
+
+        do {
+            $bufferLength = strlen($this->buffer);
+            $continue = $this->iterateBuffer();
+            $iteratedBufferLength = strlen($this->buffer);
+        } while (
+            $continue &&
+            $bufferLength !== $iteratedBufferLength &&
+            $iteratedBufferLength > 0 &&
+            strpos($this->buffer, static::CRLF) !== false
+        );
+    }
+
+    protected function iterateBuffer()
+    {
+        if (strlen($this->buffer) <= 1) {
+            return false;
+        }
+
+        if ($this->nextChunkIsLength) {
+            $crlfPosition = strpos($this->buffer, static::CRLF);
+            if ($crlfPosition === false && strlen($this->buffer) > 1024) {
+                $this->emit('error', [
+                    new Exception('Chunk length header longer then 1024 bytes'),
+                ]);
+                $this->close();
+                return false;
+            }
+            if ($crlfPosition === false) {
+                return false; // Chunk header hasn't completely come in yet
+            }
+            $this->nextChunkIsLength = false;
+            $lengthChunk = substr($this->buffer, 0, $crlfPosition);
+            if (strpos($lengthChunk, ';') !== false) {
+                list($lengthChunk) = explode(';', $lengthChunk, 2);
+            }
+            if (dechex(hexdec($lengthChunk)) !== $lengthChunk) {
+                $this->emit('error', [
+                    new Exception('Unable to validate "' . $lengthChunk . '" as chunk length header'),
+                ]);
+                $this->close();
+                return false;
+            }
+            $this->remainingLength = hexdec($lengthChunk);
+            $this->buffer = substr($this->buffer, $crlfPosition + 2);
+            return true;
+        }
+
+        if ($this->remainingLength > 0) {
+            $chunkLength = $this->getChunkLength();
+            if ($chunkLength === 0) {
+                return true;
+            }
+            $this->emit('data', array(
+                substr($this->buffer, 0, $chunkLength),
+                $this
+            ));
+            $this->remainingLength -= $chunkLength;
+            $this->buffer = substr($this->buffer, $chunkLength);
+            return true;
+        }
+
+        $this->nextChunkIsLength = true;
+        $this->buffer = substr($this->buffer, 2);
+        return true;
+    }
+
+    protected function getChunkLength()
+    {
+        $bufferLength = strlen($this->buffer);
+
+        if ($bufferLength >= $this->remainingLength) {
+            return $this->remainingLength;
+        }
+
+        return $bufferLength;
+    }
+
+    public function pause()
+    {
+        $this->stream->pause();
+    }
+
+    public function resume()
+    {
+        $this->stream->resume();
+    }
+
+    public function isReadable()
+    {
+        return $this->stream->isReadable();
+    }
+
+    public function pipe(WritableStreamInterface $dest, array $options = array())
+    {
+        Util::pipe($this, $dest, $options);
+
+        return $dest;
+    }
+
+    public function close()
+    {
+        $this->closed = true;
+        return $this->stream->close();
+    }
+
+    /** @internal */
+    public function handleEnd()
+    {
+        if ($this->closed) {
+            return;
+        }
+
+        if ($this->buffer === '') {
+            $this->emit('end');
+            $this->close();
+            return;
+        }
+
+        $this->emit(
+            'error',
+            [
+                new Exception('Stream ended with incomplete control code')
+            ]
+        );
+        $this->close();
+    }
+}

src/Response.php

@@ -33,10 +33,22 @@ public function __construct(DuplexStreamInterface $stream, $protocol, $version,
         $this->code = $code;
         $this->reasonPhrase = $reasonPhrase;
         $this->headers = $headers;
+        $normalizedHeaders = array_change_key_case($headers, CASE_LOWER);
 
-        $stream->on('data', array($this, 'handleData'));
-        $stream->on('error', array($this, 'handleError'));
-        $stream->on('end', array($this, 'handleEnd'));
+        if (isset($normalizedHeaders['transfer-encoding']) && strtolower($normalizedHeaders['transfer-encoding']) === 'chunked') {
+            $this->stream = new ChunkedStreamDecoder($stream);
+
+            foreach ($this->headers as $key => $value) {
+                if (strcasecmp('transfer-encoding', $key) === 0) {
+                    unset($this->headers[$key]);
+                    break;
+                }
+            }
+        }
+
+        $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,129 @@
+<?php
+
+namespace React\Tests\HttpClient;
+
+use Exception;
+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; foo=[bar,beer,pool,cue,win,won]\r\n", " in\r\n", "\r\nchunks.\r\n0\r\n\r\n"],
+            ],
+            [
+                ["4; foo=bar\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"],
+            ],
+            [
+                str_split("4\r\nWiki\r\n5\r\npedia\r\ne\r\n in\r\n\r\nchunks.\r\n0\r\n\r\n"),
+            ],
+            [
+                str_split("6\r\nWi\r\nki\r\n5\r\npedia\r\ne\r\n in\r\n\r\nchunks.\r\n0\r\n\r\n"),
+                "Wi\r\nkipedia in\r\n\r\nchunks."
+            ],
+            [
+                ["6\r\nWi\r\n", "ki\r\n5\r\npedia\r\ne\r\n in\r\n\r\nchunks.\r\n0\r\n\r\n"],
+                "Wi\r\nkipedia in\r\n\r\nchunks."
+            ],
+        ];
+    }
+
+    /**
+     * @test
+     * @dataProvider provideChunkedEncoding
+     */
+    public function testChunkedEncoding(array $strings, $expected = "Wikipedia in\r\n\r\nchunks.")
+    {
+        $stream = new ThroughStream();
+        $response = new ChunkedStreamDecoder($stream);
+        $buffer = '';
+        $response->on('data', function ($data) use (&$buffer) {
+            $buffer .= $data;
+        });
+        $response->on('error', function (Exception $exception) {
+            throw $exception;
+        });
+        foreach ($strings as $string) {
+            $stream->write($string);
+        }
+        $this->assertSame($expected, $buffer);
+    }
+
+    public function provideInvalidChunkedEncoding()
+    {
+        return [
+            [
+                ["4\r\nWiwot40n98w3498tw3049nyn039409t34\r\n", "ki\r\n5\r\npedia\r\ne\r\n in\r\n\r\nchunks.\r\n0\r\n\r\n"],
+            ],
+            [
+                str_split("xyz\r\nWi\r\nki\r\n5\r\npedia\r\ne\r\n in\r\n\r\nchunks.\r\n0\r\n\r\n")
+            ],
+            [
+                str_split(str_repeat('a', 2015) . "\r\nWi\r\nki\r\n5\r\npedia\r\ne\r\n in\r\n\r\nchunks.\r\n0\r\n\r\n")
+            ],
+        ];
+    }
+
+    /**
+     * @test
+     * @dataProvider provideInvalidChunkedEncoding
+     * @expectedException Exception
+     */
+    public function testInvalidChunkedEncoding(array $strings)
+    {
+        $stream = new ThroughStream();
+        $response = new ChunkedStreamDecoder($stream);
+        $response->on('error', function (Exception $exception) {
+            throw $exception;
+        });
+        foreach ($strings as $string) {
+            $stream->write($string);
+        }
+    }
+
+    public function testHandleEnd()
+    {
+        $ended = false;
+        $stream = new ThroughStream();
+        $response = new ChunkedStreamDecoder($stream);
+        $response->on('end', function () use (&$ended) {
+            $ended = true;
+        });
+
+        $stream->end("4\r\nWiki\r\n0\r\n\r\n");
+
+        $this->assertTrue($ended);
+    }
+
+    public function testHandleEndIncomplete()
+    {
+        $exception = null;
+        $stream = new ThroughStream();
+        $response = new ChunkedStreamDecoder($stream);
+        $response->on('error', function ($e) use (&$exception) {
+            $exception = $e;
+        });
+
+        $stream->end("4\r\nWiki");
+
+        $this->assertInstanceOf('Exception', $exception);
+    }
+}