Native::pull() - Code Metrics - Logi-CE/fmup - Measure and Improve Code Quality continuously with Scrutinizer

Native::pull() B
last analyzed 2017-12-05 10:44 UTC

↳ Parent: Native

Complexity

Conditions	6
Paths	12

Size

Total Lines	27
Code Lines	22

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	22
CRAP Score	6

Importance

Changes

Metric	Value
cc	6
eloc	22
nc	12
nop	2
dl	0
loc	27
ccs	22
cts	22
cp	1
crap	6
rs	8.439
c	0
b	0
f	0

<?php
namespace FMUP\Queue\Driver;

use FMUP\Environment;
use FMUP\Queue\Channel;
use FMUP\Queue\DriverInterface;
use FMUP\Queue\Exception;
use FMUP\Queue\Message;

class Native implements DriverInterface, Environment\OptionalInterface
{
    use Environment\OptionalTrait;

    const CONFIGURATION_PERM_UID = 'msg_perm.uid';
    const CONFIGURATION_PERM_GID = 'msg_perm.gid';
    const CONFIGURATION_PERM_MODE = 'msg_perm.mode';
    const CONFIGURATION_SENT_TIME = 'msg_stime';
    const CONFIGURATION_RECEIVED_TIME = 'msg_rtime';
    const CONFIGURATION_UPDATE_TIME = 'msg_ctime';
    const CONFIGURATION_MESSAGE_NUMBER = 'msg_qnum';
    const CONFIGURATION_MESSAGE_SIZE = 'msg_qbytes';
    const CONFIGURATION_SENDER_PID = 'msg_lspid';
    const CONFIGURATION_RECEIVER_PID = 'msg_lrpid';

    /**
     * Connect to specified queue
     * @param Channel $channel
     * @return Channel
     * @throws Exception
     */
    public function connect(Channel $channel)
    {
        if (!$channel->hasResource()) {
            $name = $this->secureQueueName($channel->getName());
            $channel->setName($name);
            $resource = $this->msgGetQueue($name);
            $channel->setResource($resource);
        }
        return $channel;
    }

    /**
     * @param string $name
     * @return resource
     * @codeCoverageIgnore
     */
    protected function msgGetQueue($name)
    {
        return msg_get_queue($name);
    }

    /**
     * Check if queue exists
     * @param Channel $channel
     * @return bool
     * @throws Exception
     */
    public function exists(Channel $channel)
    {
        $name = (!$channel->hasResource()) ? $this->secureQueueName($channel->getName()) : $channel->getName();
        return $this->msgQueueExists($name);
    }

    /**
     * @param $name
     * @return bool
     * @codeCoverageIgnore
     */
    protected function msgQueueExists($name)
    {
        return msg_queue_exists($name);
    }

    /**
     * Get a message from queue
     *
     * @param Channel $channel
     * @param string $messageType
     * @return mixed|null
     * @throws Exception
     */
    public function pull(Channel $channel, $messageType = null)
    {
        if (!$channel->hasResource()) {
            $this->connect($channel);
        }
        $messageType = $this->secureMessageType($messageType);
        $receivedMessageType = 0;
        $message = null;
        $error = 0;
        $messageSize = $this->getMessageSize($channel);
        $success = $this->msgReceive(
            $channel->getResource(),
            $messageType,
            $receivedMessageType,
            $messageSize,
            $message,
            $channel->getSettings()->getSerialize(),
            $this->getReceiveFlags($channel),
            $error
        );
        $isNonBlockReceive = (MSG_IPC_NOWAIT & $this->getParamBlockReceive($channel));
        $isNonBlockingPlusNoMessage = $isNonBlockReceive && ($error === MSG_ENOMSG);
        if (!$success && !$isNonBlockingPlusNoMessage) {
            throw new Exception("Error while receiving message", $error);
        }
        return $message ? (new Message())->setOriginal($message)->setTranslated($message) : null;
    }

