Socket::connect() - Code Metrics - mariano/disque-php - Measure and Improve Code Quality continuously with Scrutinizer

Socket::connect() A
last analyzed 2017-05-30 02:39 UTC

↳ Parent: Socket

Complexity

Conditions	3
Paths	3

Size

Total Lines	18
Code Lines	11

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	12

Importance

Changes

Metric	Value
dl	0
loc	18
ccs	0
cts	16
cp	0
rs	9.4285
c	0
b	0
f	0
cc	3
eloc	11
nc	3
nop	2
crap	12

<?php
namespace Disque\Connection;

use Exception;
use Disque\Command\CommandInterface;
use Disque\Connection\Response\ResponseException;
use Disque\Connection\Response;

/**
 * This class is greatly inspired by `Predis\Connection\StreamConnection`,
 * which is part of [predis](https://github.com/nrk/predis) and was developed
 * by Daniele Alessandri <[email protected]>. All credits go to him where
 * relevant.
 */
class Socket extends BaseConnection implements ConnectionInterface
{
    /**
     * Socket handle
     *
     * @var resource
     */
    protected $socket;

    /**
     * Response handlers
     *
     * The characters used as keys are part of the Redis/Disque protocol.
     * Disque uses the same response protocol as Redis, therefore
     * @see http://redis.io/topics/protocol
     *
     * @var array
     */
    private $responseHandlers = [
        '+' => Response\StringResponse::class,
        '-' => Response\ErrorResponse::class,
        ':' => Response\IntResponse::class,
        '$' => Response\TextResponse::class,
        '*' => Response\ArrayResponse::class
    ];

    /**
     * @inheritdoc
     */
    public function connect($connectionTimeout = 0, $responseTimeout = null)
    {
        parent::connect($connectionTimeout, $responseTimeout);

        $this->socket = $this->getSocket(
            $this->host,
            $this->port,
            (float) $connectionTimeout
        );
        if (!is_resource($this->socket)) {
            throw new ConnectionException("Could not connect to {$this->host}:{$this->port}");
        }

        stream_set_blocking($this->socket, 1);
        if (!is_null($responseTimeout)) {
            stream_set_timeout($this->socket, $responseTimeout);
        }
    }

    /**
     * @inheritdoc
     */
    public function disconnect()
    {
        if (!$this->isConnected()) {
            return;
        }
        fclose($this->socket);
        $this->socket = null;
    }

    /**
     * @inheritdoc
     */
    public function isConnected()
    {
        return (isset($this->socket) && is_resource($this->socket));
    }

    /**
     * @inheritdoc
     */
    public function execute(CommandInterface $command)
    {
        $commandName = $command->getCommand();
        $arguments = $command->getArguments();
        $totalArguments = count($arguments);

        $parts = [
            '*' . ($totalArguments + 1),
            '$' . strlen($commandName),
            $commandName
        ];

        for ($i=0; $i < $totalArguments; $i++) {
            $argument = $arguments[$i];
            $parts[] = '$' . strlen($argument);
            $parts[] = $argument;
        }

        $this->send(implode("\r\n", $parts)."\r\n");
        return $this->receive($command->isBlocking());
    }

    /**
     * Execute a command on the connection
     *
     * @param string $data Data to send
     * @throws ConnectionException
     */
    public function send($data)
    {
        $this->shouldBeConnected();

        do {
            $length = strlen($data);
            $bytes = fwrite($this->socket, $data);
            if (empty($bytes)) {
                throw new ConnectionException("Could not write {$length} bytes to client");
            } elseif ($bytes === $length) {
                break;
            }

            $data = substr($data, $bytes);
        } while ($length > 0);
    }

    /**
     * Read data from connection
     *
     * @param bool $keepWaiting If `true`, timeouts on stream read will be ignored
     * @return mixed Data received
     *
     * @throws ConnectionException
     * @throws ResponseException
     */
    public function receive($keepWaiting = false)
    {
        $this->shouldBeConnected();

        $type = $this->getType($keepWaiting);
        if (!array_key_exists($type, $this->responseHandlers)) {
            throw new ResponseException("Don't know how to handle a response of type {$type}");
        }

        $responseHandlerClass = $this->responseHandlers[$type];
        $responseHandler = new $responseHandlerClass($this->getData());
        $responseHandler->setReader(function ($bytes) {
            return fread($this->socket, $bytes);
        });
        $responseHandler->setReceiver(function () use ($keepWaiting) {
            return $this->receive($keepWaiting);
        });
        $response = $responseHandler->parse();

        /**
         * If Disque returned an error, raise it in form of an exception
         * @see Disque\Connection\Response\ErrorResponse::parse()
         */
        if ($response instanceof ResponseException) {
            throw $response;
        }

        return $response;
    }

