Issues in Queue.php - Failed Issues - Inspection of "Add Php 7 support (WIP)" - dominionenterprises/mongo-queue-php - Measure and Improve Code Quality continuously with Scrutinizer

Failed Conditions

Pull Request — master (#46)

unknown

created 2016-05-26 17:07 UTC

src/Queue.php (2 issues)

Labels

Bug 2

Severity

Major 2

Upgrade to new PHP Analysis Engine

These results are based on our legacy PHP analysis, consider migrating to our new PHP analysis engine instead. Learn more

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

namespace DominionEnterprises\Mongo;

/**
 * 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()
 */
final class Queue implements QueueInterface
{
    const MONGO_INT32_MAX = 2147483647;//2147483648 can overflow in php mongo without using the MongoInt64

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

    /**
     * Construct queue.
     *
     * @param \MongoDB\Collection|string $collectionOrUrl A MongoCollection instance or the mongo connection url.
     * @param string $db the mongo db name
     * @param string $collection the collection name to use for the queue
     *
     * @throws \InvalidArgumentException $collectionOrUrl, $db or $collection was not a string
     */
    public function __construct($collectionOrUrl, $db = null, $collection = null)
    {
        if ($collectionOrUrl instanceof \MongoDB\Collection) {
            $this->collection = $collectionOrUrl;
            return;
        }

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

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

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

        $mongo = new \MongoDB\Client($collectionOrUrl, [], ['typeMap' => ['root' => 'array', 'document' => 'array', 'array' => 'array']]);
        $mongoDb = $mongo->selectDatabase($db);
        $this->collection = $mongoDb->selectCollection($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
     */
    public function ensureGetIndex(array $beforeSort = [], array $afterSort = [])
    {
        //using general rule: equality, sort, range or more equality tests in that order for index
        $completeFields = ['running' => 1];

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

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

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

        $completeFields['earliestGet'] = 1;

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

        //for the stuck messages query in get()
        $this->ensureIndex(['running' => 1, 'resetTimestamp' => 1]);
    }

    /**
     * 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 $includeRunning was not a boolean
     * @throws \InvalidArgumentException key in $fields was not a string
     * @throws \InvalidArgumentException value of $fields is not 1 or -1 for ascending and descending
     */
    public function ensureCountIndex(array $fields, $includeRunning)
    {
        if (!is_bool($includeRunning)) {
            throw new \InvalidArgumentException('$includeRunning was not a boolean');
        }

        $completeFields = [];

        if ($includeRunning) {
            $completeFields['running'] = 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 $runningResetDuration, $waitDurationInMillis or $pollDurationInMillis was not
     *                                   an int
     * @throws \InvalidArgumentException key in $query was not a string
     */
    public function get(array $query, $runningResetDuration, $waitDurationInMillis = 3000, $pollDurationInMillis = 200)
    {
        if (!is_int($runningResetDuration)) {
            throw new \InvalidArgumentException('$runningResetDuration was not an int');
        }

        if (!is_int($waitDurationInMillis)) {
            throw new \InvalidArgumentException('$waitDurationInMillis was not an int');
        }

        if (!is_int($pollDurationInMillis)) {
            throw new \InvalidArgumentException('$pollDurationInMillis was not an int');
        }

        if ($pollDurationInMillis < 0) {
            $pollDurationInMillis = 0;
        }

        //reset stuck messages
        $this->collection->updateMany(
            ['running' => true, 'resetTimestamp' => ['$lte' => new \MongoDB\BSON\UTCDateTime(microtime(true) * 1000)]],
            ['$set' => ['running' => false]]
        );

        $completeQuery = ['running' => false];
        foreach ($query as $key => $value) {
            if (!is_string($key)) {
                throw new \InvalidArgumentException('key in $query was not a string');
            }

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

        $completeQuery['earliestGet'] = ['$lte' => new \MongoDB\BSON\UTCDateTime(microtime(true) * 1000)];

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

        $update = ['$set' => ['resetTimestamp' => new \MongoDB\BSON\UTCDateTime($resetTimestamp * 1000), 'running' => true]];
        $options = ['sort' => ['priority' => 1, 'created' => 1]];

        //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
     */
    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 !== null) {
            $totalQuery['running'] = $running;
        }

        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
     */
    public function ack(array $message)
    {
        $id = null;
        if (array_key_exists('id', $message)) {
            $id = $message['id'];
        }

        if (!($id instanceof \MongoDB\BSON\ObjectID)) {
if ($x instanceof DoesNotExist) {
    // Do something.
}
            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 $earliestGet was not an int
     * @throws \InvalidArgumentException $priority was not a float
     * @throws \InvalidArgumentException $priority is NaN
     * @throws \InvalidArgumentException $newTimestamp was not a bool
     */
    public function ackSend(array $message, array $payload, $earliestGet = 0, $priority = 0.0, $newTimestamp = true)
    {
        $id = null;
        if (array_key_exists('id', $message)) {
            $id = $message['id'];
        }

        if (!($id instanceof \MongoDB\BSON\ObjectId)) {
if ($x instanceof DoesNotExist) {
    // Do something.
}
            throw new \InvalidArgumentException('$message does not have a field "id" that is a ObjectId');
        }

        if (!is_int($earliestGet)) {
            throw new \InvalidArgumentException('$earliestGet was not an int');
        }

        if (!is_float($priority)) {
            throw new \InvalidArgumentException('$priority was not a float');
        }

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

        if ($newTimestamp !== true && $newTimestamp !== false) {
            throw new \InvalidArgumentException('$newTimestamp was not a bool');
        }

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

        $toSet = [
            'payload' => $payload,
            'running' => false,
            'resetTimestamp' => new \MongoDB\BSON\UTCDateTime(self::MONGO_INT32_MAX),
            'earliestGet' => new \MongoDB\BSON\UTCDateTime($earliestGet),
            'priority' => $priority,
        ];
        if ($newTimestamp) {
            $toSet['created'] = new \MongoDB\BSON\UTCDateTime(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 $earliestGet was not an int
     * @throws \InvalidArgumentException $priority was not a float
     * @throws \InvalidArgumentException priority is NaN
     * @throws \InvalidArgumentException $newTimestamp was not a bool
     */
    public function requeue(array $message, $earliestGet = 0, $priority = 0.0, $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 $earliestGet was not an int
     * @throws \InvalidArgumentException $priority was not a float
     * @throws \InvalidArgumentException $priority is NaN
     */
    public function send(array $payload, $earliestGet = 0, $priority = 0.0)
    {
        if (!is_int($earliestGet)) {
            throw new \InvalidArgumentException('$earliestGet was not an int');
        }

        if (!is_float($priority)) {
            throw new \InvalidArgumentException('$priority was not a float');
        }

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

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

        $message = [
            'payload' => $payload,
            'running' => false,
            'resetTimestamp' => new \MongoDB\BSON\UTCDateTime(self::MONGO_INT32_MAX * 1000),
            'earliestGet' => new \MongoDB\BSON\UTCDateTime($earliestGet * 1000),
            'priority' => $priority,
            'created' => new \MongoDB\BSON\UTCDateTime(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
     */
    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;
                    }
                }

                //@codeCoverageIgnoreStart
            }
        }

        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
     */
    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;
        }
    }
}


Pull Request — master (#46)

src/Queue.php (2 issues)

Labels

Severity

Introduced By

Upgrade to new PHP Analysis Engine

1. Missing dependencies

2. Missing use statement

1. Missing dependencies

2. Missing use statement

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

dominionenterprises / mongo-queue-php

Pull Request — master (#46)

src/Queue.php (2 issues)

Labels

Severity

Introduced By

Upgrade to new PHP Analysis Engine

1. Missing dependencies

2. Missing use statement

1. Missing dependencies

2. Missing use statement

Duplication Side-by-Side

Filter issues like