AbstractQueue::ackSend() - Code Metrics - Inspection of "[WIP] Version 3" - traderinteractive/mongo-queue-php - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#60)

by Chad

created 2018-02-05 17:54 UTC

AbstractQueue::ackSend() B

↳ Parent: AbstractQueue

Complexity

Conditions	5
Paths	8

Size

Total Lines	36
Code Lines	21

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
dl	0
loc	36
rs	8.439
c	0
b	0
f	0
cc	5
eloc	21
nc	8
nop	5

<?php
/**
 * Defines the TraderInteractive\Mongo\Queue class.
 */

namespace TraderInteractive\Mongo;

use MongoDB\BSON\UTCDateTime;

/**
 * Abstraction of mongo db collection as priority queue.
 *
 * Tied priorities are ordered by time. So you may use a single priority for normal queuing (default args exist for
 * this purpose).  Using a random priority achieves random get()
 */
abstract class AbstractQueue implements QueueInterface
{
    /**
     * Maximum millisecond value to use for UTCDateTime creation.
     *
     * @var integer
     */
    const MONGO_INT32_MAX = PHP_INT_MAX;

    /**
     * mongo collection to use for queue.
     *
     * @var \MongoDB\Collection
     */
    protected $collection;

    /**
     * Ensure an index for the get() method.
     *
     * @param array $beforeSort Fields in get() call to index before the sort field in same format
     *                          as \MongoDB\Collection::ensureIndex()
     * @param array $afterSort  Fields in get() call to index after the sort field in same format as
     *                          \MongoDB\Collection::ensureIndex()
     *
     * @return void
     *
     * @throws \InvalidArgumentException value of $beforeSort or $afterSort is not 1 or -1 for ascending and descending
     * @throws \InvalidArgumentException key in $beforeSort or $afterSort was not a string
     */
    final public function ensureGetIndex(array $beforeSort = [], array $afterSort = [])
    {
        //using general rule: equality, sort, range or more equality tests in that order for index
        $completeFields = ['earliestGet' => 1];

        self::verifySort($beforeSort, 'beforeSort', $completeFields);

        $completeFields['priority'] = 1;
        $completeFields['created'] = 1;

        self::verifySort($afterSort, 'afterSort', $completeFields);

        //for the main query in get()
        $this->ensureIndex($completeFields);
    }

    /**
     * Ensure an index for the count() method.
     * Is a no-op if the generated index is a prefix of an existing one. If you have a similar ensureGetIndex call,
     * call it first.
     *
     * @param array $fields fields in count() call to index in same format as \MongoDB\Collection::createIndex()
     * @param bool $includeRunning whether to include the running field in the index
     *
     * @return void
     *
     * @throws \InvalidArgumentException key in $fields was not a string
     * @throws \InvalidArgumentException value of $fields is not 1 or -1 for ascending and descending
     */
    final public function ensureCountIndex(array $fields, bool $includeRunning)
    {
        $completeFields = [];

        if ($includeRunning) {
            $completeFields['earliestGet'] = 1;
        }

        self::verifySort($fields, 'fields', $completeFields);

        $this->ensureIndex($completeFields);
    }

