Decode chunked transfer encoding for incoming requests

src/ChunkedDecoder.php

@@ -0,0 +1,144 @@
+<?php
+namespace React\Http;
+
+use Evenement\EventEmitter;
+use React\Stream\ReadableStreamInterface;
+use React\Stream\WritableStreamInterface;
+use React\Stream\Util;
+use Exception;
+
+/** @internal */
+class ChunkedDecoder extends EventEmitter implements ReadableStreamInterface
+{
+ const CRLF = "\r\n";
+
+ private $closed = false;
+ private $input;
+ private $buffer = '';
+ private $chunkSize = 0;
+ private $actualChunksize = 0;
+
+ public function __construct(ReadableStreamInterface $input)
+ {
+ $this->input = $input;
+
+ $this->input->on('data', array($this, 'handleChunkHeader'));
+ $this->input->on('end', array($this, 'handleEnd'));
+ $this->input->on('error', array($this, 'handleError'));
+ $this->input->on('close', array($this, 'close'));
+ }
+
+ public function isReadable()
+ {
+ return ! $this->closed && $this->input->isReadable();
+ }
+
+ public function pause()
+ {
+ $this->input->pause();
+ }
+
+ public function resume()
+ {
+ $this->input->resume();
+ }
+
+ public function pipe(WritableStreamInterface $dest, array $options = array())
+ {
+ Util::pipe($this, $dest, $options);
+
+ return $dest;
+ }
+
+ public function close()
+ {
+ if ($this->closed) {
+ return;
+ }
+
+ $this->buffer = '';
+
+ $this->closed = true;
+
+ $this->input->close();
+
+ $this->emit('close');
+ $this->removeAllListeners();
+ }
+
+
+ /** @internal */
+ public function handleChunkHeader($data)
+ {
+ $this->buffer .= $data;
+
+ $hexValue = strtok($this->buffer, static::CRLF);
+ if (dechex(hexdec($hexValue)) != $hexValue) {
+ $this->emit('error', array(new \Exception('Unable to identify ' . $hexValue . 'as hexadecimal number')));
+ $this->close();
+ return;
+ }
+
+ if (strpos($this->buffer, static::CRLF) !== false) {
+ $this->chunkSize = hexdec($hexValue);
+
+ $data = substr($this->buffer, strlen($hexValue) + 2);
+ $this->buffer = '';
+ // Chunk header is complete
+ $this->input->removeListener('data', array($this, 'handleChunkHeader'));
+ $this->input->on('data', array($this, 'handleChunkData'));
+ if ($data !== '') {
+ $this->input->emit('data', array($data));
+ }
+ }
+ }
+
+ /** @internal */
+ public function handleChunkData($data)
+ {
+ $this->buffer .= $data;
+ $chunk = substr($this->buffer, 0, $this->chunkSize);
+ $this->actualChunksize = strlen($chunk);
+
+ if ($this->chunkSize === $this->actualChunksize) {
+ if ($this->chunkSize === 0 && strpos($this->buffer, static::CRLF) !== false) {
+ $this->emit('end', array());
+ $this->close();
+ return;
+ }
+
+ if (strpos($this->buffer, static::CRLF) === false) {
+ return;
+ }
+
+ $data = substr($this->buffer , $this->chunkSize + 2);
+ $this->emit('data', array($chunk));
+
+ $this->buffer = '';
+ $this->chunkSize = 0;
+ // chunk body is complete
+ $this->input->removeListener('data', array($this, 'handleChunkData'));
+ $this->input->on('data', array($this, 'handleChunkHeader'));
+ if ($data !== '') {
+ $this->input->emit('data', array($data));
+ }
+ }
+ }
+
+ /** @internal */
+ public function handleEnd()
+ {
+ if (!$this->closed) {
+ $this->buffer = '';
+ $this->emit('end');
+ $this->close();
+ }
+ }
+
+ /** @internal */
+ public function handleError(\Exception $e)
+ {
+ $this->emit('error', array($e));
+ $this->close();
+ }
+}

src/Server.php

@@ -22,26 +22,21 @@ public function __construct(SocketServerInterface $io)
  // TODO: multipart parsing
 
  $parser = new RequestHeaderParser();
- $parser->on('headers', function (Request $request, $bodyBuffer) use ($conn, $parser, $that) {
+ $listener = array($parser, 'feed');
+
+ $parser->on('headers', function (Request $request, $bodyBuffer) use ($conn, $parser, $that, $listener) {
  // attach remote ip to the request as metadata
  $request->remoteAddress = $conn->getRemoteAddress();
 
  // forward pause/resume calls to underlying connection
  $request->on('pause', array($conn, 'pause'));
  $request->on('resume', array($conn, 'resume'));
 
- $that->handleRequest($conn, $request, $bodyBuffer);
+ $conn->removeListener('data', $listener);
 
- $conn->removeListener('data', array($parser, 'feed'));
- $conn->on('end', function () use ($request) {
- $request->emit('end');
- });
- $conn->on('data', function ($data) use ($request) {
- $request->emit('data', array($data));
- });
+ $that->handleRequest($conn, $request, $bodyBuffer);
  });
 
- $listener = array($parser, 'feed');
  $conn->on('data', $listener);
  $parser->on('error', function() use ($conn, $listener, $that) {
  // TODO: return 400 response
@@ -62,7 +57,24 @@ public function handleRequest(ConnectionInterface $conn, Request $request, $body
  return;
  }
 
+ $stream = $conn;
+ if ($request->hasHeader('Transfer-Encoding')) {
+ $transferEncodingHeader = $request->getHeader('Transfer-Encoding');
+ // 'chunked' must always be the final value of 'Transfer-Encoding' according to: https://tools.ietf.org/html/rfc7230#section-3.3.1
+ if (strtolower(end($transferEncodingHeader)) === 'chunked') {
+ $stream = new ChunkedDecoder($conn);
+ }
+ }
+
+ $stream->on('data', function ($data) use ($request) {
+ $request->emit('data', array($data));
+ });
+
+ $stream->on('end', function () use ($request) {
+ $request->emit('end', array());
+ });
+
  $this->emit('request', array($request, $response));
- $request->emit('data', array($bodyBuffer));
+ $conn->emit('data', array($bodyBuffer));
  }
 }