@param \Riak\Client\Core\Transport\Proto\ProtoStream $socket
@param integer                                       $length

@return \Riak\Client\Core\Transport\Proto\ProtoStream
513645
private function receiveMessageBody(ProtoStream $socket, $length)
    {
        $readSize   = 0;
        $resource   = fopen('php://temp', 'r+');
        $bodyBuffer = new ProtoStream($resource);

        while ($readSize < $length) {
            $size     = min(8192, $length - $readSize);
            $part     = $socket->read($size);
            $readSize = $readSize + $size;

            if ($part === false) {
                throw new RiakTransportException('Fail to read socket response');
            }

            $bodyBuffer->write($part);
        }

        $bodyBuffer->seek(0);

        return $bodyBuffer;
    }
1063748
protected function createStream($socket, RequestInterface $request, ResponseInterface $response): Stream
    {
        $size = null;

        if ($response->hasHeader('Content-Length')) {
            $size = (int) $response->getHeaderLine('Content-Length');
        }

        return new Stream($request, $socket, $size);
    }
1202404
protected function getStream($socket)
    {
        if (class_exists(DuplexResourceStream::class)) {
            return new DuplexResourceStream($socket, $this->getLoop());
        }

        return new Stream($socket, $this->getLoop());
    }
817629
private static function _createSocket($proto)
    {
        $sock = ($proto === 'tcp') ?
            socket_create(AF_INET, SOCK_STREAM, SOL_TCP):
            socket_create(AF_INET, SOCK_DGRAM, SOL_UDP)
        ;
        if (!$sock) {
            throw new \UnexpectedValueException("socket_create() failed: " . socket_strerror(socket_last_error()));
        }
        return $sock;
    }
116694
public function read($length = 4096)
  {
    $this->connect();
    $this->waitForReadyRead();

    $data = @stream_get_contents($this->stream, $length);

    Signal::getInstance()->emit(strtolower($this->getAdapterType()) . "DataRead", $data);

    if($data === FALSE)
    {
      throw new Exception("connection to server '" . $this->config["host"] . ":" . $this->config["port"] . "' lost");
    }

    return new Str($data);
  }
727853
public static function read($socket)
    {
        $header = '';
        do {
            $buf = socket_read($socket, 12 - self::stringLength($header));
            $header .= $buf;
        } while ($buf !== false &&
                 $buf !== '' && self::stringLength($header) < 12);

        if ($buf === '') {
            throw new Exception('Connection was reset');
        }

        if (self::stringLength($header) == 0) {
            return array();
        }
        $resp = @unpack('a4magic/Ntype/Nlen', $header);

        if (!count($resp) == 3) {
            throw new Exception('Received an invalid response');
        }

        if (!isset(self::$magic[$resp['type']])) {
            throw new Exception(
                'Invalid response magic returned: ' . $resp['type']
            );
        }

        $return = array();
        if ($resp['len'] > 0) {
            $data = '';
            while (self::stringLength($data) < $resp['len']) {
                $data .= socket_read($socket, $resp['len'] - self::stringLength($data));
            }

            $d = explode("\x00", $data);
            foreach (self::$magic[$resp['type']][1] as $i => $a) {
                $return[$a] = $d[$i];
            }
        }

        $function = self::$magic[$resp['type']][0];
        if ($function == 'error') {
            if (!self::stringLength($return['err_text'])) {
                $return['err_text'] = 'Unknown error; see error code.';
            }

            throw new Exception(
                $return['err_text'], $return['err_code']
            );
        }

        return array('function' => self::$magic[$resp['type']][0],
                     'type' => $resp['type'],
                     'data' => $return, );
    }
856666
public function slice(int $offset, int $length = null):ByteArray{

		// keep an extended class
		/** @var \chillerlan\Traits\ArrayHelpers\ByteArray $slice */
		$slice  = (new ReflectionClass($this))->newInstanceArgs([$length ?? ($this->count() - $offset)]);

		foreach($slice as $i => $_){
			$slice[$i] = $this[$offset + $i];
		}

		return $slice;
	}
1235598
public function write($buffer, $length = 0)
    {
        if ("\n" != substr($buffer, -1)) {
            $buffer .= "\n";
        }
        if (!$length) {
            $length = strlen($buffer);
        }

        return socket_write($this->socket, $buffer, $length);
    }
1155079
public function write($data, $length = self::DEFAULT_LENGTH, $wait = self::DEFAULT_WAIT)
		{
			$bytes = dio_write($this->serial, $data);
			usleep($wait);
			return $bytes;
		}
86059
public function substr($start, $length=null)
    {
        return $this->newInstance(mb_substr($this->toString(), $start, $length, $this->getEncoding()));
    }
682504
public function readStream($length)
    {
        $stream  = self::fromString();
        $target  = $stream->stream;
        $source  = $this->stream;

        if ($length < 1) {
            return $stream;
        }

        $written = stream_copy_to_stream($source, $target, $length);

        if ($written !== $length) {
            throw new RuntimeException('Failed to read stream with ' . $length . ' bytes');
        }

        $stream->seek(0);

        return $stream;
    }
334009
private function forceRead($length)
    {
        $read = '';

        do {
            $read .= $this->stream->read($length - \strlen($read));
        } while (\strlen($read) < $length && !$this->stream->eof());

        return $read;
    }
749728
private function receiveMessageCode(ProtoStream $socket)
    {
        $codeBin = $socket->read(1);

        if ($codeBin === false) {
            throw new RiakTransportException('Fail to read response code');
        }

        $unpack = unpack("C", $codeBin);
        $values = array_values($unpack);
        $code   = isset($values[0]) ? $values[0] : 0;

        return $code;
    }
1033492
public function remove($startIndex, $length)
    {
        $start = UTF8::substr
        (
            (string)$this->str,
            0,
            $startIndex,
            $this->str->getEncoding()
        );

        $end = UTF8::substr
        (
            (string)$this->str,
            $startIndex + $length,
            $this->str->getLength(),
            $this->str->getEncoding()
        );

        $this->str = new Str($start.$end);

        return $this;
    }
288073
public function getWriteBuffer(int $length, string $append = ''): Promise
    {
        if (strlen($append)) {
            $this->append = $append;
            $this->append_after = $length - strlen($append);
        }

        return new \Amp\Success($this);
    }
167127
private static function createSocket($host, $port, $sendTimeout, $receiveTimeout)
    {
        $socket = new TSocket($host, $port, true);
        if (null !== $sendTimeout) {
            $socket->setSendTimeout($sendTimeout);
        }

        if (null !== $receiveTimeout) {
            $socket->setRecvTimeout($receiveTimeout);
        }

        return $socket;
    }
1305430
private function newLimitStream(StreamInterface $stream, $length, $start)
    {
        return new SeekingLimitStream(
            $this->newNonClosingStream($stream),
            $length,
            $start
        );
    }