    /**
     * @param $queue
     * @param $desiredMsgType
     * @param $msgType
     * @param $maxsize
     * @param $message
     * @param bool|true $unSerialize
     * @param int $flags
     * @param null $errorCode
     * @return bool
     * @codeCoverageIgnore
     */
    protected function msgReceive(
        $queue,
        $desiredMsgType,
        &$msgType,
        $maxsize,
        &$message,
        $unSerialize = true,
        $flags = 0,
        &$errorCode = null
    ) {
        return msg_receive($queue, $desiredMsgType, $msgType, $maxsize, $message, $unSerialize, $flags, $errorCode);
    }

    /**
     * Retrieve message maximum size for a given queue
     * @param Channel $channel
     * @return int
     */
    private function getMessageSize(Channel $channel)
    {
        $messageSize = $channel->getSettings()->getMaxMessageSize();
        if (!$messageSize) {
            $configuration = $this->getConfiguration($channel->getResource());
            $messageSize = (int)$configuration[self::CONFIGURATION_MESSAGE_SIZE];
            $channel->getSettings()->setMaxMessageSize($messageSize);
        }
        return $messageSize;
    }

    /**
     * Push a message in queue
     * @param Channel $channel
     * @param mixed $message
     * @param string|int $messageType
     * @return bool
     * @throws Exception
     */
    public function push(Channel $channel, $message, $messageType = null)
    {
        if (!$channel->hasResource()) {
            $this->connect($channel);
        }
        $messageType = $this->secureMessageType($messageType);
        $error = 0;
        $blockSend = (bool)$channel->getSettings()->getBlockSend();
        $maxSendRetry = (int)$channel->getSettings()->getMaxSendRetryTime();
        $serialize = (bool)$channel->getSettings()->getSerialize();
        $retry = 0;
        $success = false;
        while ($retry < $maxSendRetry) {
            $success = $this->msgSend($channel->getResource(), $messageType, $message, $serialize, $blockSend, $error);
/**
 * @return array|string
 */
function returnsDifferentValues($x) {
    if ($x) {
        return 'foo';
    }

    return array();
}

$x = returnsDifferentValues($y);
if (is_array($x)) {
    // $x is an array.
}
            $retry = ($success || (!$success && $error != MSG_EAGAIN)) ? $maxSendRetry : $retry + 1;
        }
        if (!$success) {
            throw new Exception("Error while sending message", $error);
        }
        return $success;
    }

    /**
     * @param $queue
     * @param $msgType
     * @param $message
     * @param bool|true $serialize
     * @param bool|true $blocking
     * @param null $errorCode
     * @return bool
     * @codeCoverageIgnore
     */
    protected function msgSend($queue, $msgType, $message, $serialize = true, $blocking = true, &$errorCode = null)
    {
        return msg_send($queue, $msgType, $message, $serialize, $blocking, $errorCode);
    }

    /**
     * Secure queue name to be int (due to semaphore)
     * @param string|int $name
     * @return int
     * @throws Exception
     */
    private function secureQueueName($name)
    {
        $name = (int)$this->stringToUniqueId($name);
        if ($name === 0) {
            throw new Exception('Queue name must be in INT > 0 to use semaphores');
        }
        return $name;
    }

    /**
     * Secure message type
     * @param string|int $messageType
     * @return int|null
     * @throws Exception
     */
    private function secureMessageType($messageType = null)
    {
        if (is_null($messageType)) {
            $messageType = 1;
        }
        $messageType = (int)$this->stringToUniqueId($messageType);
        if ($messageType === 0) {
            throw new Exception('Message Type must be in INT > 0 to use semaphores');
        }
        return $messageType;
    }

    /**
     * Convert string to a unique id
     * @param string $string
     * @return int
     */
    private function stringToUniqueId($string)
    {
        if (is_numeric($string)) {
            return (int)$string;
        }
        if ($this->hasEnvironment()) {
            $string .= $this->getEnvironment()->get();
        }
        $length = strlen($string);
        $return = 0;
        for ($i = 0; $i < $length; $i++) {

            $return += ord($string{$i});
        }
        return (int)($length . '0' . $return);
    }