    /**
     * Get a non running message from the queue.
     *
     * @param array $query in same format as \MongoDB\Collection::find() where top level fields do not contain
     *                     operators. Lower level fields can however. eg: valid {a: {$gt: 1}, "b.c": 3},
     *                     invalid {$and: [{...}, {...}]}
     * @param int $runningResetDuration second duration the message can stay unacked before it resets and can be
     *                                  retreived again.
     * @param int $waitDurationInMillis millisecond duration to wait for a message.
     * @param int $pollDurationInMillis millisecond duration to wait between polls.
     *
     * @return array|null the message or null if one is not found
     *
     * @throws \InvalidArgumentException key in $query was not a string
     */
    final public function get(
        array $query,
        int $runningResetDuration,
        int $waitDurationInMillis = 3000,
        int $pollDurationInMillis = 200
    ) {
        if ($pollDurationInMillis < 0) {
            $pollDurationInMillis = 0;
        }

        $completeQuery = ['earliestGet' => ['$lte' => new UTCDateTime((int)(microtime(true) * 1000))]];
        foreach ($query as $key => $value) {

            if (!is_string($key)) {
                throw new \InvalidArgumentException('key in $query was not a string');
            }

            $completeQuery["payload.{$key}"] = $value;
        }

        $resetTimestamp = time() + $runningResetDuration;
        //ints overflow to floats
        if (!is_int($resetTimestamp)) {
            $resetTimestamp = $runningResetDuration > 0 ? self::MONGO_INT32_MAX : 0;
        }

        $resetTimestamp = min(max(0, $resetTimestamp * 1000), self::MONGO_INT32_MAX);

        $update = ['$set' => ['earliestGet' => new UTCDateTime($resetTimestamp)]];
        $options = [
            'sort' => ['priority' => 1, 'created' => 1],
            'typeMap' => ['root' => 'array', 'document' => 'array', 'array' => 'array'],
        ];

        //ints overflow to floats, should be fine
        $end = microtime(true) + ($waitDurationInMillis / 1000.0);

        $sleepTime = $pollDurationInMillis * 1000;
        //ints overflow to floats and already checked $pollDurationInMillis was positive
        if (!is_int($sleepTime)) {
            //ignore since testing a giant sleep takes too long
            //@codeCoverageIgnoreStart
            $sleepTime = PHP_INT_MAX;
        }   //@codeCoverageIgnoreEnd

        while (true) {
            $message = $this->collection->findOneAndUpdate($completeQuery, $update, $options);
            //checking if _id exist because findAndModify doesnt seem to return null when it can't match the query on
            //older mongo extension
            if ($message !== null && array_key_exists('_id', $message)) {
                // findOneAndUpdate does not correctly return result according to typeMap options so just refetch.
                $message = $this->collection->findOne(['_id' => $message['_id']]);
                //id on left of union operator so a possible id in payload doesnt wipe it out the generated one
                return ['id' => $message['_id']] + (array)$message['payload'];
            }

            if (microtime(true) >= $end) {
                return null;
            }

            usleep($sleepTime);
        }

        //ignore since always return from the function from the while loop
        //@codeCoverageIgnoreStart
    }
    //@codeCoverageIgnoreEnd

    /**
     * Count queue messages.
     *
     * @param array $query in same format as \MongoDB\Collection::find() where top level fields do not contain
     *                     operators. Lower level fields can however. eg: valid {a: {$gt: 1}, "b.c": 3},
     *                     invalid {$and: [{...}, {...}]}
     * @param bool|null $running query a running message or not or all
     *
     * @return int the count
     *
     * @throws \InvalidArgumentException $running was not null and not a bool
     * @throws \InvalidArgumentException key in $query was not a string
     */
    final public function count(array $query, $running = null)
    {
        if ($running !== null && !is_bool($running)) {
            throw new \InvalidArgumentException('$running was not null and not a bool');
        }

        $totalQuery = [];

        if ($running === true || $running === false) {
            $key = $running ? '$gt' : '$lte';
            $totalQuery['earliestGet'] = [$key => new UTCDateTime((int)(microtime(true) * 1000))];
        }

        foreach ($query as $key => $value) {

            if (!is_string($key)) {
                throw new \InvalidArgumentException('key in $query was not a string');
            }

            $totalQuery["payload.{$key}"] = $value;
        }

        return $this->collection->count($totalQuery);
    }

    /**
     * Acknowledge a message was processed and remove from queue.
     *
     * @param array $message message received from get()
     *
     * @return void
     *
     * @throws \InvalidArgumentException $message does not have a field "id" that is a MongoDB\BSON\ObjectID
     */
    final public function ack(array $message)
    {
        $id = null;
        if (array_key_exists('id', $message)) {
            $id = $message['id'];
        }

        if (!(is_a($id, 'MongoDB\BSON\ObjectID'))) {
            throw new \InvalidArgumentException('$message does not have a field "id" that is a ObjectID');
        }

        $this->collection->deleteOne(['_id' => $id]);
    }

