Cassandra\Connection\Stream

Complexity

Total Complexity

21

Size/Duplication

Total Lines	129
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	1

Importance

Changes	5
Bugs	0	Features	3

7 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	5	1
A	_connect()	0	13	4
A	getOptions()	0	3	1
B	read()	0	21	5
A	readOnce()	0	14	4
B	write()	0	17	5
A	close()	0	3	1

<?php
namespace Cassandra\Connection;

class Stream {

	/**
	 * @var resource
	 */
	protected $_stream;

	/**
	 * @var array
	 */
	protected $_options = [
		'host'		=> null,
		'port'		=> 9042,
		'username'	=> null,
		'password'	=> null,
		'timeout'	=> 30,
		'connectTimeout'=>5,
		'persistent'=> false,
	];

	/**
	 * @param array $options
	 */
	public function __construct(array $options) {
		$this->_options = array_merge($this->_options, $options);

		$this->_connect();
	}

	/**
	 *
	 * @throws StreamException
	 * @return resource
	 */
	protected function _connect() {
		if (!empty($this->_stream)) return $this->_stream;

		$this->_stream = $this->_options['persistent']
			? pfsockopen($this->_options['host'], $this->_options['port'], $errorCode, $errorMessage, $this->_options['connectTimeout'])
			: fsockopen($this->_options['host'], $this->_options['port'], $errorCode, $errorMessage, $this->_options['connectTimeout']);

		if ($this->_stream === false){
			throw new StreamException($errorMessage, $errorCode);
		}

		stream_set_timeout($this->_stream,$this->_options['timeout']);
	}

	/**
	 * @return array
	 */
	public function getOptions() {
		return $this->_options;
	}

	/**
	 * @param $length
	 * @throws StreamException
	 * @return string
	 */
	public function read($length) {
		$data = '';
		do{
			$readData = fread($this->_stream, $length);

			if (feof($this->_stream))
				throw new StreamException('Connection reset by peer');

			if (stream_get_meta_data($this->_stream)['timed_out'])
				throw new StreamException('Connection timed out');

			if (strlen($readData) == 0)
				throw new StreamException("Unknown error");

			$data .= $readData;
			$length -= strlen($readData);
		}
		while($length > 0);
		
		return $data;
	}

	/**
	 * @param $length
	 * @throws StreamException
	 * @return string
	 */
	public function readOnce($length){
		$readData = fread($this->_stream, $length);

		if (feof($this->_stream))
			throw new StreamException('Connection reset by peer');

		if (stream_get_meta_data($this->_stream)['timed_out'])
			throw new StreamException('Connection timed out');

		if (strlen($readData) == 0)
			throw new StreamException("Unknown error");

		return $readData;
	}

	/**
	 *
	 * @param string $binary
	 * @throws StreamException
	 */
	public function write($binary){
		do{
			$sentBytes = fwrite($this->_stream, $binary);
			
			if (feof($this->_stream))
				throw new StreamException('Connection reset by peer');
			
			if (stream_get_meta_data($this->_stream)['timed_out'])
				throw new StreamException('Connection timed out');
			
			if ($sentBytes == 0)
				throw new StreamException("Uknown error");
			
			$binary = substr($binary, $sentBytes);
		}
		while(!empty($binary));
	}

	public function close(){
		 fclose($this->_stream);
	}
}