Dazzle\Stream\StreamReader

Complexity

Total Complexity

22

Size/Duplication

Total Lines	187
Duplicated Lines	30.48 %

Coupling/Cohesion

Components	1
Dependencies	3

Test Coverage

Coverage

80%

Importance

Changes

0

11 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	16	16	2
A	__destruct()	0	9	1
A	isPaused()	0	4	1
A	setBufferSize()	0	4	1
A	getBufferSize()	0	4	1
A	pause()	0	9	2
A	resume()	0	12	4
A	read()	18	18	4
B	handleRead()	23	23	4
A	getHandleReadFunction()	0	4	1
A	handleClose()	0	6	1

<?php

namespace Dazzle\Stream;

use Dazzle\Loop\LoopAwareTrait;
use Dazzle\Loop\LoopInterface;
use Dazzle\Throwable\Exception\Logic\InvalidArgumentException;
use Dazzle\Throwable\Exception\Runtime\ReadException;
use Error;
use Exception;

class StreamReader extends Sync\StreamReader implements StreamReaderInterface
{
    use LoopAwareTrait;

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

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

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

    /**
     * @param resource $resource
     * @param LoopInterface $loop
     * @param bool $autoClose
     * @throws InvalidArgumentException
     */
    public function __construct($resource, LoopInterface $loop, $autoClose = true)

This method seems to be duplicated in your project.

    {
        parent::__construct($resource, $autoClose);

        if (function_exists('stream_set_read_buffer'))
        {
            stream_set_read_buffer($this->resource, 0);
        }

        $this->loop = $loop;
        $this->reading = false;
        $this->readingStarted = false;
        $this->paused = true;

        $this->resume();
    }

    /**
     *
     */
    public function __destruct()
    {
        parent::__destruct();

        unset($this->loop);
        unset($this->reading);
        unset($this->readingStarted);
        unset($this->paused);
    }

    /**
     * @override
     * @inheritDoc
     */
    public function isPaused()
    {
        return $this->paused;
    }

    /**
     * @override
     * @inheritDoc
     */
    public function setBufferSize($bufferSize)
    {
        $this->bufferSize = $bufferSize;
    }

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

    /**
     * @override
     * @inheritDoc
     */
    public function pause()
    {
        if (!$this->paused)
        {
            $this->paused = true;
            $this->reading = false;
            $this->loop->removeReadStream($this->resource);
        }
    }

    /**
     * @override
     * @inheritDoc
     */
    public function resume()
    {
        if ($this->readable && $this->paused)
        {
            $this->paused = false;
            if ($this->readingStarted)
            {
                $this->reading = true;
                $this->loop->addReadStream($this->resource, $this->getHandleReadFunction());
            }
        }
    }

    /**
     * @override
     * @inheritDoc
     */
    public function read($length = null)

This method seems to be duplicated in your project.

    {
        if (!$this->readable)
        {
            return $this->throwAndEmitException(
                new ReadException('Stream is no longer readable.')
            );
        }

        if (!$this->reading && !$this->paused)
        {
            $this->reading = true;
            $this->readingStarted = true;
            $this->loop->addReadStream($this->resource, $this->getHandleReadFunction());
        }

        return '';
    }

    /**
     * Handle the incoming stream.
     *
     * @internal
     */
    public function handleRead()

This method seems to be duplicated in your project.

    {
        $length = $this->bufferSize;
        $ret = fread($this->resource, $length);

        if ($ret === false)
        {
            $this->emit('error', [ $this, new ReadException('Error occurred while reading from the stream resource.') ]);
            return;
        }

        if ($ret !== '')
        {
            $this->emit('data', [ $this, $ret ]);

            if (strlen($ret) < $length)
            {
                $this->loop->removeReadStream($this->resource);
                $this->reading = false;
                $this->emit('end', [ $this ]);
            }
        }
    }

    /**
     * Get function that should be invoked on read event.
     *
     * @return callable
     */
    protected function getHandleReadFunction()
    {
        return [ $this, 'handleRead' ];
    }

    /**
     * Handle close.
     *
     * @internal
     */
    public function handleClose()
    {
        $this->pause();

        parent::handleClose();
    }
}