Added batch queue support.

Author: th3third

src/Bus/MongoDatabaseBatchRepository.php

@@ -0,0 +1,36 @@
+<?php
+
+    namespace Jenssegers\Mongodb\Bus;
+
+    use Closure;
+    use Illuminate\Bus\DatabaseBatchRepository;
+
+    class MongoDatabaseBatchRepository extends DatabaseBatchRepository {
+        protected function updateAtomicValues(string $batchId, Closure $callback) {
+            return $this->connection->transaction(function () use ($batchId, $callback) {
+                $batch = (object)$this->connection->table($this->table)->where('id', $batchId)
+                            ->lockForUpdate()
+                            ->first();
+
+                return is_null($batch) ? [] : tap($callback($batch), function ($values) use ($batchId) {
+                    $this->connection->table($this->table)->where('id', $batchId)->update($values);
+                });
+            });
+        }
+
+        public function incrementTotalJobs(string $batchId, int $amount) {
+            $this->connection->table($this->table)->where('id', $batchId)->update([
+                '$set' => [
+                    'finished_at' => null
+                ],
+                '$inc' => [
+                    'total_jobs' => $amount,
+                    'pending_jobs' => $amount
+                ]
+            ]);
+        }
+
+        protected function toBatch($batch) {
+            return parent::toBatch((object)$batch);
+        }
+    }

src/MongodbServiceProvider.php

@@ -2,7 +2,9 @@
 
 namespace Jenssegers\Mongodb;
 
+use Illuminate\Bus\{BatchFactory, DatabaseBatchRepository};
 use Illuminate\Support\ServiceProvider;
+use Jenssegers\Mongodb\Bus\MongoDatabaseBatchRepository;
 use Jenssegers\Mongodb\Eloquent\Model;
 use Jenssegers\Mongodb\Queue\MongoConnector;
 
@@ -38,5 +40,13 @@ public function register()
                 return new MongoConnector($this->app['db']);
             });
         });
+
+        $this->app->extend(DatabaseBatchRepository::class, function ($command, $app) {
+            return new MongoDatabaseBatchRepository(
+                $app->make(BatchFactory::class),
+                $app->make('db')->connection($app->config->get('queue.batching.database')),
+                $app->config->get('queue.batching.table', 'job_batches')
+            );
+        });
     }
 }