Gecko [ unitednationsplay.com ]

Name	Size	Permission	Date
UnwrapReadableStream.php	3.65 KB	-rw-r--r--	2017-12-22 09:00
UnwrapWritableStream.php	4.42 KB	-rw-r--r--	2017-12-22 09:00
functions.php	6.17 KB	-rw-r--r--	2017-12-22 09:00
functions_include.php	137 B	-rw-r--r--	2017-12-22 09:00

Rename

<?php

namespace React\Promise\Stream;

use Evenement\EventEmitter;
use InvalidArgumentException;
use React\Promise\CancellablePromiseInterface;
use React\Promise\PromiseInterface;
use React\Stream\WritableStreamInterface;

/**
 * @internal
 * @see unwrapWritable() instead
 */
class UnwrapWritableStream extends EventEmitter implements WritableStreamInterface
{
    private $promise;
    private $stream;
    private $buffer = '';
    private $closed = false;
    private $ending = false;

/**
     * Instantiate new unwrapped writable stream for given `Promise` which resolves with a `WritableStreamInterface`.
     *
     * @param PromiseInterface $promise Promise<WritableStreamInterface, Exception>
     */
    public function __construct(PromiseInterface $promise)
    {
        $out = $this;
        $store  =& $this->stream;
        $buffer =& $this->buffer;
        $ending =& $this->ending;
        $closed =& $this->closed;

$this->promise = $promise->then(
            function ($stream) {
                if (!($stream instanceof WritableStreamInterface)) {
                    throw new InvalidArgumentException('Not a writable stream');
                }
                return $stream;
            }
        )->then(
            function (WritableStreamInterface $stream) use ($out, &$store, &$buffer, &$ending, &$closed) {
                // stream is already closed, make sure to close output stream
                if (!$stream->isWritable()) {
                    $out->close();
                    return $stream;
                }

// resolves but output is already closed, make sure to close stream silently
                if ($closed) {
                    $stream->close();
                    return $stream;
                }

// forward drain events for back pressure
                $stream->on('drain', function () use ($out) {
                    $out->emit('drain', array($out));
                });

// error events cancel output stream
                $stream->on('error', function ($error) use ($out) {
                    $out->emit('error', array($error, $out));
                    $out->close();
                });

// close both streams once either side closes
                $stream->on('close', array($out, 'close'));
                $out->on('close', array($stream, 'close'));

if ($buffer !== '') {
                    // flush buffer to stream and check if its buffer is not exceeded
                    $drained = $stream->write($buffer) !== false;
                    $buffer = '';

if ($drained) {
                        // signal drain event, because the output stream previous signalled a full buffer
                        $out->emit('drain', array($out));
                    }
                }

if ($ending) {
                    $stream->end();
                } else {
                    $store = $stream;
                }

return $stream;
            },
            function ($e) use ($out, &$closed) {
                if (!$closed) {
                    $out->emit('error', array($e, $out));
                    $out->close();
                }
            }
        );
    }

public function write($data)
    {
        if ($this->ending) {
            return;
        }

// forward to inner stream if possible
        if ($this->stream !== null) {
            return $this->stream->write($data);
        }

// append to buffer and signal the buffer is full
        $this->buffer .= $data;
        return false;
    }

public function end($data = null)
    {
        if ($this->ending) {
            return;
        }

$this->ending = true;

// forward to inner stream if possible
        if ($this->stream !== null) {
            return $this->stream->end($data);
        }

// append to buffer
        if ($data !== null) {
            $this->buffer .= $data;
        }
    }

public function isWritable()
    {
        return !$this->ending;
    }

public function close()
    {
        if ($this->closed) {
            return;
        }

$this->buffer = '';
        $this->ending = true;
        $this->closed = true;

// try to cancel promise once the stream closes
        if ($this->promise instanceof CancellablePromiseInterface) {
            $this->promise->cancel();
        }

$this->emit('close', array($this));
    }
}