Dazzle\Stream\StreamWriter

Complexity

Total Complexity

33

Size/Duplication

Total Lines	241
Duplicated Lines	44.4 %

Coupling/Cohesion

Components	1
Dependencies	5

Test Coverage

Coverage

75.56%

Importance

Changes

0

13 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()	5	12	4
A	write()	19	19	4
A	close()	20	20	3
B	handleWrite()	27	27	6
A	handleClose()	0	6	1
A	getHandleWriteFunction()	0	4	1
B	writeEnd()	20	20	6

<?php

namespace Dazzle\Stream;

use Dazzle\Throwable\Exception\Runtime\WriteException;
use Dazzle\Throwable\Exception\Logic\InvalidArgumentException;
use Dazzle\Loop\LoopAwareTrait;
use Dazzle\Loop\LoopInterface;
use Dazzle\Util\Buffer\Buffer;
use Dazzle\Util\Buffer\BufferInterface;
use Error;
use Exception;

class StreamWriter extends Sync\StreamWriter implements StreamWriterInterface
{
    use LoopAwareTrait;

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

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

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

    /**
     * @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_write_buffer'))
        {
            stream_set_write_buffer($this->resource, 0);
        }

        $this->loop = $loop;
        $this->writing = false;
        $this->paused = true;
        $this->buffer = new Buffer();

        $this->resume();
    }

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

        unset($this->loop);
        unset($this->writing);
        unset($this->paused);
        unset($this->buffer);
    }

    /**
     * @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->writing = false;
            $this->loop->removeWriteStream($this->resource);
        }
    }

    /**
     * @override
     * @inheritDoc
     */
    public function resume()
    {
        if ($this->writable && $this->paused)
        {
            $this->paused = false;
            if ($this->buffer->isEmpty() === false)

This code seems to be duplicated across your project.

            {
                $this->writing = true;
                $this->loop->addWriteStream($this->resource, $this->getHandleWriteFunction());
            }
        }
    }

    /**
     * @override
     * @inheritDoc
     */
    public function write($text = '')

This method seems to be duplicated in your project.

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

        $this->buffer->push($text);

        if (!$this->writing && !$this->paused)
        {
            $this->writing = true;
            $this->loop->addWriteStream($this->resource, $this->getHandleWriteFunction());
        }

        return $this->buffer->length() < $this->bufferSize;
    }

    /**
     * @override
     * @inheritDoc
     */
    public function close()

This method seems to be duplicated in your project.

    {
        if ($this->closing)
        {
            return;
        }

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

The property readable does not exist. Did you maybe forget to declare it?

        $this->writable = false;

        if ($this->buffer->isEmpty() === false)
        {
            $this->writeEnd();
        }

        $this->emit('close', [ $this ]);
        $this->handleClose();
        $this->emit('done', [ $this ]);
    }

    /**
     * Handle the outcoming stream.
     *
     * @internal
     */
    public function handleWrite()

This method seems to be duplicated in your project.

    {
        $text = $this->buffer->peek();
        $sent = fwrite($this->resource, $text, $this->bufferSize);

        if ($sent === false)
        {
            $this->emit('error', [ $this, new WriteException('Error occurred while writing to the stream resource.') ]);
            return;
        }

        $lenBefore = strlen($text);
        $lenAfter = $lenBefore - $sent;
        $this->buffer->remove($sent);

        if ($lenAfter > 0 && $lenBefore >= $this->bufferSize && $lenAfter < $this->bufferSize)
        {
            $this->emit('drain', [ $this ]);
        }
        else if ($lenAfter === 0)
        {
            $this->loop->removeWriteStream($this->resource);
            $this->writing = false;
            $this->emit('drain', [ $this ]);
            $this->emit('finish', [ $this ]);
        }
    }

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

        parent::handleClose();
    }

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

    /**
     *
     */
    private function writeEnd()

This method seems to be duplicated in your project.

    {
        do
        {
            try
            {
                $sent = @fwrite($this->resource, $this->buffer->peek());
                $this->buffer->remove($sent);
            }
            catch (Error $ex)
            {
                $sent = 0;
            }
            catch (Exception $ex)
            {
                $sent = 0;
            }
        }
        while (is_resource($this->resource) && $sent > 0 && !$this->buffer->isEmpty());
    }
}