-
-
Notifications
You must be signed in to change notification settings - Fork 17
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #50 from clue-labs/stream-semantics
Change JSON stream to always report `data` events instead of `progress`, follow strict stream semantics, support backpressure and improve error handling
- Loading branch information
Showing
13 changed files
with
362 additions
and
144 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,129 @@ | ||
| <?php | ||
|
|
||
| namespace Clue\React\Docker\Io; | ||
|
|
||
| use Clue\JsonStream\StreamingJsonParser; | ||
| use React\Stream\ReadableStreamInterface; | ||
| use React\Stream\Util; | ||
| use React\Stream\WritableStreamInterface; | ||
| use Evenement\EventEmitter; | ||
|
|
||
| /** | ||
| * Parser for Docker's JSON stream format used for log messages etc. | ||
| * | ||
| * @internal | ||
| */ | ||
| class ReadableJsonStream extends EventEmitter implements ReadableStreamInterface | ||
| { | ||
| private $closed = false; | ||
| private $input; | ||
| private $parser; | ||
|
|
||
| public function __construct(ReadableStreamInterface $input) | ||
| { | ||
| $this->input = $input; | ||
| $this->parser = $parser = new StreamingJsonParser(); | ||
| if (!$input->isReadable()) { | ||
| $this->close(); | ||
| return; | ||
| } | ||
|
|
||
| // pass all input data chunks through the parser | ||
| $input->on('data', array($this, 'handleData')); | ||
|
|
||
| // forward end event to output | ||
| $out = $this; | ||
| $closed =& $this->closed; | ||
| $input->on('end', function () use ($out, $parser, &$closed) { | ||
| // ignore duplicate end events | ||
| if ($closed) { | ||
| return; | ||
| } | ||
|
|
||
| if ($parser->isEmpty()) { | ||
| $out->emit('end'); | ||
| } else { | ||
| $out->emit('error', array(new \RuntimeException('Stream ended within incomplete JSON data'))); | ||
| } | ||
| $out->close(); | ||
| }); | ||
|
|
||
| // forward error event to output | ||
| $input->on('error', function ($error) use ($out) { | ||
| $out->emit('error', array($error)); | ||
| $out->close(); | ||
| }); | ||
|
|
||
| // forward close event to output | ||
| $input->on('close', function () use ($out) { | ||
| $out->close(); | ||
| }); | ||
| } | ||
|
|
||
| /** | ||
| * push the given stream chunk into the parser buffer and try to extract all JSON messages | ||
| * | ||
| * @internal | ||
| * @param string $data | ||
| */ | ||
| public function handleData($data) | ||
| { | ||
| // forward each data chunk to the streaming JSON parser | ||
| try { | ||
| $objects = $this->parser->push($data); | ||
| } catch (\Exception $e) { | ||
| $this->emit('error', array($e)); | ||
| $this->close(); | ||
| return; | ||
| } | ||
|
|
||
| foreach ($objects as $object) { | ||
| // stop emitting data if stream is already closed | ||
| if ($this->closed) { | ||
| return; | ||
| } | ||
|
|
||
| if (isset($object['error'])) { | ||
| $this->emit('error', array(new \RuntimeException($object['error']))); | ||
| $this->close(); | ||
| return; | ||
| } | ||
| $this->emit('data', array($object)); | ||
| } | ||
| } | ||
|
|
||
| public function pause() | ||
| { | ||
| $this->input->pause(); | ||
| } | ||
|
|
||
| public function resume() | ||
| { | ||
| $this->input->resume(); | ||
| } | ||
|
|
||
| public function isReadable() | ||
| { | ||
| return $this->input->isReadable(); | ||
| } | ||
|
|
||
| public function pipe(WritableStreamInterface $dest, array $options = array()) | ||
| { | ||
| return Util::pipe($this, $dest, $options); | ||
| } | ||
|
|
||
| public function close() | ||
| { | ||
| if ($this->closed) { | ||
| return; | ||
| } | ||
|
|
||
| $this->closed = true; | ||
|
|
||
| // closing output stream closes input stream | ||
| $this->input->close(); | ||
|
|
||
| $this->emit('close'); | ||
| $this->removeAllListeners(); | ||
| } | ||
| } |
Oops, something went wrong.