PEAR2/Net/Transmitter/StreamTransmitter.php

Go to the documentation of this file.

<?php

namespace PEAR2\Net\Transmitter;

class StreamTransmitter
{
    const DIRECTION_BOTH = '|||';
    const DIRECTION_SEND = '<<<';
    const DIRECTION_RECEIVE = '>>>';

    protected $stream;

    protected $persist;

    public function __construct($stream)
    {
        if (!self::isStream($stream)) {
            throw $this->createException('Invalid stream supplied.', 1);
        }
        $this->stream = $stream;
        $this->persist = (bool) preg_match(
            '#\s?persistent\s?#sm', get_resource_type($stream)
        );
    }

    public static function isStream($var)
    {
        return is_resource($var)
            && (bool) preg_match('#\s?stream$#sm', get_resource_type($var));
    }

    public function isFresh()
    {
        return ftell($this->stream) === 0;
    }
    
    public function setTimeout($seconds, $microseconds = 0)
    {
        return stream_set_timeout($this->stream, $seconds, $microseconds);
    }
    
    public function setBuffer($size, $direction = self::DIRECTION_BOTH)
    {
        switch($direction) {
        case self::DIRECTION_SEND:
            return stream_set_write_buffer($this->stream, $size) === 0;
        case self::DIRECTION_RECEIVE:
            return stream_set_read_buffer($this->stream, $size) === 0;
        default:
            return stream_set_write_buffer($this->stream, $size) === 0
                && stream_set_read_buffer($this->stream, $size) === 0;
        }
    }

    public function send($string)
    {
        $bytes = 0;
        $bytesToSend = (double) sprintf('%u', strlen($string));
        while ($bytes < $bytesToSend) {
            if ($this->isAcceptingData()) {
                $bytesNow = @fwrite(
                    $this->stream, substr($string, $bytes, 0xFFFFF)
                );
                if (0 != $bytesNow) {
                    $bytes += $bytesNow;
                } else {
                    throw $this->createException(
                        'Failed while sending string.', 2
                    );
                }
            }
        }
        return $bytes;
    }

    public function sendStream($stream)
    {
        $bytes = 0;
        while (!feof($stream)) {
            if ($this->isAcceptingData()) {
                $bytesNow = @stream_copy_to_stream(
                    $stream, $this->stream, 0xFFFFF
                );
                if (0 != $bytesNow) {
                    $bytes += $bytesNow;
                } else {
                    throw $this->createException(
                        'Failed while sending stream.', 3
                    );
                }
            }
        }
        fseek($stream, -$bytes, SEEK_CUR);
        return $bytes;
    }

    public function receive($length, $what = 'data')
    {
        $result = '';
        while ($length > 0) {
            if ($this->isAvailable()) {
                while ($this->isDataAwaiting()) {
                    $fragment = fread($this->stream, min($length, 0xFFFFF));
                    if ('' !== $fragment) {
                        $length -= strlen($fragment);
                        $result .= $fragment;
                        continue 2;
                    }
                }
            }
            throw $this->createException(
                "Failed while receiving {$what}", 4
            );
        }
        return $result;
    }

    public function receiveStream(
        $length, array $filters = array(), $what = 'stream data'
    ) {
        $result = fopen('php://temp', 'r+b');
        $appliedFilters = array();
        foreach ($filters as $filtername => $params) {
            $appliedFilters[] = stream_filter_append(
                $result, $filtername, STREAM_FILTER_WRITE, $params
            );
        }
        
        while ($length > 0) {
            if ($this->isAvailable()) {
                while ($this->isDataAwaiting()) {
                    $fragment = fread($this->stream, min($length, 0xFFFFF));
                    if ('' !== $fragment) {
                        $length -= strlen($fragment);
                        fwrite($result, $fragment);
                        continue 2;
                    }
                }
            }
            throw $this->createException(
                "Failed while receiving {$what}", 5
            );
        }
        
        foreach ($appliedFilters as $filter) {
            stream_filter_remove($filter);
        }
        rewind($result);
        return $result;
    }

    public function isAvailable()
    {
        return self::isStream($this->stream) && !feof($this->stream);
    }

    public function isDataAwaiting()
    {
        return $this->isAvailable();
    }

    public function isAcceptingData()
    {
        $r = $e = null;
        $w = array($this->stream);
        return self::isStream($this->stream)
            && 1 === @/* due to PHP bug #54563 */stream_select($r, $w, $e, 0);
    }

    public function __destruct()
    {
        if (!$this->persist) {
            $this->close();
        }
    }

    public function close()
    {
        return self::isStream($this->stream) && fclose($this->stream);
    }

    protected function createException($message, $code = 0)
    {
        return new \Exception($message, $code);
    }

}