[WIP] Streaming parser bufferedsink

src/StreamingBodyParser/BodyBufferedSink.php

@@ -0,0 +1,32 @@
+<?php
+
+namespace React\Http\StreamingBodyParser;
+
+use React\Http\File;
+use React\Promise;
+
+class BodyBufferedSink
+{
+    /**
+     * @param ParserInterface $parser
+     * @return Promise\PromiseInterface
+     */
+    public static function createPromise(ParserInterface $parser)
+    {
+        if ($parser instanceof NoBodyParser) {
+            return Promise\resolve('');
+        }
+
+        $deferred = new Promise\Deferred();
+        $body = '';
+
+        $parser->on('body', function ($rawBody) use (&$body) {
+            $body = $rawBody;
+        });
+        $parser->on('end', function () use ($deferred, &$body) {
+            $deferred->resolve($body);
+        });
+
+        return $deferred->promise();
+    }
+}

src/StreamingBodyParser/BufferedSink.php

@@ -0,0 +1,90 @@
+<?php
+
+namespace React\Http\StreamingBodyParser;
+
+use React\Http\File;
+use React\Promise\Deferred;
+use React\Promise\PromiseInterface;
+use React\Stream\BufferedSink as StreamBufferedSink;
+
+class BufferedSink
+{
+    /**
+     * @param ParserInterface $parser
+     * @return PromiseInterface
+     */
+    public static function createPromise(ParserInterface $parser)
+    {
+        if ($parser instanceof NoBodyParser) {
+            return \React\Promise\resolve([
+                'post' => [],
+                'files' => [],
+                'body' => '',
+            ]);
+        }
+
+        $deferred = new Deferred();
+        $postFields = [];
+        $files = [];
+        $body = '';
+        $parser->on('post', function ($key, $value) use (&$postFields) {
+            self::extractPost($postFields, $key, $value);
+        });
+        $parser->on('file', function ($name, File $file) use (&$files) {
+            StreamBufferedSink::createPromise($file->getStream())->done(function ($buffer) use ($name, $file, &$files) {
+                $files[] = [
+                    'name' => $name,
+                    'file' => $file,
+                    'buffer' => $buffer,
+                ];
+            });
+        });
+        $parser->on('body', function ($rawBody) use (&$body) {
+            $body = $rawBody;
+        });
+        $parser->on('end', function () use ($deferred, &$postFields, &$files, &$body) {
+            $deferred->resolve([
+                'post' => $postFields,
+                'files' => $files,
+                'body' => $body,
+            ]);
+        });
+
+        return $deferred->promise();
+    }
+
+    public static function extractPost(&$postFields, $key, $value)
+    {
+        $chunks = explode('[', $key);
+        if (count($chunks) == 1) {
+            $postFields[$key] = $value;
+            return;
+        }
+
+        $chunkKey = $chunks[0];
+        if (!isset($postFields[$chunkKey])) {
+            $postFields[$chunkKey] = [];
+        }
+
+        $parent = &$postFields;
+        for ($i = 1; $i < count($chunks); $i++) {
+            $previousChunkKey = $chunkKey;
+            if (!isset($parent[$previousChunkKey])) {
+                $parent[$previousChunkKey] = [];
+            }
+            $parent = &$parent[$previousChunkKey];
+            $chunkKey = $chunks[$i];
+
+            if ($chunkKey == ']') {
+                $parent[] = $value;
+                return;
+            }
+
+            $chunkKey = rtrim($chunkKey, ']');
+            if ($i == count($chunks) - 1) {
+                $parent[$chunkKey] = $value;
+                return;
+            }
+        }
+    }
+}

src/StreamingBodyParser/FilesBufferedSink.php

@@ -0,0 +1,39 @@
+<?php
+
+namespace React\Http\StreamingBodyParser;
+
+use React\Http\File;
+use React\Promise;
+use React\Stream\BufferedSink as StreamBufferedSink;
+
+class FilesBufferedSink
+{
+    /**
+     * @param ParserInterface $parser
+     * @return Promise\PromiseInterface
+     */
+    public static function createPromise(ParserInterface $parser)
+    {
+        if ($parser instanceof NoBodyParser) {
+            return Promise\resolve([]);
+        }
+
+        $deferred = new Promise\Deferred();
+        $files = [];
+
+        $parser->on('file', function ($name, File $file) use (&$files) {
+            StreamBufferedSink::createPromise($file->getStream())->done(function ($buffer) use ($name, $file, &$files) {
+                $files[] = [
+                    'name' => $name,
+                    'file' => $file,
+                    'buffer' => $buffer,
+                ];
+            });
+        });
+        $parser->on('end', function () use ($deferred, &$files) {
+            $deferred->resolve($files);
+        });
+
+        return $deferred->promise();
+    }
+}

src/StreamingBodyParser/PostBufferedSink.php

