Streaming body parsers foundation

src/StreamingBodyParser/Factory.php

@@ -0,0 +1,26 @@
+<?php
+
+namespace React\Http\StreamingBodyParser;
+
+use React\Http\Request;
+
+class Factory
+{
+    /**
+     * @param Request $request
+     * @return ParserInterface
+     */
+    public static function create(Request $request)
+    {
+        $headers = $request->getHeaders();
+        $headers = array_change_key_case($headers, CASE_LOWER);
+
+        if (
+            !isset($headers['content-length'])
+        ) {
+            return NoBodyParser::create($request);
+        }
+
+        return RawBodyParser::create($request);
+    }
+}

src/StreamingBodyParser/NoBodyParser.php

@@ -0,0 +1,48 @@
+<?php
+
+namespace React\Http\StreamingBodyParser;
+
+use Evenement\EventEmitterTrait;
+use React\Http\Request;
+
+class NoBodyParser implements ParserInterface
+{
+    use EventEmitterTrait;
+
+    /**
+     * @var Request
+     */
+    private $request;
+
+    /**
+     * @param Request $request
+     * @return ParserInterface
+     */
+    public static function create(Request $request)
+    {
+        return new static($request);
+    }
+
+    /**
+     * @param Request $request
+     */
+    private function __construct(Request $request)
+    {
+        $this->request = $request;
+        $this->request->on('end', [$this, 'end']);
+    }
+
+    public function cancel()
+    {
+        $this->request->removeListener('end', [$this, 'end']);
+        $this->emit('end');
+    }
+
+    /**
+     * @internal
+     */
+    public function end()
+    {
+        $this->emit('end');
+    }
+}

src/StreamingBodyParser/ParserInterface.php

@@ -0,0 +1,32 @@
+<?php
+
+namespace React\Http\StreamingBodyParser;
+
+use Evenement\EventEmitterInterface;
+use React\Http\Request;
+
+/**
+ * Parser can emit the following events:
+ *     end - When done or canceled (required)
+ *     body - Raw request body (optional)
+ *     post - Post field (optional)
+ *     file - Uploaded file (optional)
+ */
+interface ParserInterface extends EventEmitterInterface
+{
+    /**
+     * Factory method creating the parser.
+     *
+     * @param Request $request
+     * @return ParserInterface
+     */
+    public static function create(Request $request);
+
+    /**
+     * Cancel parsing the request body stream.
+     * Parser will still emit end event to any listeners.
+     *
+     * @return void
+     */
+    public function cancel();
+}

src/StreamingBodyParser/RawBodyParser.php

@@ -0,0 +1,56 @@
+<?php
+
+namespace React\Http\StreamingBodyParser;
+
+use Evenement\EventEmitterTrait;
+use React\Http\Request;
+use React\Promise\ExtendedPromiseInterface;
+
+class RawBodyParser implements ParserInterface
+{
+    use EventEmitterTrait;
+
+    /**
+     * @var ExtendedPromiseInterface
+     */
+    private $promise;
+
+    /**
+     * @param Request $request
+     * @return ParserInterface
+     */
+    public static function create(Request $request)
+    {
+        return new static($request);
+    }
+
+    /**
+     * @param Request $request
+     */
+    private function __construct(Request $request)
+    {
+        $headers = $request->getHeaders();
+        $headers = array_change_key_case($headers, CASE_LOWER);
+
+        $this->promise = ContentLengthBufferedSink::createPromise(
+            $request,
+            $headers['content-length']
+        )->then([$this, 'finish']);
+    }
+
+    /**
+     * @param string $buffer
+     *
+     * @internal
+     */
+    public function finish($buffer)
+    {
+        $this->emit('body', [$buffer]);
+        $this->emit('end');
+    }
+
+    public function cancel()
+    {
+        $this->promise->cancel();
+    }
+}

tests/StreamingBodyParser/FactoryTest.php

@@ -0,0 +1,26 @@
+<?php
+
+namespace React\Tests\Http\StreamingBodyParser;
+
+use React\Http\StreamingBodyParser\Factory;
+use React\Http\Request;
+use React\Tests\Http\TestCase;
+
+class FactoryTest extends TestCase
+{
+    public function testContentLength()
+    {
+        $request = new Request('POST', 'http://example.com/', [], 1.1, [
+            'content-length' => 123,
+        ]);
+        $parser = Factory::create($request);
+        $this->assertInstanceOf('React\Http\StreamingBodyParser\RawBodyParser', $parser);
+    }
+
+    public function testNoHeaders()
+    {
+        $request = new Request('POST', 'http://example.com/');
+        $parser = Factory::create($request);
+        $this->assertInstanceOf('React\Http\StreamingBodyParser\NoBodyParser', $parser);
+    }
+}

tests/StreamingBodyParser/NoBodyParserTest.php

@@ -0,0 +1,26 @@
+<?php
+
+namespace React\Tests\Http\StreamingBodyParser;
+
+use React\Http\Request;
+use React\Http\StreamingBodyParser\NoBodyParser;
+use React\Tests\Http\TestCase;
+
+class NoBodyParserTest extends TestCase
+{
+    public function testRequestEnd()
+    {
+        $request = new Request('POST', 'http://example.com/');
+        $parser = NoBodyParser::create($request);
+        $parser->on('end', $this->expectCallableOnce());
+        $request->close();
+    }
+
+    public function testCancelParser()
+    {
+        $request = new Request('POST', 'http://example.com/');
+        $parser = NoBodyParser::create($request);
+        $parser->on('end', $this->expectCallableOnce());
+        $parser->cancel();
+    }
+}

tests/StreamingBodyParser/RawBodyParserTest.php

@@ -0,0 +1,24 @@
+<?php
+
+namespace React\Tests\Http\StreamingBodyParser;
+
+use React\Http\StreamingBodyParser\RawBodyParser;
+use React\Http\Request;
+use React\Tests\Http\TestCase;
+
+class RawBodyParserTest extends TestCase
+{
+    public function testNoContentLength()
+    {
+        $body = '';
+        $request = new Request('POST', 'http://example.com/', [], 1.1, [
+            'content-length' => 3,
+        ]);
+        $parser = RawBodyParser::create($request);
+        $parser->on('body', function ($rawBody) use (&$body) {
+            $body = $rawBody;
+        });
+        $request->emit('data', ['abc']);
+        $this->assertSame('abc', $body);
+    }
+}