Merge pull request #71 from chrisryan/unordered

Add UnorderedQueue class.

Author: chadicus

.github/workflows/php.yml

@@ -8,7 +8,7 @@ on:
 
 jobs:
   build:
-    runs-on: ubuntu-18.04
+    runs-on: ubuntu-latest
     strategy:
       matrix:
         php-versions: ['7.2', '7.3', '7.4', '8.0', '8.1', '8.2']

src/AbstractQueue.php

@@ -157,7 +157,7 @@ final public function get(array $query, array $options = []) : array
             $query
         );
 
-        $options += self::DEFAULT_GET_OPTIONS;
+        $options += static::DEFAULT_GET_OPTIONS;
         $update = ['$set' => ['earliestGet' => $this->calculateEarliestGet($options['runningResetDuration'])]];
         $end = $this->calculateEndTime($options['waitDurationInMillis']);
         $sleepTime = $this->calculateSleepTime($options['pollDurationInMillis']);
@@ -305,12 +305,12 @@ private function calculateEarliestGet(int $runningResetDuration) : UTCDateTime
     {
         $resetTimestamp = time() + $runningResetDuration;
         //ints overflow to floats, max at PHP_INT_MAX
-        return new UTCDateTime(min(max(0, $resetTimestamp * 1000), self::MONGO_INT32_MAX));
+        return new UTCDateTime(min(max(0, $resetTimestamp * 1000), static::MONGO_INT32_MAX));
     }
 
     private function tryFindOneAndUpdate(array $query, array $update, ArrayObject $messages) : bool
     {
-        $document = $this->collection->findOneAndUpdate($query, $update, self::FIND_ONE_AND_UPDATE_OPTIONS);
+        $document = $this->collection->findOneAndUpdate($query, $update, static::FIND_ONE_AND_UPDATE_OPTIONS);
         if ($document === null) {
             return false;
         }

src/UnorderedQueue.php

@@ -0,0 +1,48 @@
+<?php
+/**
+ * Defines the TraderInteractive\Mongo\UnorderedQueue class.
+ */
+
+namespace TraderInteractive\Mongo;
+
+use MongoDB\Client;
+use MongoDB\Operation\FindOneAndUpdate;
+
+/**
+ * Abstraction of mongo db collection as unordered queue.
+ */
+final class UnorderedQueue extends AbstractQueue implements QueueInterface
+{
+    /**
+     * Override from AbstractQueue class to remove sort options.
+     *
+     * @var array
+     */
+    const FIND_ONE_AND_UPDATE_OPTIONS = [
+        'typeMap' => ['root' => 'array', 'document' => 'array', 'array' => 'array'],
+        'returnDocument' => FindOneAndUpdate::RETURN_DOCUMENT_AFTER,
+    ];
+
+    /**
+     * 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, string $db = null, string $collection = null)
+    {
+        if ($collectionOrUrl instanceof \MongoDB\Collection) {
+            $this->collection = $collectionOrUrl;
+            return;
+        }
+
+        if (!is_string($collectionOrUrl)) {
+            throw new \InvalidArgumentException('$collectionOrUrl was not a string');
+        }
+
+        $this->collection = (new Client($collectionOrUrl))->selectDatabase($db)->selectCollection($collection);
+    }
+}