Merge pull request #36 from aloware/fix/number-of-processed-jobs

Fix Number of Processed Jobs

Author: hamed-aloware

src/FairQueueServiceProvider.php

@@ -56,7 +56,6 @@ public function boot(): void
         $this->registerRoutes();
         $this->registerResources();
         $this->publishAssets();
-        $this->registerQueueEvents();
 
         $this->callAfterResolving(Schedule::class, function (Schedule $schedule) {
             if(config('fair-queue.recover_lost_jobs.enabled')) {
@@ -90,34 +89,6 @@ protected function registerRoutes(): void
         });
     }
 
-    /**
-     * Register the FairQueue Queue Events.
-     *
-     * @return void
-     */
-    protected function registerQueueEvents(): void
-    {
-        Queue::after(function (JobProcessed $event) {
-            $redis = FairQueue::getConnection();
-            $payload = $event->job->payload();
-            if (!isset($payload['data']) || !isset($payload['data']['command'])) {
-                return;
-            }
-            $command = unserialize($payload['data']['command']);
-            if (!$command instanceof FairSignalJob) {
-                return;
-            }
-            $queue = $command->queue;
-            $partition = $command->partition;
-            $past_minute_key = $this->partitionProcessedJobsInPastMinutesKey($queue, $partition, 1);
-            $past_20minute_key = $this->partitionProcessedJobsInPastMinutesKey($queue, $partition, 20);
-            $past_60minute_key = $this->partitionProcessedJobsInPastMinutesKey($queue, $partition, 60);
-            $redis->zadd($past_minute_key, now()->getPreciseTimestamp(3), $payload['id']);
-            $redis->zadd($past_20minute_key, now()->getPreciseTimestamp(3), $payload['id']);
-            $redis->zadd($past_60minute_key, now()->getPreciseTimestamp(3), $payload['id']);
-        });
-    }
-
     /**
      * Register the FairQueue resources.
      *

src/FairSignalJob.php

@@ -2,7 +2,9 @@
 
 namespace Aloware\FairQueue;
 
+use Aloware\FairQueue\Facades\FairQueue;
 use Aloware\FairQueue\Interfaces\RepositoryInterface;
+use Aloware\FairQueue\Repositories\RedisKeys;
 use Illuminate\Bus\Queueable;
 use Illuminate\Contracts\Queue\ShouldQueue;
 use Illuminate\Foundation\Bus\Dispatchable;
@@ -11,7 +13,7 @@
 
 class FairSignalJob implements ShouldQueue
 {
-    use Dispatchable, InteractsWithQueue, Queueable;
+    use Dispatchable, InteractsWithQueue, Queueable, RedisKeys;
 
     public $partition;
 
@@ -63,6 +65,10 @@ public function handle()
             }
 
             $job->handle();
+
+            // Update Fair Queue Stats
+            $this->updateStats($job->uuid);
+
         } catch (\Throwable $e) {
             printf('[%s] %s' . PHP_EOL, get_class($job), $e->getMessage());
 
@@ -171,4 +177,18 @@ private function selectPartition($repository, $partitionsMethod = 'partitions')
 
         return $partitions[$partitionIndex];
     }
+
+    public function updateStats($uuid)
+    {
+        $redis = FairQueue::getConnection();
+        $queue = $this->queue;
+        $partition = $this->partition;
+
+        $past_minute_key = $this->partitionProcessedJobsInPastMinutesKey($queue, $partition, 1);
+        $past_20minute_key = $this->partitionProcessedJobsInPastMinutesKey($queue, $partition, 20);
+        $past_60minute_key = $this->partitionProcessedJobsInPastMinutesKey($queue, $partition, 60);
+        $redis->zadd($past_minute_key, now()->getPreciseTimestamp(3), $uuid);
+        $redis->zadd($past_20minute_key, now()->getPreciseTimestamp(3), $uuid);
+        $redis->zadd($past_60minute_key, now()->getPreciseTimestamp(3), $uuid);
+    }
 }