    /**
     * Atomically acknowledge and send a message to the queue.
     *
     * @param array $message the message to ack received from get()
     * @param array $payload the data to store in the message to send. Data is handled same way
     *                       as \MongoDB\Collection::insertOne()
     * @param int $earliestGet earliest unix timestamp the message can be retreived.
     * @param float $priority priority for order out of get(). 0 is higher priority than 1
     * @param bool $newTimestamp true to give the payload a new timestamp or false to use given message timestamp
     *
     * @return void
     *
     * @throws \InvalidArgumentException $message does not have a field "id" that is a ObjectID
     * @throws \InvalidArgumentException $priority is NaN
     */
    final public function ackSend(
        array $message,
        array $payload,
        int $earliestGet = 0,
        float $priority = 0.0,
        bool $newTimestamp = true
    ) {
        $id = null;
        if (array_key_exists('id', $message)) {
            $id = $message['id'];
        }

        if (!(is_a($id, 'MongoDB\BSON\ObjectID'))) {
            throw new \InvalidArgumentException('$message does not have a field "id" that is a ObjectID');
        }

        if (is_nan($priority)) {
            throw new \InvalidArgumentException('$priority was NaN');
        }

        //Ensure $earliestGet is between 0 and MONGO_INT32_MAX
        $earliestGet = min(max(0, $earliestGet * 1000), self::MONGO_INT32_MAX);

        $toSet = [
            'payload' => $payload,
            'earliestGet' => new UTCDateTime($earliestGet),
            'priority' => $priority,
        ];
        if ($newTimestamp) {
            $toSet['created'] = new UTCDateTime((int)(microtime(true) * 1000));
        }

        //using upsert because if no documents found then the doc was removed (SHOULD ONLY HAPPEN BY SOMEONE MANUALLY)
        //so we can just send
        $this->collection->updateOne(['_id' => $id], ['$set' => $toSet], ['upsert' => true]);
    }

    /**
     * Requeue message to the queue. Same as ackSend() with the same message.
     *
     * @param array $message message received from get().
     * @param int $earliestGet earliest unix timestamp the message can be retreived.
     * @param float $priority priority for order out of get(). 0 is higher priority than 1
     * @param bool $newTimestamp true to give the payload a new timestamp or false to use given message timestamp
     *
     * @return void
     *
     * @throws \InvalidArgumentException $message does not have a field "id" that is a ObjectID
     * @throws \InvalidArgumentException priority is NaN
     */
    final public function requeue(
        array $message,
        int $earliestGet = 0,
        float $priority = 0.0,
        bool $newTimestamp = true
    ) {
        $forRequeue = $message;
        unset($forRequeue['id']);
        $this->ackSend($message, $forRequeue, $earliestGet, $priority, $newTimestamp);
    }

    /**
     * Send a message to the queue.
     *
     * @param array $payload the data to store in the message. Data is handled same way
     *                       as \MongoDB\Collection::insertOne()
     * @param int $earliestGet earliest unix timestamp the message can be retreived.
     * @param float $priority priority for order out of get(). 0 is higher priority than 1
     *
     * @return void
     *
     * @throws \InvalidArgumentException $priority is NaN
     */
    final public function send(array $payload, int $earliestGet = 0, float $priority = 0.0)
    {
        if (is_nan($priority)) {
            throw new \InvalidArgumentException('$priority was NaN');
        }

        //Ensure $earliestGet is between 0 and MONGO_INT32_MAX
        $earliestGet = min(max(0, $earliestGet * 1000), self::MONGO_INT32_MAX);

        $message = [
            'payload' => $payload,
            'earliestGet' => new UTCDateTime($earliestGet),
            'priority' => $priority,
            'created' => new UTCDateTime((int)(microtime(true) * 1000)),
        ];

        $this->collection->insertOne($message);
    }