@@ -0,0 +1,66 @@
+<?php
+
+namespace React\Http\StreamingBodyParser;
+
+use React\Promise;
+
+class PostBufferedSink
+{
+    /**
+     * @param ParserInterface $parser
+     * @return Promise\PromiseInterface
+     */
+    public static function createPromise(ParserInterface $parser)
+    {
+        if ($parser instanceof NoBodyParser) {
+            return Promise\resolve([]);
+        }
+
+        $deferred = new Promise\Deferred();
+        $postFields = [];
+
+        $parser->on('post', function ($key, $value) use (&$postFields) {
+            self::extractPost($postFields, $key, $value);
+        });
+        $parser->on('end', function () use ($deferred, &$postFields) {
+            $deferred->resolve($postFields);
+        });
+
+        return $deferred->promise();
+    }
+
+    public static function extractPost(&$postFields, $key, $value)
+    {
+        $chunks = explode('[', $key);
+        if (count($chunks) == 1) {
+            $postFields[$key] = $value;
+            return;
+        }
+
+        $chunkKey = $chunks[0];
+        if (!isset($postFields[$chunkKey])) {
+            $postFields[$chunkKey] = [];
+        }
+
+        $parent = &$postFields;
+        for ($i = 1; $i < count($chunks); $i++) {
+            $previousChunkKey = $chunkKey;
+            if (!isset($parent[$previousChunkKey])) {
+                $parent[$previousChunkKey] = [];
+            }
+            $parent = &$parent[$previousChunkKey];
+            $chunkKey = $chunks[$i];
+
+            if ($chunkKey == ']') {
+                $parent[] = $value;
+                return;
+            }
+
+            $chunkKey = rtrim($chunkKey, ']');
+            if ($i == count($chunks) - 1) {
+                $parent[$chunkKey] = $value;
+                return;
+            }
+        }
+    }
+}

tests/StreamingBodyParser/BodyBufferedSinkTest.php

@@ -0,0 +1,34 @@
+<?php
+
+namespace React\Tests\Http\StreamingBodyParser;
+
+use Clue\React\Block;
+use React\EventLoop\Factory;
+use React\Http\StreamingBodyParser\BodyBufferedSink;
+use React\Http\StreamingBodyParser\NoBodyParser;
+use React\Http\Request;
+use React\Tests\Http\TestCase;
+
+class BodyBufferedSinkTest extends TestCase
+{
+    public function testDoneParser()
+    {
+        $parser = new NoBodyParser(new Request('get', 'http://example.com'));
+        $deferredStream = BodyBufferedSink::createPromise($parser);
+        $result = Block\await($deferredStream, Factory::create(), 10);
+        $this->assertSame('', $result);
+    }
+
+    public function testDeferredStream()
+    {
+        $parser = new DummyParser(new Request('get', 'http://example.com'));
+        $deferredStream = BodyBufferedSink::createPromise($parser);
+
+        $parser->emit('body', ['abc']);
+        $parser->emit('end');
+
+        $result = Block\await($deferredStream, Factory::create(), 10);
+
+        $this->assertSame('abc', $result);
+    }
+}

tests/StreamingBodyParser/BufferedSinkTest.php

@@ -0,0 +1,100 @@
+<?php
+
+namespace React\Tests\Http\StreamingBodyParser;
+
+use Clue\React\Block;
+use React\EventLoop\Factory;
+use React\Http\File;
+use React\Http\StreamingBodyParser\BufferedSink;
+use React\Http\StreamingBodyParser\NoBodyParser;
+use React\Http\Request;
+use React\Stream\ThroughStream;
+use React\Tests\Http\TestCase;
+
+class BufferedSinkTest extends TestCase
+{
+    public function testDoneParser()
+    {
+        $parser = new NoBodyParser(new Request('get', 'http://example.com'));
+        $deferredStream = BufferedSink::createPromise($parser);
+        $result = Block\await($deferredStream, Factory::create(), 10);
+        $this->assertSame([
+            'post' => [],
+            'files' => [],
+            'body' => '',
+        ], $result);
+    }
+
+    public function testDeferredStream()
+    {
+        $parser = new DummyParser(new Request('get', 'http://example.com'));
+        $deferredStream = BufferedSink::createPromise($parser);
+        $parser->emit('post', ['foo', 'bar']);
+        $parser->emit('post', ['array[]', 'foo']);
+        $parser->emit('post', ['array[]', 'bar']);
+        $parser->emit('post', ['dem[two]', 'bar']);
+        $parser->emit('post', ['dom[two][]', 'bar']);
+
+        $stream = new ThroughStream();
+        $file = new File('bar.ext', 'text', $stream);
+        $parser->emit('file', ['foo', $file]);
+        $stream->end('foo.bar');
+
+        $parser->emit('body', ['abc']);
+
+        $parser->emit('end');
+
+        $result = Block\await($deferredStream, Factory::create(), 10);
+        $this->assertSame([
+            'foo' => 'bar',
+            'array' => [
+                'foo',
+                'bar',
+            ],
+            'dem' => [
+                'two' => 'bar',
+            ],
+            'dom' => [
+                'two' => [
+                    'bar',
+                ],
+            ],
+        ], $result['post']);
+
+        $this->assertSame('foo', $result['files'][0]['name']);
+        $this->assertSame('bar.ext', $result['files'][0]['file']->getFilename());
+        $this->assertSame('text', $result['files'][0]['file']->getContentType());
+        $this->assertSame('foo.bar', $result['files'][0]['buffer']);
+
+        $this->assertSame('abc', $result['body']);
+    }
+
+    public function testExtractPost()
+    {
+        $postFields = [];
+        BufferedSink::extractPost($postFields, 'dem', 'value');
+        BufferedSink::extractPost($postFields, 'dom[one][two][]', 'value_a');
+        BufferedSink::extractPost($postFields, 'dom[one][two][]', 'value_b');
+        BufferedSink::extractPost($postFields, 'dam[]', 'value_a');
+        BufferedSink::extractPost($postFields, 'dam[]', 'value_b');
+        BufferedSink::extractPost($postFields, 'dum[sum]', 'value');
+        $this->assertSame([
+            'dem' => 'value',
+            'dom' => [
+                'one' => [
+                    'two' => [
+                        'value_a',
+                        'value_b',
+                    ],
+                ],
+            ],
+            'dam' => [
+                'value_a',
+                'value_b',
+            ],
+            'dum' => [
+                'sum' => 'value',
+            ],
+        ], $postFields);
+    }
+}