123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299 |
- <?php
- namespace Predis\Connection;
- use Predis\NotSupportedException;
- use Predis\ResponseError;
- use Predis\ResponseQueued;
- use Predis\Command\CommandInterface;
- use Predis\Iterator\MultiBulkResponseSimple;
- class StreamConnection extends AbstractConnection
- {
- private $mbiterable;
-
- public function __construct(ConnectionParametersInterface $parameters)
- {
- $this->mbiterable = (bool) $parameters->iterable_multibulk;
- parent::__construct($parameters);
- }
-
- public function __destruct()
- {
- if (!$this->parameters->persistent) {
- $this->disconnect();
- }
- }
-
- protected function createResource()
- {
- $parameters = $this->parameters;
- $initializer = "{$parameters->scheme}StreamInitializer";
- return $this->$initializer($parameters);
- }
-
- private function tcpStreamInitializer(ConnectionParametersInterface $parameters)
- {
- $uri = "tcp://{$parameters->host}:{$parameters->port}/";
- $flags = STREAM_CLIENT_CONNECT;
- if (isset($parameters->async_connect) && $parameters->async_connect === true) {
- $flags |= STREAM_CLIENT_ASYNC_CONNECT;
- }
- if (isset($parameters->persistent) && $parameters->persistent === true) {
- $flags |= STREAM_CLIENT_PERSISTENT;
- }
- $resource = @stream_socket_client($uri, $errno, $errstr, $parameters->timeout, $flags);
- if (!$resource) {
- $this->onConnectionError(trim($errstr), $errno);
- }
- if (isset($parameters->read_write_timeout)) {
- $rwtimeout = $parameters->read_write_timeout;
- $rwtimeout = $rwtimeout > 0 ? $rwtimeout : -1;
- $timeoutSeconds = floor($rwtimeout);
- $timeoutUSeconds = ($rwtimeout - $timeoutSeconds) * 1000000;
- stream_set_timeout($resource, $timeoutSeconds, $timeoutUSeconds);
- }
- return $resource;
- }
-
- private function unixStreamInitializer(ConnectionParametersInterface $parameters)
- {
- $uri = "unix://{$parameters->path}";
- $flags = STREAM_CLIENT_CONNECT;
- if ($parameters->persistent === true) {
- $flags |= STREAM_CLIENT_PERSISTENT;
- }
- $resource = @stream_socket_client($uri, $errno, $errstr, $parameters->timeout, $flags);
- if (!$resource) {
- $this->onConnectionError(trim($errstr), $errno);
- }
- return $resource;
- }
-
- public function connect()
- {
- parent::connect();
- if (count($this->initCmds) > 0){
- $this->sendInitializationCommands();
- }
- }
-
- public function disconnect()
- {
- if ($this->isConnected()) {
- fclose($this->getResource());
- parent::disconnect();
- }
- }
-
- private function sendInitializationCommands()
- {
- foreach ($this->initCmds as $command) {
- $this->writeCommand($command);
- }
- foreach ($this->initCmds as $command) {
- $this->readResponse($command);
- }
- }
-
- protected function writeBytes($buffer)
- {
- $socket = $this->getResource();
- while (($length = strlen($buffer)) > 0) {
- $written = fwrite($socket, $buffer);
- if ($length === $written) {
- return;
- }
- if ($written === false || $written === 0) {
- $this->onConnectionError('Error while writing bytes to the server');
- }
- $buffer = substr($buffer, $written);
- }
- }
-
- public function read()
- {
- $socket = $this->getResource();
- $chunk = fgets($socket);
- if ($chunk === false || $chunk === '') {
- $this->onConnectionError('Error while reading line from the server');
- }
- $prefix = $chunk[0];
- $payload = substr($chunk, 1, -2);
- switch ($prefix) {
- case '+':
- switch ($payload) {
- case 'OK':
- return true;
- case 'QUEUED':
- return new ResponseQueued();
- default:
- return $payload;
- }
- case '$':
- $size = (int) $payload;
- if ($size === -1) {
- return null;
- }
- $bulkData = '';
- $bytesLeft = ($size += 2);
- do {
- $chunk = fread($socket, min($bytesLeft, 4096));
- if ($chunk === false || $chunk === '') {
- $this->onConnectionError('Error while reading bytes from the server');
- }
- $bulkData .= $chunk;
- $bytesLeft = $size - strlen($bulkData);
- } while ($bytesLeft > 0);
- return substr($bulkData, 0, -2);
- case '*':
- $count = (int) $payload;
- if ($count === -1) {
- return null;
- }
- if ($this->mbiterable === true) {
- return new MultiBulkResponseSimple($this, $count);
- }
- $multibulk = array();
- for ($i = 0; $i < $count; $i++) {
- $multibulk[$i] = $this->read();
- }
- return $multibulk;
- case ':':
- return (int) $payload;
- case '-':
- return new ResponseError($payload);
- default:
- $this->onProtocolError("Unknown prefix: '$prefix'");
- }
- }
-
- public function writeCommand(CommandInterface $command)
- {
- $commandId = $command->getId();
- $arguments = $command->getArguments();
- $cmdlen = strlen($commandId);
- $reqlen = count($arguments) + 1;
- $buffer = "*{$reqlen}\r\n\${$cmdlen}\r\n{$commandId}\r\n";
- for ($i = 0; $i < $reqlen - 1; $i++) {
- $argument = $arguments[$i];
- $arglen = strlen($argument);
- $buffer .= "\${$arglen}\r\n{$argument}\r\n";
- }
- $this->writeBytes($buffer);
- }
-
- public function __sleep()
- {
- return array_merge(parent::__sleep(), array('mbiterable'));
- }
- }
|