    /**
     * Ensure index of correct specification and a unique name whether the specification or name already exist or not.
     * Will not create index if $index is a prefix of an existing index
     *
     * @param array $index index to create in same format as \MongoDB\Collection::createIndex()
     *
     * @return void
     *
     * @throws \Exception couldnt create index after 5 attempts
     */
    final private function ensureIndex(array $index)
    {
        //if $index is a prefix of any existing index we are good
        foreach ($this->collection->listIndexes() as $existingIndex) {
            $slice = array_slice($existingIndex['key'], 0, count($index), true);
            if ($slice === $index) {
                return;
            }
        }

        for ($i = 0; $i < 5; ++$i) {
            for ($name = uniqid(); strlen($name) > 0; $name = substr($name, 0, -1)) {
                //creating an index with same name and different spec does nothing.
                //creating an index with same spec and different name does nothing.
                //so we use any generated name, and then find the right spec after we have called,
                //and just go with that name.
                try {
                    $this->collection->createIndex($index, ['name' => $name, 'background' => true]);
                } catch (\MongoDB\Exception\Exception $e) {
                    //this happens when the name was too long, let continue
                }

                foreach ($this->collection->listIndexes() as $existingIndex) {
                    if ($existingIndex['key'] === $index) {
                        return;
                    }
                }
            }
        }

        throw new \Exception('couldnt create index after 5 attempts');
        //@codeCoverageIgnoreEnd
    }

    /**
     * Helper method to validate keys and values for the given sort array
     *
     * @param array  $sort             The proposed sort for a mongo index.
     * @param string $label            The name of the variable given to the public ensureXIndex method.
     * @param array  &$completedFields The final index array with payload. prefix added to fields.
     *
     * @return void
     */
    final private static function verifySort(array $sort, $label, &$completeFields)
    {
        foreach ($sort as $key => $value) {
            if (!is_string($key)) {
                throw new \InvalidArgumentException("key in \${$label} was not a string");
            }

            if ($value !== 1 && $value !== -1) {
                throw new \InvalidArgumentException(
                    "value of \${$label} is not 1 or -1 for ascending and descending"
                );
            }

            $completeFields["payload.{$key}"] = $value;
        }
    }
}