    /**
     * Get queue configuration
     * @param resource $queueResource
     * @return array
     * @codeCoverageIgnore
     */
    protected function getConfiguration($queueResource)
    {
        return msg_stat_queue($queueResource);
    }

    /**
     * Define queue configuration
     * @param Channel $channel
     * @param array $params
     * @return bool
     */
    public function setConfiguration(Channel $channel, $params)
    {
        if (isset($params[self::CONFIGURATION_MESSAGE_SIZE])) {
            $channel->getSettings()->setMaxMessageSize((int)$params[self::CONFIGURATION_MESSAGE_SIZE]);
        }
        return $this->msgSetQueue($channel->getResource(), (array)$params);
    }

    /**
     * @param $queue
     * @param array $data
     * @return bool
     * @codeCoverageIgnore
     */
    protected function msgSetQueue($queue, array $data)
    {
        return msg_set_queue($queue, $data);
    }

    /**
     * Destroy a queue
     * @param Channel $channel
     * @return bool
     */
    public function destroy(Channel $channel)
    {
        if ($channel->hasResource()) {
            if ($this->msgRemoveQueue($channel->getResource())) {
                $channel->setResource(null);
                return true;
            }
            return false;
        }
        return true;
    }

    /**
     * @param $queue
     * @return bool
     * @codeCoverageIgnore
     */
    protected function msgRemoveQueue($queue)
    {
        return msg_remove_queue($queue);
    }

    /**
     * Reception options
     * @param Channel $channel
     * @return int
     */
    private function getReceiveFlags(Channel $channel)
    {
        $modeExcept = 0;
        $forceSize = 0;
        $settings = $channel->getSettings();
        if ($settings instanceof Channel\Settings\Native) {
            $modeExcept = $settings->getReceiveModeExcept() ? MSG_EXCEPT : 0;
            $forceSize = $settings->getReceiveForceSize() ? MSG_NOERROR : 0;
        }
        $blockReceive = $this->getParamBlockReceive($channel);

        return $blockReceive | $modeExcept | $forceSize;
    }

    /**
     * Check whether block or not on reception
     * @param Channel $channel
     * @return int
     */
    private function getParamBlockReceive(Channel $channel)
    {
        return $channel->getSettings()->getBlockReceive() ? 0 : MSG_IPC_NOWAIT;
    }

    /**
     * @todo factorize this method
     * @param Channel $channel
     * @return array
     */
    public function getStats(Channel $channel)
    {
        if (!$channel->hasResource()) {
            $this->connect($channel);
        }
        return $this->getConfiguration($channel->getResource());
    }

    /**
     * This methods do nothing since messages are auto-acked in SystemV. Sorry :(
     * @param Channel $channel
     * @param Message $message
     * @return $this
     */
    public function ackMessage(Channel $channel, Message $message)
    {
        return $this;
    }
}


