mgdigital / BusQue

src/SchedulerWorker.php

Spacing +3 added lines, -3 removed lines

@@ -9,7 +9,7 @@ 
 
     private $implementation;
 
-    public function __construct(Implementation $implementation)
+    public function __construct (Implementation $implementation)
     {
         $this->implementation = $implementation;
     }
@@ -21,7 +21,7 @@ 
      * @param int $uSleepTime The number of microseconds to usleep between each query to the scheduler.
      * @param \DateInterval|null $expiry The expiry interval for an overdue unqueued command.
      */
-    public function work(
+    public function work (
         int $limit = null,
         int $throttle = self::DEFAULT_THROTTLE,
         int $time = null,
@@ -44,7 +44,7 @@ 
         }
     }
 
-    private function iterate(int $throttle, \DateInterval $expiry = null): int
+    private function iterate (int $throttle, \DateInterval $expiry = null): int
     {
         $count = 0;
         $now = $this->implementation->getClock()->getTime();

src/Logging/LoggingErrorHandler.php

Spacing +3 added lines, -3 removed lines

@@ -10,12 +10,12 @@ 
 
     private $logger;
 
-    public function __construct(LoggerInterface $logger)
+    public function __construct (LoggerInterface $logger)
     {
         $this->logger = $logger;
     }
 
-    public function handleCommandError($command, \Throwable $error)
+    public function handleCommandError ($command, \Throwable $error)
     {
         $this->logger->error($error->getMessage(), [
             'command' => $command,
@@ -23,7 +23,7 @@ 
         ]);
     }
 
-    public function handleUnserializationError(
+    public function handleUnserializationError (
         string $queueName,
         string $commandId,
         string $serialized,

src/BusQue.php

Spacing +18 added lines, -18 removed lines

@@ -7,93 +7,93 @@
 
     private $implementation;
 
-    public function __construct(Implementation $implementation)
+    public function __construct (Implementation $implementation)
     {
         $this->implementation = $implementation;
     }
 
-    public function getQueueName($command): string
+    public function getQueueName ($command): string
     {
         return $this->implementation->getQueueResolver()->resolveQueueName($command);
     }
 
-    public function serializeCommand($command): string
+    public function serializeCommand ($command): string
     {
         return $this->implementation->getCommandSerializer()->serialize($command);
     }
 
-    public function unserializeCommand(string $serialized)
+    public function unserializeCommand (string $serialized)
     {
         return $this->implementation->getCommandSerializer()->unserialize($serialized);
     }
 
-    public function generateCommandId($command): string
+    public function generateCommandId ($command): string
     {
         return $this->implementation->getCommandIdGenerator()->generateId($command);
     }
 
-    public function queueCommand($command, string $commandId = null)
+    public function queueCommand ($command, string $commandId = null)
     {
         $this->implementation->getCommandBusAdapter()->handle(new QueuedCommand($command, $commandId));
     }
 
-    public function scheduleCommand($command, \DateTime $dateTime, string $commandId = null)
+    public function scheduleCommand ($command, \DateTime $dateTime, string $commandId = null)
     {
         $this->implementation->getCommandBusAdapter()->handle(new ScheduledCommand($command, $dateTime, $commandId));
     }
 
-    public function getCommandStatus(string $queueName, string $commandId): string
+    public function getCommandStatus (string $queueName, string $commandId): string
     {
         return $this->implementation->getQueueAdapter()->getCommandStatus($queueName, $commandId);
     }
 
-    public function getQueuedCount(string $queueName): int
+    public function getQueuedCount (string $queueName): int
     {
         return $this->implementation->getQueueAdapter()->getQueuedCount($queueName);
     }
 
-    public function purgeCommand(string $queueName, string $commandId)
+    public function purgeCommand (string $queueName, string $commandId)
     {
         $this->implementation->getQueueAdapter()->purgeCommand($queueName, $commandId);
     }
 
-    public function emptyQueue(string $queueName)
+    public function emptyQueue (string $queueName)
     {
         $this->implementation->getQueueAdapter()->emptyQueue($queueName);
     }
 
-    public function listQueues(): array
+    public function listQueues (): array
     {
         return $this->implementation->getQueueAdapter()->getQueueNames();
     }
 
-    public function listQueuedIds(string $queueName, int $offset = 0, int $limit = 10): array
+    public function listQueuedIds (string $queueName, int $offset = 0, int $limit = 10): array
     {
         return $this->implementation->getQueueAdapter()->getQueuedIds($queueName, $offset, $limit);
     }
 
-    public function getInProgressCount(string $queueName): int
+    public function getInProgressCount (string $queueName): int
     {
         return $this->implementation->getQueueAdapter()->getConsumingCount($queueName);
     }
 
-    public function listInProgressIds(string $queueName, int $offset = 0, int $limit = 10): array
+    public function listInProgressIds (string $queueName, int $offset = 0, int $limit = 10): array
     {
         return $this->implementation->getQueueAdapter()->getConsumingIds($queueName, $offset, $limit);
     }
 
-    public function getCommand(string $queueName, string $id)
+    public function getCommand (string $queueName, string $id)
     {
         $serialized = $this->implementation->getQueueAdapter()->readCommand($queueName, $id);
         return $this->unserializeCommand($serialized);
     }
 
-    public function workQueue(string $queueName, int $n = null, int $time = null)
+    public function workQueue (string $queueName, int $n = null, int $time = null)
     {
         (new QueueWorker($this->implementation))->work($queueName, $n, $time);
     }
 
-    public function workSchedule(int $n = null, int $time = null)
+    public function workSchedule (int $n = null, int $time = null)
     {
         (new SchedulerWorker($this->implementation))->work($n, $time);
     }

src/ClockInterface.php

Spacing +1 added lines, -1 removed lines

@@ -5,5 +5,5 @@
 interface ClockInterface
 {
 
-    public function getTime(): \DateTime;
+    public function getTime (): \DateTime;
 }

src/QueueWorker.php

Spacing +3 added lines, -3 removed lines

@@ -9,12 +9,12 @@ 
 
     private $implementation;
 
-    public function __construct(Implementation $implementation)
+    public function __construct (Implementation $implementation)
     {
         $this->implementation = $implementation;
     }
 
-    public function work(string $queueName, int $n = null, int $time = null)
+    public function work (string $queueName, int $n = null, int $time = null)
     {
         $stopwatchStart = time();
         while ($n === null || $n > 0) {
@@ -26,7 +26,7 @@ 
         }
     }
 
-    private function iterate(string $queueName, int $time = null)
+    private function iterate (string $queueName, int $time = null)
     {
         try {
             $received = $this->implementation->getQueueAdapter()

src/CommandSerializerInterface.php

Spacing +2 added lines, -2 removed lines

@@ -5,11 +5,11 @@
 interface CommandSerializerInterface
 {
 
-    public function serialize($command): string;
+    public function serialize ($command): string;
 
     /**
      * @param string $serialized
      * @return mixed A command object which can be handled by the commandbus.
      */
-    public function unserialize(string $serialized);
+    public function unserialize (string $serialized);
 }

src/Predis/PredisAdapter.php

Spacing +40 added lines, -40 removed lines

@@ -18,15 +18,15 @@ 
 
     private $client;
 
-    public function __construct(Client $client)
+    public function __construct (Client $client)
     {
         $this->client = $client;
     }
 
-    public function queueCommand(string $queueName, string $id, string $serialized)
+    public function queueCommand (string $queueName, string $id, string $serialized)
     {
         if (!$this->storeCommandAndCheckIfIdReserved($queueName, $id, $serialized)) {
-            $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 ]);
@@ -34,7 +34,7 @@ 
         }
     }
 
-    public function awaitCommand(string $queueName, int $timeout = null): ReceivedCommand
+    public function awaitCommand (string $queueName, int $timeout = null): ReceivedCommand
     {
         $stopwatchStart = time();
         $this->client->ping();
@@ -53,7 +53,7 @@ 
             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 ]);
@@ -61,58 +61,58 @@ 
         return new ReceivedCommand($queueName, $id, $serialized);
     }
 
-    public function getCommandStatus(string $queueName, string $id): string
+    public function getCommandStatus (string $queueName, string $id): string
     {
         return $this->client->hget(":{$queueName}:command_status", $id) ?? self::STATUS_NOT_FOUND;
     }
 
-    public function setCommandCompleted(string $queueName, string $id)
+    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)
+    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)
+    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);
         });
     }
 