1		<?php
2		/**
3		* Defines the TraderInteractive\Mongo\Queue class.
4		*/
5
6		namespace TraderInteractive\Mongo;
7
8		use MongoDB\BSON\UTCDateTime;
9
10		/**
11		* Abstraction of mongo db collection as priority queue.
12		*
13		* Tied priorities are ordered by time. So you may use a single priority for normal queuing (default args exist for
14		* this purpose). Using a random priority achieves random get()
15		*/
16		abstract class AbstractQueue implements QueueInterface
17		{
18		/**
19		* Maximum millisecond value to use for UTCDateTime creation.
20		*
21		* @var integer
22		*/
23		const MONGO_INT32_MAX = PHP_INT_MAX;
24
25		/**
26		* mongo collection to use for queue.
27		*
28		* @var \MongoDB\Collection
29		*/
30		protected $collection;
31
32		/**
33		* Ensure an index for the get() method.
34		*
35		* @param array $beforeSort Fields in get() call to index before the sort field in same format
36		* as \MongoDB\Collection::ensureIndex()
37		* @param array $afterSort Fields in get() call to index after the sort field in same format as
38		* \MongoDB\Collection::ensureIndex()
39		*
40		* @return void
41		*
42		* @throws \InvalidArgumentException value of $beforeSort or $afterSort is not 1 or -1 for ascending and descending
43		* @throws \InvalidArgumentException key in $beforeSort or $afterSort was not a string
44		*/
45		final public function ensureGetIndex(array $beforeSort = [], array $afterSort = [])
46		{
47		//using general rule: equality, sort, range or more equality tests in that order for index
48		$completeFields = ['earliestGet' => 1];
49
50		self::verifySort($beforeSort, 'beforeSort', $completeFields);
51
52		$completeFields['priority'] = 1;
53		$completeFields['created'] = 1;
54
55		self::verifySort($afterSort, 'afterSort', $completeFields);
56
57		//for the main query in get()
58		$this->ensureIndex($completeFields);
59		}
60
61		/**
62		* Ensure an index for the count() method.
63		* Is a no-op if the generated index is a prefix of an existing one. If you have a similar ensureGetIndex call,
64		* call it first.
65		*
66		* @param array $fields fields in count() call to index in same format as \MongoDB\Collection::createIndex()
67		* @param bool $includeRunning whether to include the running field in the index
68		*
69		* @return void
70		*
71		* @throws \InvalidArgumentException key in $fields was not a string
72		* @throws \InvalidArgumentException value of $fields is not 1 or -1 for ascending and descending
73		*/
74		final public function ensureCountIndex(array $fields, bool $includeRunning)
75		{
76		$completeFields = [];
77
78		if ($includeRunning) {
79		$completeFields['earliestGet'] = 1;
80		}
81
82		self::verifySort($fields, 'fields', $completeFields);
83
84		$this->ensureIndex($completeFields);
85		}
86
87		/**
88		* Get a non running message from the queue.
89		*
90		* @param array $query in same format as \MongoDB\Collection::find() where top level fields do not contain
91		* operators. Lower level fields can however. eg: valid {a: {$gt: 1}, "b.c": 3},
92		* invalid {$and: [{...}, {...}]}
93		* @param int $runningResetDuration second duration the message can stay unacked before it resets and can be
94		* retreived again.
95		* @param int $waitDurationInMillis millisecond duration to wait for a message.
96		* @param int $pollDurationInMillis millisecond duration to wait between polls.
97		*
98		* @return array\|null the message or null if one is not found
99		*
100		* @throws \InvalidArgumentException key in $query was not a string
101		*/
102		final public function get(
103		array $query,
104		int $runningResetDuration,
105		int $waitDurationInMillis = 3000,
106		int $pollDurationInMillis = 200
107		) {
108		if ($pollDurationInMillis < 0) {
109		$pollDurationInMillis = 0;
110		}
111
112		$completeQuery = ['earliestGet' => ['$lte' => new UTCDateTime((int)(microtime(true) * 1000))]];
113	View Code Duplication	foreach ($query as $key => $value) {
		0 ignored issues – show Duplication introduced 2017-12-13 16:51 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...
114		if (!is_string($key)) {
115		throw new \InvalidArgumentException('key in $query was not a string');
116		}
117
118		$completeQuery["payload.{$key}"] = $value;
119		}
120
121		$resetTimestamp = time() + $runningResetDuration;
122		//ints overflow to floats
123		if (!is_int($resetTimestamp)) {
124		$resetTimestamp = $runningResetDuration > 0 ? self::MONGO_INT32_MAX : 0;
125		}
126
127		$resetTimestamp = min(max(0, $resetTimestamp * 1000), self::MONGO_INT32_MAX);
128
129		$update = ['$set' => ['earliestGet' => new UTCDateTime($resetTimestamp)]];
130		$options = [
131		'sort' => ['priority' => 1, 'created' => 1],
132		'typeMap' => ['root' => 'array', 'document' => 'array', 'array' => 'array'],
133		];
134
135		//ints overflow to floats, should be fine
136		$end = microtime(true) + ($waitDurationInMillis / 1000.0);
137
138		$sleepTime = $pollDurationInMillis * 1000;
139		//ints overflow to floats and already checked $pollDurationInMillis was positive
140		if (!is_int($sleepTime)) {
141		//ignore since testing a giant sleep takes too long
142		//@codeCoverageIgnoreStart
143		$sleepTime = PHP_INT_MAX;
144		} //@codeCoverageIgnoreEnd
145
146		while (true) {
147		$message = $this->collection->findOneAndUpdate($completeQuery, $update, $options);
148		//checking if _id exist because findAndModify doesnt seem to return null when it can't match the query on
149		//older mongo extension
150		if ($message !== null && array_key_exists('_id', $message)) {
151		// findOneAndUpdate does not correctly return result according to typeMap options so just refetch.
152		$message = $this->collection->findOne(['_id' => $message['_id']]);
153		//id on left of union operator so a possible id in payload doesnt wipe it out the generated one
154		return ['id' => $message['_id']] + (array)$message['payload'];
155		}
156
157		if (microtime(true) >= $end) {
158		return null;
159		}
160
161		usleep($sleepTime);
162		}
163
164		//ignore since always return from the function from the while loop
165		//@codeCoverageIgnoreStart
166		}
167		//@codeCoverageIgnoreEnd
168
169		/**
170		* Count queue messages.
171		*
172		* @param array $query in same format as \MongoDB\Collection::find() where top level fields do not contain
173		* operators. Lower level fields can however. eg: valid {a: {$gt: 1}, "b.c": 3},
174		* invalid {$and: [{...}, {...}]}
175		* @param bool\|null $running query a running message or not or all
176		*
177		* @return int the count
178		*
179		* @throws \InvalidArgumentException $running was not null and not a bool
180		* @throws \InvalidArgumentException key in $query was not a string
181		*/
182		final public function count(array $query, $running = null)
183		{
184		if ($running !== null && !is_bool($running)) {
185		throw new \InvalidArgumentException('$running was not null and not a bool');
186		}
187
188		$totalQuery = [];
189
190		if ($running === true \|\| $running === false) {
191		$key = $running ? '$gt' : '$lte';
192		$totalQuery['earliestGet'] = [$key => new UTCDateTime((int)(microtime(true) * 1000))];
193		}
194
195	View Code Duplication	foreach ($query as $key => $value) {
		0 ignored issues – show Duplication introduced 2017-12-13 16:51 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...
196		if (!is_string($key)) {
197		throw new \InvalidArgumentException('key in $query was not a string');
198		}
199
200		$totalQuery["payload.{$key}"] = $value;
201		}
202
203		return $this->collection->count($totalQuery);
204		}
205
206		/**
207		* Acknowledge a message was processed and remove from queue.
208		*
209		* @param array $message message received from get()
210		*
211		* @return void
212		*
213		* @throws \InvalidArgumentException $message does not have a field "id" that is a MongoDB\BSON\ObjectID
214		*/
215		final public function ack(array $message)
216		{
217		$id = null;
218		if (array_key_exists('id', $message)) {
219		$id = $message['id'];
220		}
221
222		if (!(is_a($id, 'MongoDB\BSON\ObjectID'))) {
223		throw new \InvalidArgumentException('$message does not have a field "id" that is a ObjectID');
224		}
225
226		$this->collection->deleteOne(['_id' => $id]);
227		}
228
229		/**
230		* Atomically acknowledge and send a message to the queue.
231		*
232		* @param array $message the message to ack received from get()
233		* @param array $payload the data to store in the message to send. Data is handled same way
234		* as \MongoDB\Collection::insertOne()
235		* @param int $earliestGet earliest unix timestamp the message can be retreived.
236		* @param float $priority priority for order out of get(). 0 is higher priority than 1
237		* @param bool $newTimestamp true to give the payload a new timestamp or false to use given message timestamp
238		*
239		* @return void
240		*
241		* @throws \InvalidArgumentException $message does not have a field "id" that is a ObjectID
242		* @throws \InvalidArgumentException $priority is NaN
243		*/
244		final public function ackSend(
245		array $message,
246		array $payload,
247		int $earliestGet = 0,
248		float $priority = 0.0,
249		bool $newTimestamp = true
250		) {
251		$id = null;
252		if (array_key_exists('id', $message)) {
253		$id = $message['id'];
254		}
255
256		if (!(is_a($id, 'MongoDB\BSON\ObjectID'))) {
257		throw new \InvalidArgumentException('$message does not have a field "id" that is a ObjectID');
258		}
259
260		if (is_nan($priority)) {
261		throw new \InvalidArgumentException('$priority was NaN');
262		}
263
264		//Ensure $earliestGet is between 0 and MONGO_INT32_MAX
265		$earliestGet = min(max(0, $earliestGet * 1000), self::MONGO_INT32_MAX);
266
267		$toSet = [
268		'payload' => $payload,
269		'earliestGet' => new UTCDateTime($earliestGet),
270		'priority' => $priority,
271		];
272		if ($newTimestamp) {
273		$toSet['created'] = new UTCDateTime((int)(microtime(true) * 1000));
274		}
275
276		//using upsert because if no documents found then the doc was removed (SHOULD ONLY HAPPEN BY SOMEONE MANUALLY)
277		//so we can just send
278		$this->collection->updateOne(['_id' => $id], ['$set' => $toSet], ['upsert' => true]);
279		}
280
281		/**
282		* Requeue message to the queue. Same as ackSend() with the same message.
283		*
284		* @param array $message message received from get().
285		* @param int $earliestGet earliest unix timestamp the message can be retreived.
286		* @param float $priority priority for order out of get(). 0 is higher priority than 1
287		* @param bool $newTimestamp true to give the payload a new timestamp or false to use given message timestamp
288		*
289		* @return void
290		*
291		* @throws \InvalidArgumentException $message does not have a field "id" that is a ObjectID
292		* @throws \InvalidArgumentException priority is NaN
293		*/
294		final public function requeue(
295		array $message,
296		int $earliestGet = 0,
297		float $priority = 0.0,
298		bool $newTimestamp = true
299		) {
300		$forRequeue = $message;
301		unset($forRequeue['id']);
302		$this->ackSend($message, $forRequeue, $earliestGet, $priority, $newTimestamp);
303		}
304
305		/**
306		* Send a message to the queue.
307		*
308		* @param array $payload the data to store in the message. Data is handled same way
309		* as \MongoDB\Collection::insertOne()
310		* @param int $earliestGet earliest unix timestamp the message can be retreived.
311		* @param float $priority priority for order out of get(). 0 is higher priority than 1
312		*
313		* @return void
314		*
315		* @throws \InvalidArgumentException $priority is NaN
316		*/
317		final public function send(array $payload, int $earliestGet = 0, float $priority = 0.0)
318		{
319		if (is_nan($priority)) {
320		throw new \InvalidArgumentException('$priority was NaN');
321		}
322
323		//Ensure $earliestGet is between 0 and MONGO_INT32_MAX
324		$earliestGet = min(max(0, $earliestGet * 1000), self::MONGO_INT32_MAX);
325
326		$message = [
327		'payload' => $payload,
328		'earliestGet' => new UTCDateTime($earliestGet),
329		'priority' => $priority,
330		'created' => new UTCDateTime((int)(microtime(true) * 1000)),
331		];
332
333		$this->collection->insertOne($message);
334		}
335
336		/**
337		* Ensure index of correct specification and a unique name whether the specification or name already exist or not.
338		* Will not create index if $index is a prefix of an existing index
339		*
340		* @param array $index index to create in same format as \MongoDB\Collection::createIndex()
341		*
342		* @return void
343		*
344		* @throws \Exception couldnt create index after 5 attempts
345		*/
346		final private function ensureIndex(array $index)
347		{
348		//if $index is a prefix of any existing index we are good
349		foreach ($this->collection->listIndexes() as $existingIndex) {
350		$slice = array_slice($existingIndex['key'], 0, count($index), true);
351		if ($slice === $index) {
352		return;
353		}
354		}
355
356		for ($i = 0; $i < 5; ++$i) {
357		for ($name = uniqid(); strlen($name) > 0; $name = substr($name, 0, -1)) {
358		//creating an index with same name and different spec does nothing.
359		//creating an index with same spec and different name does nothing.
360		//so we use any generated name, and then find the right spec after we have called,
361		//and just go with that name.
362		try {
363		$this->collection->createIndex($index, ['name' => $name, 'background' => true]);
364		} catch (\MongoDB\Exception\Exception $e) {
365		//this happens when the name was too long, let continue
366		}
367
368		foreach ($this->collection->listIndexes() as $existingIndex) {
369		if ($existingIndex['key'] === $index) {
370		return;
371		}
372		}
373		}
374		}
375
376		throw new \Exception('couldnt create index after 5 attempts');
377		//@codeCoverageIgnoreEnd
378		}
379
380		/**
381		* Helper method to validate keys and values for the given sort array
382		*
383		* @param array $sort The proposed sort for a mongo index.
384		* @param string $label The name of the variable given to the public ensureXIndex method.
385		* @param array &$completedFields The final index array with payload. prefix added to fields.
386		*
387		* @return void
388		*/
389		final private static function verifySort(array $sort, $label, &$completeFields)
390		{
391		foreach ($sort as $key => $value) {
392		if (!is_string($key)) {
393		throw new \InvalidArgumentException("key in \${$label} was not a string");
394		}
395
396		if ($value !== 1 && $value !== -1) {
397		throw new \InvalidArgumentException(
398		"value of \${$label} is not 1 or -1 for ascending and descending"
399		);
400		}
401
402		$completeFields["payload.{$key}"] = $value;
403		}
404		}
405		}
406

traderinteractive / mongo-queue-php

Pull Request — master (#60)

AbstractQueue::ackSend() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like