1			<?php
2			namespace FMUP\Queue\Driver;
3
4			use FMUP\Environment;
5			use FMUP\Queue\Channel;
6			use FMUP\Queue\DriverInterface;
7			use FMUP\Queue\Exception;
8			use FMUP\Queue\Message;
9
10			class Native implements DriverInterface, Environment\OptionalInterface
11			{
12			use Environment\OptionalTrait;
13
14			const CONFIGURATION_PERM_UID = 'msg_perm.uid';
15			const CONFIGURATION_PERM_GID = 'msg_perm.gid';
16			const CONFIGURATION_PERM_MODE = 'msg_perm.mode';
17			const CONFIGURATION_SENT_TIME = 'msg_stime';
18			const CONFIGURATION_RECEIVED_TIME = 'msg_rtime';
19			const CONFIGURATION_UPDATE_TIME = 'msg_ctime';
20			const CONFIGURATION_MESSAGE_NUMBER = 'msg_qnum';
21			const CONFIGURATION_MESSAGE_SIZE = 'msg_qbytes';
22			const CONFIGURATION_SENDER_PID = 'msg_lspid';
23			const CONFIGURATION_RECEIVER_PID = 'msg_lrpid';
24
25			/**
26			* Connect to specified queue
27			* @param Channel $channel
28			* @return Channel
29			* @throws Exception
30			*/
31	6		public function connect(Channel $channel)
32			{
33	6		if (!$channel->hasResource()) {
34	6		$name = $this->secureQueueName($channel->getName());
35	5		$channel->setName($name);
36	5		$resource = $this->msgGetQueue($name);
37	5		$channel->setResource($resource);
38			}
39	5		return $channel;
40			}
41
42			/**
43			* @param string $name
44			* @return resource
45			* @codeCoverageIgnore
46			*/
47			protected function msgGetQueue($name)
48			{
49			return msg_get_queue($name);
50			}
51
52			/**
53			* Check if queue exists
54			* @param Channel $channel
55			* @return bool
56			* @throws Exception
57			*/
58	1		public function exists(Channel $channel)
59			{
60	1		$name = (!$channel->hasResource()) ? $this->secureQueueName($channel->getName()) : $channel->getName();
61	1		return $this->msgQueueExists($name);
62			}
63
64			/**
65			* @param $name
66			* @return bool
67			* @codeCoverageIgnore
68			*/
69			protected function msgQueueExists($name)
70			{
71			return msg_queue_exists($name);
72			}
73
74			/**
75			* Get a message from queue
76			*
77			* @param Channel $channel
78			* @param string $messageType
79			* @return mixed\|null
80			* @throws Exception
81			*/
82	3		public function pull(Channel $channel, $messageType = null)
83			{
84	3		if (!$channel->hasResource()) {
85	3		$this->connect($channel);
86			}
87	3		$messageType = $this->secureMessageType($messageType);
88	2		$receivedMessageType = 0;
89	2		$message = null;
90	2		$error = 0;
91	2		$messageSize = $this->getMessageSize($channel);
92	2		$success = $this->msgReceive(
93	2		$channel->getResource(),
94	2		$messageType,
95	2		$receivedMessageType,
96	2		$messageSize,
97	2		$message,
98	2		$channel->getSettings()->getSerialize(),
99	2		$this->getReceiveFlags($channel),
100	2		$error
101			);
102	2		$isNonBlockReceive = (MSG_IPC_NOWAIT & $this->getParamBlockReceive($channel));
103	2		$isNonBlockingPlusNoMessage = $isNonBlockReceive && ($error === MSG_ENOMSG);
104	2		if (!$success && !$isNonBlockingPlusNoMessage) {
105	1		throw new Exception("Error while receiving message", $error);
106			}
107	1		return $message ? (new Message())->setOriginal($message)->setTranslated($message) : null;
108			}
109
110			/**
111			* @param $queue
112			* @param $desiredMsgType
113			* @param $msgType
114			* @param $maxsize
115			* @param $message
116			* @param bool\|true $unSerialize
117			* @param int $flags
118			* @param null $errorCode
119			* @return bool
120			* @codeCoverageIgnore
121			*/
122			protected function msgReceive(
123			$queue,
124			$desiredMsgType,
125			&$msgType,
126			$maxsize,
127			&$message,
128			$unSerialize = true,
129			$flags = 0,
130			&$errorCode = null
131			) {
132			return msg_receive($queue, $desiredMsgType, $msgType, $maxsize, $message, $unSerialize, $flags, $errorCode);
133			}
134
135			/**
136			* Retrieve message maximum size for a given queue
137			* @param Channel $channel
138			* @return int
139			*/
140	2		private function getMessageSize(Channel $channel)
141			{
142	2		$messageSize = $channel->getSettings()->getMaxMessageSize();
143	2		if (!$messageSize) {
144	2		$configuration = $this->getConfiguration($channel->getResource());
145	2		$messageSize = (int)$configuration[self::CONFIGURATION_MESSAGE_SIZE];
146	2		$channel->getSettings()->setMaxMessageSize($messageSize);
147			}
148	2		return $messageSize;
149			}
150
151			/**
152			* Push a message in queue
153			* @param Channel $channel
154			* @param mixed $message
155			* @param string\|int $messageType
156			* @return bool
157			* @throws Exception
158			*/
159	2		public function push(Channel $channel, $message, $messageType = null)
160			{
161	2		if (!$channel->hasResource()) {
162	2		$this->connect($channel);
163			}
164	2		$messageType = $this->secureMessageType($messageType);
165	2		$error = 0;
166	2		$blockSend = (bool)$channel->getSettings()->getBlockSend();
167	2		$maxSendRetry = (int)$channel->getSettings()->getMaxSendRetryTime();
168	2		$serialize = (bool)$channel->getSettings()->getSerialize();
169	2		$retry = 0;
170	2		$success = false;
171	2		while ($retry < $maxSendRetry) {
172	2		$success = $this->msgSend($channel->getResource(), $messageType, $message, $serialize, $blockSend, $error);
			0 ignored issues – show Bug introduced 2017-04-18 14:53 UTC by Report Bug Copy Issue Report It seems like `$error` defined by `0` on line `165` can also be of type `integer`; however, `FMUP\Queue\Driver\Native::msgSend()` does only seem to accept `null`, maybe add an additional type check? If a method or function can return multiple different values and unless you are sure that you only can receive a single value in this context, we recommend to add an additional type check: /** * @return array\|string */ function returnsDifferentValues($x) { if ($x) { return 'foo'; } return array(); } $x = returnsDifferentValues($y); if (is_array($x)) { // $x is an array. } If this a common case that PHP Analyzer should handle natively, please let us know by opening an issue. Loading history...
173	2		$retry = ($success \|\| (!$success && $error != MSG_EAGAIN)) ? $maxSendRetry : $retry + 1;
174			}
175	2		if (!$success) {
176	1		throw new Exception("Error while sending message", $error);
177			}
178	1		return $success;
179			}
180
181			/**
182			* @param $queue
183			* @param $msgType
184			* @param $message
185			* @param bool\|true $serialize
186			* @param bool\|true $blocking
187			* @param null $errorCode
188			* @return bool
189			* @codeCoverageIgnore
190			*/
191			protected function msgSend($queue, $msgType, $message, $serialize = true, $blocking = true, &$errorCode = null)
192			{
193			return msg_send($queue, $msgType, $message, $serialize, $blocking, $errorCode);
194			}
195
196			/**
197			* Secure queue name to be int (due to semaphore)
198			* @param string\|int $name
199			* @return int
200			* @throws Exception
201			*/
202	7		private function secureQueueName($name)
203			{
204	7		$name = (int)$this->stringToUniqueId($name);
205	7		if ($name === 0) {
206	1		throw new Exception('Queue name must be in INT > 0 to use semaphores');
207			}
208	6		return $name;
209			}
210
211			/**
212			* Secure message type
213			* @param string\|int $messageType
214			* @return int\|null
215			* @throws Exception
216			*/
217	5		private function secureMessageType($messageType = null)
218			{
219	5		if (is_null($messageType)) {
220	4		$messageType = 1;
221			}
222	5		$messageType = (int)$this->stringToUniqueId($messageType);
223	5		if ($messageType === 0) {
224	1		throw new Exception('Message Type must be in INT > 0 to use semaphores');
225			}
226	4		return $messageType;
227			}
228
229			/**
230			* Convert string to a unique id
231			* @param string $string
232			* @return int
233			*/
234	9		private function stringToUniqueId($string)
235			{
236	9		if (is_numeric($string)) {
237	8		return (int)$string;
238			}
239	1		if ($this->hasEnvironment()) {
240	1		$string .= $this->getEnvironment()->get();
241			}
242	1		$length = strlen($string);
243	1		$return = 0;
244	1	View Code Duplication	for ($i = 0; $i < $length; $i++) {
			0 ignored issues – show Duplication introduced 2017-04-18 14:53 UTC by Report Bug Copy Issue Report This code seems to be duplicated across your project. Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation. You can also find more detailed suggestions in the “Code” section of your repository. Loading history...
245	1		$return += ord($string{$i});
246			}
247	1		return (int)($length . '0' . $return);
248			}
249
250			/**
251			* Get queue configuration
252			* @param resource $queueResource
253			* @return array
254			* @codeCoverageIgnore
255			*/
256			protected function getConfiguration($queueResource)
257			{
258			return msg_stat_queue($queueResource);
259			}
260
261			/**
262			* Define queue configuration
263			* @param Channel $channel
264			* @param array $params
265			* @return bool
266			*/
267	1		public function setConfiguration(Channel $channel, $params)
268			{
269	1		if (isset($params[self::CONFIGURATION_MESSAGE_SIZE])) {
270	1		$channel->getSettings()->setMaxMessageSize((int)$params[self::CONFIGURATION_MESSAGE_SIZE]);
271			}
272	1		return $this->msgSetQueue($channel->getResource(), (array)$params);
273			}
274
275			/**
276			* @param $queue
277			* @param array $data
278			* @return bool
279			* @codeCoverageIgnore
280			*/
281			protected function msgSetQueue($queue, array $data)
282			{
283			return msg_set_queue($queue, $data);
284			}
285
286			/**
287			* Destroy a queue
288			* @param Channel $channel
289			* @return bool
290			*/
291	1		public function destroy(Channel $channel)
292			{
293	1		if ($channel->hasResource()) {
294	1		if ($this->msgRemoveQueue($channel->getResource())) {
295	1		$channel->setResource(null);
296	1		return true;
297			}
298	1		return false;
299			}
300	1		return true;
301			}
302
303			/**
304			* @param $queue
305			* @return bool
306			* @codeCoverageIgnore
307			*/
308			protected function msgRemoveQueue($queue)
309			{
310			return msg_remove_queue($queue);
311			}
312
313			/**
314			* Reception options
315			* @param Channel $channel
316			* @return int
317			*/
318	2		private function getReceiveFlags(Channel $channel)
319			{
320	2		$modeExcept = 0;
321	2		$forceSize = 0;
322	2		$settings = $channel->getSettings();
323	2		if ($settings instanceof Channel\Settings\Native) {
324	2		$modeExcept = $settings->getReceiveModeExcept() ? MSG_EXCEPT : 0;
325	2		$forceSize = $settings->getReceiveForceSize() ? MSG_NOERROR : 0;
326			}
327	2		$blockReceive = $this->getParamBlockReceive($channel);
328
329	2		return $blockReceive \| $modeExcept \| $forceSize;
330			}
331
332			/**
333			* Check whether block or not on reception
334			* @param Channel $channel
335			* @return int
336			*/
337	2		private function getParamBlockReceive(Channel $channel)
338			{
339	2		return $channel->getSettings()->getBlockReceive() ? 0 : MSG_IPC_NOWAIT;
340			}
341
342			/**
343			* @todo factorize this method
344			* @param Channel $channel
345			* @return array
346			*/
347	1		public function getStats(Channel $channel)
348			{
349	1		if (!$channel->hasResource()) {
350	1		$this->connect($channel);
351			}
352	1		return $this->getConfiguration($channel->getResource());
353			}
354
355			/**
356			* This methods do nothing since messages are auto-acked in SystemV. Sorry :(
357			* @param Channel $channel
358			* @param Message $message
359			* @return $this
360			*/
361	1		public function ackMessage(Channel $channel, Message $message)
362			{
363	1		return $this;
364			}
365			}
366

Logi-CE / fmup

Native::pull() B last analyzed 2017-12-05 10:44 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

Native::pull() B
last analyzed 2017-12-05 10:44 UTC