Thruster\Component\Stream\BufferedSink

Complexity

Total Complexity

8

Size/Duplication

Total Lines	57
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	4

Test Coverage

Coverage

0%

7 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	9	1
A	handlePipeEvent()	0	4	1
A	handleErrorEvent()	0	4	1
A	write()	0	4	1
A	close()	0	10	2
A	promise()	0	4	1
A	createPromise()	0	8	1

<?php

namespace Thruster\Component\Stream;

use Thruster\Component\Promise\Deferred;
use Thruster\Component\Promise\PromiseInterface;
use Thruster\Component\Promise\PromisorInterface;

/**
 * Class BufferedSink
 *
 * @package Thruster\Component\Stream
 * @author  Aurimas Niekis <aurimas@niekis.lt>
 */
class BufferedSink extends WritableStream implements PromisorInterface
{
    use UtilsTrait;

    private $buffer = '';
    private $deferred;

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

        $this->on('pipe', [$this, 'handlePipeEvent']);
        $this->on('error', [$this, 'handleErrorEvent']);

        parent::__construct();
    }

    public function handlePipeEvent($source)
    {
        $this->forwardEvents($source, $this, ['error']);
    }

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

    public function write($data)
    {
        $this->buffer .= $data;
    }

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

        parent::close();

        $this->deferred->resolve($this->buffer);
    }

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

    public static function createPromise(ReadableStreamInterface $stream) : PromiseInterface
    {
        $sink = new static();

        $stream->pipe($sink);

        return $sink->promise();
    }
}