    /**
     * Build actual socket
     *
     * @param string $host Host
     * @param int $port Port
     * @param float $timeout Timeout
     * @return resource Socket
     */
    protected function getSocket($host, $port, $timeout)
    {
        return stream_socket_client("tcp://{$host}:{$port}", $error, $message, $timeout, STREAM_CLIENT_CONNECT | STREAM_CLIENT_PERSISTENT);
    }

    /**
     * Get the first byte from Disque, which contains the data type
     *
     * @param bool $keepWaiting If `true`, timeouts on stream read will be ignored
     * @return string A single char
     * @throws ConnectionException
     */
    private function getType($keepWaiting = false)
    {
        $type = null;
        while (!feof($this->socket)) {
            $type = fgetc($this->socket);
            if ($type !== false && $type !== '') {
                break;
            }

            $info = stream_get_meta_data($this->socket);
            if (!$keepWaiting || !$info['timed_out']) {
                break;
            }
        }

        if ($type === false || $type === '') {
            throw new ConnectionException('Nothing received while reading from client');
        }

        return $type;
    }

    /**
     * Get a line of data
     *
     * @return string Line of data
     * @throws ConnectionException
     */
    private function getData()
    {
        $data = fgets($this->socket);
        if ($data === false || $data === '') {
            throw new ConnectionException('Nothing received while reading from client');
        }
        return $data;
    }

    /**
     * We should be connected
     *
     * @return void
     * @throws ConnectionException
     */
    private function shouldBeConnected()
    {
        if (!$this->isConnected()) {
            throw new ConnectionException('No connection established');
        }
    }
}

