mgdigital / BusQue

src/Predis/PredisAdapter.php

Spacing +39 added lines, -39 removed lines

@@ -26,16 +26,16 @@ 
     public function queueCommand(string $queueName, string $id, string $serialized)
     {
         list ($isReserved) = $this->client->pipeline(
-            function (ClientContextInterface $client) use ($queueName, $id, $serialized) {
+            function(ClientContextInterface $client) use ($queueName, $id, $serialized) {
                 self::cIsCommandIdReserved($client, $queueName, $id);
                 self::cStoreCommand($client, $queueName, $id, $serialized);
             }
         );
         if (!$isReserved) {
-            $this->client->pipeline(function (ClientContextInterface $client) use ($queueName, $id) {
+            $this->client->pipeline(function(ClientContextInterface $client) use ($queueName, $id) {
                 self::cReserveCommandId($client, $queueName, $id);
                 self::cUpdateCommandStatus($client, $queueName, $id, self::STATUS_QUEUED);
-                $client->lpush(":{$queueName}:queue", [$id]);
+                $client->lpush(":{$queueName}:queue", [ $id ]);
             });
         }
     }
@@ -59,10 +59,10 @@ 
             return $this->awaitCommand($queueName, $timeout);
         }
         /* @var $id string */
-        list(, $serialized) = $this->client->pipeline(function (ClientContextInterface $client) use ($queueName, $id) {
+        list(, $serialized) = $this->client->pipeline(function(ClientContextInterface $client) use ($queueName, $id) {
             self::cUpdateCommandStatus($client, $queueName, $id, self::STATUS_IN_PROGRESS);
             self::cRetrieveCommand($client, $queueName, $id);
-            self::cReleaseReservedCommandIds($client, $queueName, [$id]);
+            self::cReleaseReservedCommandIds($client, $queueName, [ $id ]);
         });
         return new ReceivedCommand($queueName, $id, $serialized);
     }
@@ -74,21 +74,21 @@ 
 
     public function setCommandCompleted(string $queueName, string $id)
     {
-        $this->client->pipeline(function (ClientContextInterface $client) use ($queueName, $id) {
+        $this->client->pipeline(function(ClientContextInterface $client) use ($queueName, $id) {
             self::cEndCommand($client, $queueName, $id, self::STATUS_COMPLETED);
         });
     }
 
     public function setCommandFailed(string $queueName, string $id)
     {
-        $this->client->pipeline(function (ClientContextInterface $client) use ($queueName, $id) {
+        $this->client->pipeline(function(ClientContextInterface $client) use ($queueName, $id) {
             self::cEndCommand($client, $queueName, $id, self::STATUS_FAILED);
         });
     }
 
     public function putQueue(string $queueName)
     {
-        $this->client->pipeline(function (ClientContextInterface $client) use ($queueName) {
+        $this->client->pipeline(function(ClientContextInterface $client) use ($queueName) {
             self::cAddQueue($client, $queueName);
         });
     }
@@ -134,18 +134,18 @@ 
 
     public function deleteQueue(string $queueName)
     {
-        $this->client->pipeline(function (ClientContextInterface $client) use ($queueName) {
+        $this->client->pipeline(function(ClientContextInterface $client) use ($queueName) {
             self::cDeleteQueue($client, $queueName);
         });
     }
 
     public function purgeCommand(string $queueName, string $id)
     {
-        $this->client->pipeline(function (ClientContextInterface $client) use ($queueName, $id) {
-            $client->hdel(":{$queueName}:command_store", [$id]);
-            $client->hdel(":{$queueName}:command_status", [$id]);
-            self::cReleaseReservedCommandIds($client, $queueName, [$id]);
-            $json = json_encode([$queueName, $id]);
+        $this->client->pipeline(function(ClientContextInterface $client) use ($queueName, $id) {
+            $client->hdel(":{$queueName}:command_store", [ $id ]);
+            $client->hdel(":{$queueName}:command_status", [ $id ]);
+            self::cReleaseReservedCommandIds($client, $queueName, [ $id ]);
+            $json = json_encode([ $queueName, $id ]);
             $client->lrem(":{$queueName}:queue", 1, $id);
             $client->lrem(":{$queueName}:consuming", 1, $id);
             $client->zrem(':schedule', $json);
@@ -155,18 +155,18 @@ 
     public function scheduleCommand(string $queueName, string $id, string $serialized, \DateTime $dateTime)
     {
         list ($isReserved) = $this->client->pipeline(
-            function (ClientContextInterface $client) use ($queueName, $id, $serialized) {
+            function(ClientContextInterface $client) use ($queueName, $id, $serialized) {
                 self::cIsCommandIdReserved($client, $queueName, $id);
                 self::cStoreCommand($client, $queueName, $id, $serialized);
             }
         );
         if (!$isReserved) {
             $this->client->pipeline(
-                function (ClientContextInterface $client) use ($queueName, $id, $dateTime) {
+                function(ClientContextInterface $client) use ($queueName, $id, $dateTime) {
                     self::cReserveCommandId($client, $queueName, $id);
                     self::cUpdateCommandStatus($client, $queueName, $id, self::STATUS_SCHEDULED);
-                    $json = json_encode([$queueName, $id]);
-                    $client->zadd(':schedule', [$json => $dateTime->getTimestamp()]);
+                    $json = json_encode([ $queueName, $id ]);
+                    $client->zadd(':schedule', [ $json => $dateTime->getTimestamp() ]);
                 }
             );
         }
@@ -187,13 +187,13 @@ 
         foreach ($queueNames as $queueName) {
             $result = $this->client->zrangebyscore(':schedule', $lowScore, $highScore);
             if (!empty($result)) {
-                $this->client->pipeline(function (ClientContextInterface $client) use ($result, $queueName) {
-                    $idsToRelease = [];
+                $this->client->pipeline(function(ClientContextInterface $client) use ($result, $queueName) {
+                    $idsToRelease = [ ];
                     foreach ($result as $json => $score) {
                         list($thisQueueName, $id) = json_decode($json, true);
                         if ($thisQueueName === $queueName) {
                             $client->zrem(':schedule', $json);
-                            $idsToRelease[] = $id;
+                            $idsToRelease[ ] = $id;
                         }
                     }
                     self::cReleaseReservedCommandIds($client, $queueName, $idsToRelease);
@@ -213,21 +213,21 @@ 
             $start = $startTime->getTimestamp();
         }
         $result = $this->client->zrangebyscore(':schedule', $start, $now->getTimestamp(), [
-            'limit' => [0, $limit],
+            'limit' => [ 0, $limit ],
             'withscores' => true,
         ]);
-        $commands = [];
-        if ($result !== []) {
-            $queueNamesById = $idsByJson = [];
+        $commands = [ ];
+        if ($result !== [ ]) {
+            $queueNamesById = $idsByJson = [ ];
             $pipelineReturn = $this->client->pipeline(
-                function (ClientContextInterface $client) use ($result, &$queueNamesById, &$idsByJson) {
-                    $idsByQueueName = [];
+                function(ClientContextInterface $client) use ($result, &$queueNamesById, &$idsByJson) {
+                    $idsByQueueName = [ ];
                     foreach ($result as $json => $score) {
                         list($queueName, $id) = json_decode($json, true);
                         self::cRetrieveCommand($client, $queueName, $id);
-                        $idsByQueueName[$queueName][] = $id;
-                        $queueNamesById[$id] = $queueName;
-                        $idsByJson[$json] = $id;
+                        $idsByQueueName[ $queueName ][ ] = $id;
+                        $queueNamesById[ $id ] = $queueName;
+                        $idsByJson[ $json ] = $id;
                     }
                     $client->zrem(':schedule', array_keys($result));
                     foreach ($idsByQueueName as $queueName => $ids) {
@@ -236,12 +236,12 @@ 
                 }
             );
             foreach (array_keys($result) as $index => $json) {
-                $id = $idsByJson[$json];
-                $commands[] = new ReceivedScheduledCommand(
-                    $queueNamesById[$id],
+                $id = $idsByJson[ $json ];
+                $commands[ ] = new ReceivedScheduledCommand(
+                    $queueNamesById[ $id ],
                     $id,
-                    $pipelineReturn[$index],
-                    new \DateTime('@' . $result[$json])
+                    $pipelineReturn[ $index ],
+                    new \DateTime('@'.$result[ $json ])
                 );
             }
         }
@@ -261,14 +261,14 @@ 
 
     private static function cReserveCommandId($client, string $queueName, string $id)
     {
-        $client->sadd(":{$queueName}:queue_ids", [$id]);
+        $client->sadd(":{$queueName}:queue_ids", [ $id ]);
     }
 
     private static function cEndCommand($client, string $queueName, string $id, string $status)
     {
         self::cUpdateCommandStatus($client, $queueName, $id, $status);
-        self::cReleaseReservedCommandIds($client, $queueName, [$id]);
-        $client->srem(":{$queueName}:queue_ids", [$id]);
+        self::cReleaseReservedCommandIds($client, $queueName, [ $id ]);
+        $client->srem(":{$queueName}:queue_ids", [ $id ]);
         $client->lrem(":{$queueName}:consuming", 1, $id);
     }
 
@@ -289,7 +289,7 @@ 
 
     private static function cAddQueue($client, string $queueName)
     {
-        $client->sadd(':queues', [$queueName]);
+        $client->sadd(':queues', [ $queueName ]);
     }
 
     private static function cEmptyQueue($client, string $queueName)