Thruster\Component\Stream\Stream

Complexity

Total Complexity

24

Size/Duplication

Total Lines	232
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	4

Test Coverage

Coverage

83.12%

14 Methods

Rating	Name	Duplication	Size	Complexity
B	__construct()	0	30	3
A	isReadable()	0	4	1
A	isWritable()	0	4	1
A	pause()	0	6	1
A	resume()	0	8	2
A	write()	0	8	2
A	close()	0	21	3
A	end()	0	16	2
A	pipe()	0	6	1
A	handleData()	0	10	3
A	handleClose()	0	6	2
A	getBuffer()	0	4	1
A	getBufferSize()	0	4	1
A	setBufferSize()	0	6	1

<?php

namespace Thruster\Component\Stream;

use InvalidArgumentException;
use Thruster\Component\EventEmitter\EventEmitterInterface;
use Thruster\Component\EventEmitter\EventEmitterTrait;
use Thruster\Component\EventLoop\EventLoopInterface;

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

    /**
     * @var int
     */
    protected $bufferSize;

    /**
     * @var resource
     */
    protected $stream;

    /**
     * @var bool
     */
    protected $readable;

    /**
     * @var bool
     */
    protected $writable;

    /**
     * @var bool
     */
    protected $closing;

    /**
     * @var EventLoopInterface
     */
    protected $loop;

    /**
     * @var Buffer
     */
    protected $buffer;

    /**
     * Stream constructor.
     *
     * @param resource           $stream
     * @param EventLoopInterface $loop
     * @param int                $bufferSize
     */
    public function __construct($stream, EventLoopInterface $loop, int $bufferSize = 4096)
    {
        $this->stream = $stream;

        if (false === is_resource($this->stream) || "stream" !== get_resource_type($this->stream)) {
            throw new InvalidArgumentException('First parameter must be a valid stream resource');
        }

        $this->bufferSize = $bufferSize;
        $this->readable   = true;
        $this->writable   = true;
        $this->closing    = false;

        stream_set_blocking($this->stream, 0);
        stream_set_read_buffer($this->stream, 0);

        $this->loop   = $loop;
        $this->buffer = new Buffer($this->stream, $this->loop);

        $this->buffer->on('error', function ($error) {
            $this->emit('error', [$error, $this]);
            $this->close();
        });

        $this->buffer->on('drain', function () {
            $this->emit('drain', [$this]);
        });

        $this->resume();
    }

    /**
     * @inheritdoc
     */
    public function isReadable() : bool
    {
        return $this->readable;
    }

    /**
     * @inheritdoc
     */
    public function isWritable() : bool
    {
        return $this->writable;
    }

    /**
     * @inheritdoc
     */
    public function pause() : self
    {
        $this->loop->removeReadStream($this->stream);

        return $this;
    }

    /**
     * @inheritdoc
     */
    public function resume() : self
    {
        if ($this->isReadable()) {
            $this->loop->addReadStream($this->stream, [$this, 'handleData']);
        }

        return $this;
    }

    /**
     * @inheritdoc
     */
    public function write($data)
    {
        if (false === $this->isWritable()) {
            return;
        }

        return $this->buffer->write($data);
    }

    /**
     * @inheritdoc
     */
    public function close() : self
    {
        if (false === $this->isReadable() && false === $this->closing) {
            return $this;
        }

        $this->closing = false;

        $this->readable = false;
        $this->writable = false;

        $this->emit('end', [$this]);
        $this->emit('close', [$this]);
        $this->loop->removeStream($this->stream);
        $this->buffer->removeListeners();
        $this->removeListeners();

        $this->handleClose();

        return $this;
    }

    /**
     * @inheritdoc
     */
    public function end($data = null) : self
    {
        if (false === $this->isWritable()) {
            return $this;
        }

        $this->closing  = true;
        $this->readable = false;
        $this->writable = false;

        $this->buffer->on('close', [$this, 'close']);

        $this->buffer->end($data);

        return $this;
    }

    /**
     * @inheritdoc
     */
    public function pipe(WritableStreamInterface $destination, array $options = [])
    {
        $this->pipeAll($this, $destination, $options);

        return $destination;
    }

    /**
     * @param resource $stream
     */
    public function handleData($stream)
    {
        $data = fread($stream, $this->bufferSize);

        $this->emit('data', [$data, $this]);

        if (false === is_resource($stream) || feof($stream)) {
            $this->end();
        }
    }

    public function handleClose()
    {
        if (is_resource($this->stream)) {
            fclose($this->stream);
        }
    }

    /**
     * @return Buffer
     */
    public function getBuffer()
    {
        return $this->buffer;
    }

    /**
     * @return int
     */
    public function getBufferSize() : int
    {
        return $this->bufferSize;
    }

    /**
     * @param int $bufferSize
     *
     * @return $this
     */
    public function setBufferSize(int $bufferSize) : self
    {
        $this->bufferSize = $bufferSize;

        return $this;
    }
}