1			<?php
2			namespace Disque\Connection;
3
4			use Exception;
5			use Disque\Command\CommandInterface;
6			use Disque\Connection\Response\ResponseException;
7			use Disque\Connection\Response;
8
9			/**
10			* This class is greatly inspired by `Predis\Connection\StreamConnection`,
11			* which is part of [predis](https://github.com/nrk/predis) and was developed
12			* by Daniele Alessandri <[email protected]>. All credits go to him where
13			* relevant.
14			*/
15			class Socket extends BaseConnection implements ConnectionInterface
16			{
17			/**
18			* Socket handle
19			*
20			* @var resource
21			*/
22			protected $socket;
23
24			/**
25			* Response handlers
26			*
27			* The characters used as keys are part of the Redis/Disque protocol.
28			* Disque uses the same response protocol as Redis, therefore
29			* @see http://redis.io/topics/protocol
30			*
31			* @var array
32			*/
33			private $responseHandlers = [
34			'+' => Response\StringResponse::class,
35			'-' => Response\ErrorResponse::class,
36			':' => Response\IntResponse::class,
37			'$' => Response\TextResponse::class,
38			'*' => Response\ArrayResponse::class
39			];
40
41			/**
42			* @inheritdoc
43			*/
44			public function connect($connectionTimeout = 0, $responseTimeout = null)
45			{
46			parent::connect($connectionTimeout, $responseTimeout);
47
48			$this->socket = $this->getSocket(
49			$this->host,
50			$this->port,
51			(float) $connectionTimeout
52			);
53			if (!is_resource($this->socket)) {
54			throw new ConnectionException("Could not connect to {$this->host}:{$this->port}");
55			}
56
57			stream_set_blocking($this->socket, 1);
58			if (!is_null($responseTimeout)) {
59			stream_set_timeout($this->socket, $responseTimeout);
60			}
61			}
62
63			/**
64			* @inheritdoc
65			*/
66			public function disconnect()
67			{
68			if (!$this->isConnected()) {
69			return;
70			}
71			fclose($this->socket);
72			$this->socket = null;
73			}
74
75			/**
76			* @inheritdoc
77			*/
78			public function isConnected()
79			{
80			return (isset($this->socket) && is_resource($this->socket));
81			}
82
83			/**
84			* @inheritdoc
85			*/
86			public function execute(CommandInterface $command)
87			{
88			$commandName = $command->getCommand();
89			$arguments = $command->getArguments();
90			$totalArguments = count($arguments);
91
92			$parts = [
93			'*' . ($totalArguments + 1),
94			'$' . strlen($commandName),
95			$commandName
96			];
97
98			for ($i=0; $i < $totalArguments; $i++) {
99			$argument = $arguments[$i];
100			$parts[] = '$' . strlen($argument);
101			$parts[] = $argument;
102			}
103
104			$this->send(implode("\r\n", $parts)."\r\n");
105			return $this->receive($command->isBlocking());
106			}
107
108			/**
109			* Execute a command on the connection
110			*
111			* @param string $data Data to send
112			* @throws ConnectionException
113			*/
114			public function send($data)
115			{
116			$this->shouldBeConnected();
117
118			do {
119			$length = strlen($data);
120			$bytes = fwrite($this->socket, $data);
121			if (empty($bytes)) {
122			throw new ConnectionException("Could not write {$length} bytes to client");
123			} elseif ($bytes === $length) {
124			break;
125			}
126
127			$data = substr($data, $bytes);
128			} while ($length > 0);
129			}
130
131			/**
132			* Read data from connection
133			*
134			* @param bool $keepWaiting If `true`, timeouts on stream read will be ignored
135			* @return mixed Data received
136			*
137			* @throws ConnectionException
138			* @throws ResponseException
139			*/
140			public function receive($keepWaiting = false)
141			{
142			$this->shouldBeConnected();
143
144			$type = $this->getType($keepWaiting);
145			if (!array_key_exists($type, $this->responseHandlers)) {
146			throw new ResponseException("Don't know how to handle a response of type {$type}");
147			}
148
149			$responseHandlerClass = $this->responseHandlers[$type];
150			$responseHandler = new $responseHandlerClass($this->getData());
151			$responseHandler->setReader(function ($bytes) {
152			return fread($this->socket, $bytes);
153			});
154			$responseHandler->setReceiver(function () use ($keepWaiting) {
155			return $this->receive($keepWaiting);
156			});
157			$response = $responseHandler->parse();
158
159			/**
160			* If Disque returned an error, raise it in form of an exception
161			* @see Disque\Connection\Response\ErrorResponse::parse()
162			*/
163			if ($response instanceof ResponseException) {
164			throw $response;
165			}
166
167			return $response;
168			}
169
170			/**
171			* Build actual socket
172			*
173			* @param string $host Host
174			* @param int $port Port
175			* @param float $timeout Timeout
176			* @return resource Socket
177			*/
178			protected function getSocket($host, $port, $timeout)
179			{
180			return stream_socket_client("tcp://{$host}:{$port}", $error, $message, $timeout, STREAM_CLIENT_CONNECT \| STREAM_CLIENT_PERSISTENT);
181			}
182
183			/**
184			* Get the first byte from Disque, which contains the data type
185			*
186			* @param bool $keepWaiting If `true`, timeouts on stream read will be ignored
187			* @return string A single char
188			* @throws ConnectionException
189			*/
190			private function getType($keepWaiting = false)
191			{
192			$type = null;
193			while (!feof($this->socket)) {
194			$type = fgetc($this->socket);
195			if ($type !== false && $type !== '') {
196			break;
197			}
198
199			$info = stream_get_meta_data($this->socket);
200			if (!$keepWaiting \|\| !$info['timed_out']) {
201			break;
202			}
203			}
204
205			if ($type === false \|\| $type === '') {
206			throw new ConnectionException('Nothing received while reading from client');
207			}
208
209			return $type;
210			}
211
212			/**
213			* Get a line of data
214			*
215			* @return string Line of data
216			* @throws ConnectionException
217			*/
218			private function getData()
219			{
220			$data = fgets($this->socket);
221			if ($data === false \|\| $data === '') {
222			throw new ConnectionException('Nothing received while reading from client');
223			}
224			return $data;
225			}
226
227			/**
228			* We should be connected
229			*
230			* @return void
231			* @throws ConnectionException
232			*/
233			private function shouldBeConnected()
234			{
235			if (!$this->isConnected()) {
236			throw new ConnectionException('No connection established');
237			}
238			}
239			}

mariano / disque-php

Socket::connect() A last analyzed 2017-05-30 02:39 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

Socket::connect() A
last analyzed 2017-05-30 02:39 UTC