-    public function getQueueNames(): array
+    public function getQueueNames (): array
     {
         return $this->client->smembers(':queues');
     }
 
-    public function getQueuedCount(string $queueName): int
+    public function getQueuedCount (string $queueName): int
     {
         return $this->client->llen(":{$queueName}:queue");
     }
 
-    public function getQueuedIds(string $queueName, int $offset = 0, int $limit = 10): array
+    public function getQueuedIds (string $queueName, int $offset = 0, int $limit = 10): array
     {
         return $this->client->lrange(":{$queueName}:queue", $offset, $limit);
     }
 
-    public function getConsumingCount(string $queueName): int
+    public function getConsumingCount (string $queueName): int
     {
         return $this->client->llen(":{$queueName}:consuming");
     }
 
-    public function getConsumingIds(string $queueName, int $offset = 0, int $limit = 10): array
+    public function getConsumingIds (string $queueName, int $offset = 0, int $limit = 10): array
     {
         return $this->client->lrange(":{$queueName}:consuming", $offset, $limit);
     }
 
-    public function readCommand(string $queueName, string $id): string
+    public function readCommand (string $queueName, string $id): string
     {
         $serialized = self::cRetrieveCommand($this->client, $queueName, $id);
         if ($serialized === null) {
@@ -121,21 +121,21 @@ 
         return $serialized;
     }
 
-    public function emptyQueue(string $queueName)
+    public function emptyQueue (string $queueName)
     {
         self::cEmptyQueue($this->client, $queueName);
     }
 
-    public function deleteQueue(string $queueName)
+    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)
+    public function purgeCommand (string $queueName, string $id)
     {
-        $this->client->pipeline(function (ClientContextInterface $client) use ($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 ]);
@@ -146,11 +146,11 @@ 
         });
     }
 
