BufferedSink.php

<?php

namespace React\Stream;

use React\Promise\Deferred;
use React\Promise\PromisorInterface;

class BufferedSink extends WritableStream implements PromisorInterface
{
    private $buffer = '';
    private $deferred;

    public function __construct()
    {
        $this->deferred = new Deferred();

        $this->on('pipe', array($this, 'handlePipeEvent'));
        $this->on('error', array($this, 'handleErrorEvent'));
    }

    public function handlePipeEvent($source)
    {
        Util::forwardEvents($source, $this, array('error'));
        $source->on('close', array($this, 'close'));
    }

    public function handleErrorEvent($e)
    {
        $this->deferred->reject($e);
    }

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

        $this->buffer .= $data;
        $this->deferred->progress($data);

        return true;
    }

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

        parent::close();
        $this->deferred->resolve($this->buffer);
    }

    public function promise()
    {
        return $this->deferred->promise();
    }

    public static function createPromise(ReadableStreamInterface $stream)
    {
        $sink = new static();
        $stream->pipe($sink);

        return $sink->promise();
    }
}