NokitaKaze\Queue\AbstractQueueTransport

Complexity

Total Complexity

24

Size/Duplication

Total Lines	178
Duplicated Lines	0 %

Coupling/Cohesion

Components	3
Dependencies	2

Test Coverage

Coverage

100%

Importance

Changes	1
Bugs	0	Features	0

14 Methods

Rating	Name	Duplication	Size	Complexity
A	delete_message()	0	3	1
A	produce_message()	0	4	1
A	push()	0	10	3
A	set_same_time_flag()	0	6	2
A	set_callback_closure()	0	3	1
B	listen()	0	15	5
A	__clone()	0	5	2
A	clear_consumed_keys()	0	3	1
A	is_producer()	0	3	1
A	is_consumer()	0	3	1
A	get_real_key_for_message()	0	3	1
A	change_message_in_array()	0	16	3
A	is_equal_to()	0	3	1
A	generate_rnd_postfix()	0	3	1

<?php

    namespace NokitaKaze\Queue;

    abstract class AbstractQueueTransport implements QueueTransport {
        /**
         * @var iMessage[]|object[]
         */
        protected $_pushed_for_save = [];

        protected $_same_time_consumer_and_producer = 0;

        /**
         * @var callable|null
         */
        protected $_callback_closure = null;

        /**
         * @var integer[] В keys список индексов полученных сообщений
         */
        protected $_consumed_keys = [];

        /**
         * Удалить сообщение и сразу же записать это в БД
         *
         * @param iMessage|object $message
         *
         * @return boolean
         */
        function delete_message($message) {

It is generally recommended to explicitly declare the visibility for methods.

            return !empty($this->delete_messages([$message]));
        }

        /**
         * Загоняем сообщение в очередь и явно сохраняем
         *
         * @param mixed       $data
         * @param string|null $name Название сообщения. Если null, то можно дублировать
         * @param integer     $sort
         *
         * @throws QueueException
         */
        function produce_message($data, $name = null, $sort = 5) {

It is generally recommended to explicitly declare the visibility for methods.

            $this->push(Queue::build_message($data, $name, $sort));
            $this->save();
        }

        /**
         * Загоняем сообщение в очередь необходимых для сохранения сообщений
         *
         * @param iMessage[]|iMessage|object[]|object $stream
         *
         * @throws QueueException
         */
        function push($stream) {

It is generally recommended to explicitly declare the visibility for methods.

            $this->set_same_time_flag(1);
            if (is_array($stream)) {
                foreach ($stream as &$message) {
                    $this->_pushed_for_save[] = Queue::sanify_event_object($message);
                }
            } else {
                $this->_pushed_for_save[] = Queue::sanify_event_object($stream);
            }
        }

        /**
         * @param integer $flag
         *
         * @throws QueueException
         */
        protected function set_same_time_flag($flag) {
            $this->_same_time_consumer_and_producer |= $flag;
            if ($this->_same_time_consumer_and_producer === 3) {
                throw new QueueException('Consumer and producer at the same time', 18);
            }
        }

        /**
         * @param callable $closure
         */
        function set_callback_closure($closure) {

It is generally recommended to explicitly declare the visibility for methods.

            $this->_callback_closure = $closure;
        }

        /**
         * @param double|integer $wait_time
         *
         * @throws QueueException
         */
        function listen($wait_time = -1) {

It is generally recommended to explicitly declare the visibility for methods.

            $start = microtime(true);
            if (is_null($this->_callback_closure)) {
                throw new QueueException('Event listener has not been set', 3);
            }
            $closure = $this->_callback_closure;
            do {
                $message = $this->consume_next_message($wait_time);
                if (!is_null($message)) {
                    call_user_func($closure, $message);
                } else {
                    usleep(10000);
                }
            } while (($wait_time === -1) or ($start + $wait_time >= microtime(true)));
        }

        function __clone() {

It is generally recommended to explicitly declare the visibility for methods.

            foreach ($this->_pushed_for_save as &$message) {
                $message = clone $message;
            }
        }

        function clear_consumed_keys() {

It is generally recommended to explicitly declare the visibility for methods.

            $this->_consumed_keys = [];
        }

        /**
         * @return boolean
         */
        function is_producer() {

It is generally recommended to explicitly declare the visibility for methods.

            return (($this->_same_time_consumer_and_producer & 1) == 1);
        }

        /**
         * @return boolean
         */
        function is_consumer() {

It is generally recommended to explicitly declare the visibility for methods.

            return (($this->_same_time_consumer_and_producer & 2) == 2);
        }

        /**
         * @param iMessage|object $message
         *
         * @return string
         */
        static function get_real_key_for_message($message) {

It is generally recommended to explicitly declare the visibility for methods.

            return Queue::get_real_key_for_message($message);
        }

        /**
         * @param iMessage[] $messages
         * @param iMessage   $message
         * @param string     $message_key
         *
         * @return boolean
         */
        protected function change_message_in_array(array &$messages, $message, $message_key) {
            $exists = false;
            $message->is_read = true;

Accessing is_read on the interface NokitaKaze\Queue\iMessage suggest that you code against a concrete implementation. How about adding an instanceof check?

            foreach ($messages as $inner_id => &$inner_message) {
                $inner_message_key = self::get_real_key_for_message($inner_message);
                if ($inner_message_key == $message_key) {
                    $message->time_last_update = microtime(true);

Accessing time_last_update on the interface NokitaKaze\Queue\iMessage suggest that you code against a concrete implementation. How about adding an instanceof check?

                    $messages[$inner_id] = $message;
                    $exists = true;
                } else {
                    $inner_message->is_read = true;

Accessing is_read on the interface NokitaKaze\Queue\iMessage suggest that you code against a concrete implementation. How about adding an instanceof check?

                }
            }

            return $exists;
        }

        /**
         * @param QueueTransport $queue
         *
         * @return boolean
         */
        function is_equal_to($queue) {

It is generally recommended to explicitly declare the visibility for methods.

            return (get_class($this) == get_class($queue));
        }

        /**
         * @return string
         */
        static function generate_rnd_postfix() {

It is generally recommended to explicitly declare the visibility for methods.

            return Queue::generate_rnd_postfix();
        }

        // @todo Удаление конкретных ключей из consumed. Причем туда передаётся кложур, в который передаётся название ключа
        // @todo Удаление конкретных ключей из индекса и очереди, с указанием max_create_timestamp,
        // чтобы не хранить в очереди те же сообщения, пришедшие ещё раз
    }

?>