-    public function scheduleCommand(string $queueName, string $id, string $serialized, \DateTime $dateTime)
+    public function scheduleCommand (string $queueName, string $id, string $serialized, \DateTime $dateTime)
     {
         if (!$this->storeCommandAndCheckIfIdReserved($queueName, $id, $serialized)) {
             $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 ]);
@@ -160,12 +160,12 @@ 
         }
     }
 
-    public function cancelScheduledCommand(string $queueName, string $id)
+    public function cancelScheduledCommand (string $queueName, string $id)
     {
         $this->purgeCommand($queueName, $id);
     }
 
-    public function clearSchedule(array $queueNames = null, \DateTime $start = null, \DateTime $end = null)
+    public function clearSchedule (array $queueNames = null, \DateTime $start = null, \DateTime $end = null)
     {
         $lowScore = $start ? $start->getTimestamp() : 0;
         $highScore = $end ? $end->getTimestamp() : -1;
@@ -175,7 +175,7 @@ 
         foreach ($queueNames as $queueName) {
             $result = $this->client->zrangebyscore(':schedule', $lowScore, $highScore);
             if (!empty($result)) {
-                $this->client->pipeline(function (ClientContextInterface $client) use ($result, $queueName) {
+                $this->client->pipeline(function(ClientContextInterface $client) use ($result, $queueName) {
                     $idsToRelease = [ ];
                     foreach ($result as $json => $score) {
                         list($thisQueueName, $id) = json_decode($json, true);
@@ -190,7 +190,7 @@ 
         }
     }
 
-    public function receiveDueCommands(
+    public function receiveDueCommands (
         \DateTime $now,
         int $limit = SchedulerWorker::DEFAULT_THROTTLE,
         \DateTime $startTime = null
@@ -208,7 +208,7 @@ 
         if ($result !== [ ]) {
             $queueNamesById = $idsByJson = [ ];
             $pipelineReturn = $this->client->pipeline(
-                function (ClientContextInterface $client) use ($result, &$queueNamesById, &$idsByJson) {
+                function(ClientContextInterface $client) use ($result, &$queueNamesById, &$idsByJson) {
                     $idsByQueueName = [ ];
                     foreach ($result as $json => $score) {
                         list($queueName, $id) = json_decode($json, true);
@@ -236,10 +236,10 @@ 
         return $commands;
     }
 
-    private function storeCommandAndCheckIfIdReserved(string $queueName, string $id, string $serialized): bool
+    private function storeCommandAndCheckIfIdReserved (string $queueName, string $id, string $serialized): bool
     {
         list ($isReserved) = $this->client->pipeline(
-            function (ClientContextInterface $client) use ($queueName, $id, $serialized) {
+            function(ClientContextInterface $client) use ($queueName, $id, $serialized) {
                 $client->sismember(":{$queueName}:queue_ids", $id);
                 $client->hset(":{$queueName}:command_store", $id, $serialized);
                 self::cAddQueue($client, $queueName);
@@ -248,17 +248,17 @@ 
         return $isReserved;
     }
 
-    private static function cRetrieveCommand($client, string $queueName, string $id)
+    private static function cRetrieveCommand ($client, string $queueName, string $id)
     {
         return $client->hget(":{$queueName}:command_store", $id);
     }
 
-    private static function cReserveCommandId($client, string $queueName, string $id)
+    private static function cReserveCommandId ($client, string $queueName, string $id)
     {
         $client->sadd(":{$queueName}:queue_ids", [ $id ]);
     }
 
-    private static function cEndCommand($client, string $queueName, string $id, string $status)
+    private static function cEndCommand ($client, string $queueName, string $id, string $status)
     {
         self::cUpdateCommandStatus($client, $queueName, $id, $status);
         self::cReleaseReservedCommandIds($client, $queueName, [ $id ]);
@@ -266,22 +266,22 @@ 
         $client->lrem(":{$queueName}:consuming", 1, $id);
     }
 
-    private static function cReleaseReservedCommandIds($client, string $queueName, array $ids)
+    private static function cReleaseReservedCommandIds ($client, string $queueName, array $ids)
     {
         $client->srem(":{$queueName}:queue_ids", $ids);
     }
 
-    private static function cUpdateCommandStatus($client, string $queueName, string $id, string $status)
+    private static function cUpdateCommandStatus ($client, string $queueName, string $id, string $status)
     {
         $client->hset(":{$queueName}:command_status", $id, $status);
     }
 
-    private static function cAddQueue($client, string $queueName)
+    private static function cAddQueue ($client, string $queueName)
     {
         $client->sadd(':queues', [ $queueName ]);
     }
 
-    private static function cEmptyQueue($client, string $queueName)
+    private static function cEmptyQueue ($client, string $queueName)
     {
         $client->del([
             ":{$queueName}:queue",
@@ -292,7 +292,7 @@ 
         ]);
     }
 
-    private static function cDeleteQueue($client, string $queueName)
+    private static function cDeleteQueue ($client, string $queueName)
     {
         self::cEmptyQueue($client, $queueName);
         $client->srem(':queues', $queueName);

src/QueueResolver/SimpleQueueResolver.php

Spacing +2 added lines, -2 removed lines

@@ -9,12 +9,12 @@
 
     private $queueName;
 
-    public function __construct(string $queueName)
+    public function __construct (string $queueName)
     {
         $this->queueName = $queueName;
     }
 
-    public function resolveQueueName($command): string
+    public function resolveQueueName ($command): string
     {
         return $this->queueName;
     }

src/QueueResolver/QueueVoterInterface.php

Spacing +1 added lines, -1 removed lines

@@ -9,5 +9,5 @@
      * @param mixed $command
      * @return QueueVote|null
      */
-    public function getVote($command);
+    public function getVote ($command);
 }