Dazzle\SSH\Driver\Sftp\SftpResource

Complexity

Total Complexity

49

Size/Duplication

Total Lines	319
Duplicated Lines	3.13 %

Coupling/Cohesion

Components	1
Dependencies	6

Test Coverage

Coverage

0%

Importance

Changes

0

14 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	11	1
A	getId()	0	4	1
A	isPaused()	0	4	1
A	setBufferSize()	0	4	1
A	getBufferSize()	0	4	1
A	pause()	0	9	2
B	resume()	0	17	8
A	write()	0	18	4
A	read()	0	17	4
A	close()	0	20	3
C	handleWrite()	0	47	9
C	handleRead()	10	38	7
A	handleClose()	0	6	1
B	writeEnd()	0	20	6

<?php

namespace Dazzle\SSH\Driver\Sftp;

use Dazzle\Loop\LoopAwareTrait;
use Dazzle\SSH\SSH2DriverInterface;
use Dazzle\SSH\SSH2ResourceInterface;
use Dazzle\Stream\Sync\Stream;
use Dazzle\Throwable\Exception\Runtime\ReadException;
use Dazzle\Throwable\Exception\Runtime\WriteException;
use Dazzle\Util\Buffer\Buffer;
use Dazzle\Util\Buffer\BufferInterface;
use Error;
use Exception;

class SftpResource extends Stream implements SSH2ResourceInterface

There is at least one abstract method in this class. Maybe declare it as abstract, or implement the remaining methods: copyEvent, copyEvents, delay, delayOnce, delayTimes, discardEvents, emit, findListener, flushListeners, forwardEvents, getMetadata, getMode, getResource, getResourceId, getStreamType, getWrapperType, isOpen, isReadable, isSeekable, isWritable, on, once, removeListener, removeListeners, rewind, seek, setMode, tell, times

{
    use LoopAwareTrait;

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

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

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

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

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

    /**
     * @param SSH2DriverInterface $driver
     * @param resource $resource
     */
    public function __construct(SSH2DriverInterface $driver, $resource)
    {
        parent::__construct($resource);

        $this->loop = $driver->getLoop();
        $this->writing = false;
        $this->reading = false;
        $this->readingStarted = false;
        $this->paused = false;
        $this->buffer = new Buffer();
    }

    /**
     * @return string
     */
    public function getId()
    {
        return (string) $this->getResourceId();
    }

    /**
     * @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->reading = false;
        }
    }

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

            if ($this->readable && $this->readingStarted)
            {
                $this->reading = true;
            }

            if ($this->writable && $this->buffer->isEmpty() === false)
            {
                $this->writing = true;
            }
        }
    }

    /**
     * @override
     * @inheritDoc
     */
    public function write($text = '')
    {
        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;
        }

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

    /**
     * @override
     * @inheritDoc
     */
    public function read($length = null)
    {
        if (!$this->readable)
        {
            return $this->throwAndEmitException(
                new ReadException('Stream is no longer readable.')
            );
        }

        if (!$this->reading && !$this->paused)
        {
            $this->reading = true;
            $this->readingStarted = true;
        }

        return '';
    }

    /**
     * @override
     * @inheritDoc
     */
    public function close()
    {
        if ($this->closing)
        {
            return;
        }

        $this->closing = true;
        $this->readable = false;
        $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()
    {
        try
        {
            if (!$this->writing)
            {
                return;
            }

            $text = $this->buffer->peek();

            if ($text !== '')
            {
                $sent = fwrite($this->resource, $text);
            }
            else
            {
                $sent = 0;
            }

            if ($text !== '' && !$sent)
            {
                $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);

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

            if ($lenAfter === 0 && $this->buffer->isEmpty())
            {
                $this->writing = false;
                $this->emit('finish', [ $this ]);
            }
        }
        catch (Error $ex)
        {
            $this->emit('error', [ $this, $ex ]);
        }
        catch (Exception $ex)
        {
            $this->emit('error', [ $this, $ex ]);
        }
    }

    /**
     * Handle the incoming stream.
     *
     * @internal
     */
    public function handleRead()
    {
        try
        {
            if (!$this->reading)
            {
                return;
            }

            $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 code seems to be duplicated across your project.

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

                if (strlen($ret) < $length)
                {
                    $this->reading = false;
                    $this->emit('end', [ $this ]);
                }
            }
        }
        catch (Error $ex)
        {
            $this->emit('error', [ $this, $ex ]);
        }
        catch (Exception $ex)
        {
            $this->emit('error', [ $this, $ex ]);
        }
    }

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

        parent::handleClose();
    }

    /**
     *
     */
    private function writeEnd()
    {
        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());
    }
}