QueueingDispatcher.php000066400000001263000000000000011013 0ustar00container = $container; $this->queueResolver = $queueResolver; $this->pipeline = new Pipeline($container); } /** * Dispatch a command to its appropriate handler. * * @param mixed $command * @return mixed */ public function dispatch($command) { return $this->queueResolver && $this->commandShouldBeQueued($command) ? $this->dispatchToQueue($command) : $this->dispatchNow($command); } /** * Dispatch a command to its appropriate handler in the current process. * * Queueable jobs will be dispatched to the "sync" queue. * * @param mixed $command * @param mixed $handler * @return mixed */ public function dispatchSync($command, $handler = null) { if ($this->queueResolver && $this->commandShouldBeQueued($command) && method_exists($command, 'onConnection')) { return $this->dispatchToQueue($command->onConnection('sync')); } return $this->dispatchNow($command, $handler); } /** * Dispatch a command to its appropriate handler in the current process without using the synchronous queue. * * @param mixed $command * @param mixed $handler * @return mixed */ public function dispatchNow($command, $handler = null) { $uses = class_uses_recursive($command); if (isset($uses[InteractsWithQueue::class], $uses[Queueable::class]) && ! $command->job) { $command->setJob(new SyncJob($this->container, json_encode([]), 'sync', 'sync')); } if ($handler || $handler = $this->getCommandHandler($command)) { $callback = function ($command) use ($handler) { $method = method_exists($handler, 'handle') ? 'handle' : '__invoke'; return $handler->{$method}($command); }; } else { $callback = function ($command) { $method = method_exists($command, 'handle') ? 'handle' : '__invoke'; return $this->container->call([$command, $method]); }; } return $this->pipeline->send($command)->through($this->pipes)->then($callback); } /** * Attempt to find the batch with the given ID. * * @return \Illuminate\Bus\Batch|null */ public function findBatch(string $batchId) { return $this->container->make(BatchRepository::class)->find($batchId); } /** * Create a new batch of queueable jobs. * * @param \Illuminate\Support\Collection|mixed $jobs * @return \Illuminate\Bus\PendingBatch */ public function batch($jobs) { return new PendingBatch($this->container, Collection::wrap($jobs)); } /** * Create a new chain of queueable jobs. * * @param \Illuminate\Support\Collection|array|null $jobs * @return \Illuminate\Foundation\Bus\PendingChain */ public function chain($jobs = null) { $jobs = Collection::wrap($jobs); $jobs = ChainedBatch::prepareNestedBatches($jobs); return new PendingChain($jobs->shift(), $jobs->toArray()); } /** * Determine if the given command has a handler. * * @param mixed $command * @return bool */ public function hasCommandHandler($command) { return array_key_exists(get_class($command), $this->handlers); } /** * Retrieve the handler for a command. * * @param mixed $command * @return mixed */ public function getCommandHandler($command) { if ($this->hasCommandHandler($command)) { return $this->container->make($this->handlers[get_class($command)]); } return false; } /** * Determine if the given command should be queued. * * @param mixed $command * @return bool */ protected function commandShouldBeQueued($command) { return $command instanceof ShouldQueue; } /** * Dispatch a command to its appropriate handler behind a queue. * * @param mixed $command * @return mixed * * @throws \RuntimeException */ public function dispatchToQueue($command) { $connection = $command->connection ?? null; $queue = ($this->queueResolver)($connection); if (! $queue instanceof Queue) { throw new RuntimeException('Queue resolver did not return a Queue implementation.'); } if (method_exists($command, 'queue')) { return $command->queue($queue, $command); } return $this->pushCommandToQueue($queue, $command); } /** * Push the command onto the given queue instance. * * @param \Illuminate\Contracts\Queue\Queue $queue * @param mixed $command * @return mixed */ protected function pushCommandToQueue($queue, $command) { if (isset($command->delay)) { return $queue->later($command->delay, $command, queue: $command->queue ?? null); } return $queue->push($command, queue: $command->queue ?? null); } /** * Dispatch a command to its appropriate handler after the current process. * * @param mixed $command * @param mixed $handler * @return void */ public function dispatchAfterResponse($command, $handler = null) { if (! $this->allowsDispatchingAfterResponses) { $this->dispatchSync($command); return; } $this->container->terminating(function () use ($command, $handler) { $this->dispatchSync($command, $handler); }); } /** * Set the pipes through which commands should be piped before dispatching. * * @return $this */ public function pipeThrough(array $pipes) { $this->pipes = $pipes; return $this; } /** * Map a command to a handler. * * @return $this */ public function map(array $map) { $this->handlers = array_merge($this->handlers, $map); return $this; } /** * Allow dispatching after responses. * * @return $this */ public function withDispatchingAfterResponses() { $this->allowsDispatchingAfterResponses = true; return $this; } /** * Disable dispatching after responses. * * @return $this */ public function withoutDispatchingAfterResponses() { $this->allowsDispatchingAfterResponses = false; return $this; } } BusServiceProvider.php000066400000006523000000000000011013 0ustar00app->singleton(Dispatcher::class, function ($app) { return new Dispatcher($app, function ($connection = null) use ($app) { return $app[QueueFactoryContract::class]->connection($connection); }); }); $this->registerBatchServices(); $this->app->alias( Dispatcher::class, DispatcherContract::class ); $this->app->alias( Dispatcher::class, QueueingDispatcherContract::class ); } /** * Register the batch handling services. * * @return void */ protected function registerBatchServices() { $this->app->singleton(BatchRepository::class, function ($app) { $driver = $app->config->get('queue.batching.driver', 'database'); return $driver === 'dynamodb' ? $app->make(DynamoBatchRepository::class) : $app->make(DatabaseBatchRepository::class); }); $this->app->singleton(DatabaseBatchRepository::class, function ($app) { return new DatabaseBatchRepository( $app->make(BatchFactory::class), $app->make('db')->connection($app->config->get('queue.batching.database')), $app->config->get('queue.batching.table', 'job_batches') ); }); $this->app->singleton(DynamoBatchRepository::class, function ($app) { $config = $app->config->get('queue.batching'); $dynamoConfig = [ 'region' => $config['region'], 'version' => 'latest', 'endpoint' => $config['endpoint'] ?? null, ]; if (! empty($config['key']) && ! empty($config['secret'])) { $dynamoConfig['credentials'] = Arr::only($config, ['key', 'secret']); if (! empty($config['token'])) { $dynamoConfig['credentials']['token'] = $config['token']; } } return new DynamoBatchRepository( $app->make(BatchFactory::class), new DynamoDbClient($dynamoConfig), $app->config->get('app.name'), $app->config->get('queue.batching.table', 'job_batches'), ttl: $app->config->get('queue.batching.ttl', null), ttlAttribute: $app->config->get('queue.batching.ttl_attribute', 'ttl'), ); }); } /** * Get the services provided by the provider. * * @return array */ public function provides() { return [ Dispatcher::class, DispatcherContract::class, QueueingDispatcherContract::class, BatchRepository::class, DatabaseBatchRepository::class, ]; } } BatchFactory.php000066400000003300000000000000007565 0ustar00queue = $queue; } /** * Create a new batch instance. * * @param \Illuminate\Bus\BatchRepository $repository * @param string $id * @param string $name * @param int $totalJobs * @param int $pendingJobs * @param int $failedJobs * @param array $failedJobIds * @param array $options * @param \Carbon\CarbonImmutable $createdAt * @param \Carbon\CarbonImmutable|null $cancelledAt * @param \Carbon\CarbonImmutable|null $finishedAt * @return \Illuminate\Bus\Batch */ public function make(BatchRepository $repository, string $id, string $name, int $totalJobs, int $pendingJobs, int $failedJobs, array $failedJobIds, array $options, CarbonImmutable $createdAt, ?CarbonImmutable $cancelledAt, ?CarbonImmutable $finishedAt) { return new Batch($this->queue, $repository, $id, $name, $totalJobs, $pendingJobs, $failedJobs, $failedJobIds, $options, $createdAt, $cancelledAt, $finishedAt); } } PrunableBatchRepository.php000066400000000476000000000000012041 0ustar00queue = $queue; $this->repository = $repository; $this->id = $id; $this->name = $name; $this->totalJobs = $totalJobs; $this->pendingJobs = $pendingJobs; $this->failedJobs = $failedJobs; $this->failedJobIds = $failedJobIds; $this->options = $options; $this->createdAt = $createdAt; $this->cancelledAt = $cancelledAt; $this->finishedAt = $finishedAt; } /** * Get a fresh instance of the batch represented by this ID. * * @return self */ public function fresh() { return $this->repository->find($this->id); } /** * Add additional jobs to the batch. * * @param \Illuminate\Support\Enumerable|object|array $jobs * @return self */ public function add($jobs) { $count = 0; $jobs = Collection::wrap($jobs)->map(function ($job) use (&$count) { $job = $job instanceof Closure ? CallQueuedClosure::create($job) : $job; if (is_array($job)) { $count += count($job); $chain = $this->prepareBatchedChain($job); return $chain->first() ->allOnQueue($this->options['queue'] ?? null) ->allOnConnection($this->options['connection'] ?? null) ->chain($chain->slice(1)->values()->all()); } else { $job->withBatchId($this->id); $count++; } return $job; }); $this->repository->transaction(function () use ($jobs, $count) { $this->repository->incrementTotalJobs($this->id, $count); $this->queue->connection($this->options['connection'] ?? null)->bulk( $jobs->all(), $data = '', $this->options['queue'] ?? null ); }); return $this->fresh(); } /** * Prepare a chain that exists within the jobs being added. * * @param array $chain * @return \Illuminate\Support\Collection */ protected function prepareBatchedChain(array $chain) { return (new Collection($chain))->map(function ($job) { $job = $job instanceof Closure ? CallQueuedClosure::create($job) : $job; return $job->withBatchId($this->id); }); } /** * Get the total number of jobs that have been processed by the batch thus far. * * @return int */ public function processedJobs() { return $this->totalJobs - $this->pendingJobs; } /** * Get the percentage of jobs that have been processed (between 0-100). * * @return int */ public function progress() { return $this->totalJobs > 0 ? round(($this->processedJobs() / $this->totalJobs) * 100) : 0; } /** * Record that a job within the batch finished successfully, executing any callbacks if necessary. * * @param string $jobId * @return void */ public function recordSuccessfulJob(string $jobId) { $counts = $this->decrementPendingJobs($jobId); if ($this->hasProgressCallbacks()) { $this->invokeCallbacks('progress'); } if ($counts->pendingJobs === 0) { $this->repository->markAsFinished($this->id); } if ($counts->pendingJobs === 0 && $this->hasThenCallbacks()) { $this->invokeCallbacks('then'); } if ($counts->allJobsHaveRanExactlyOnce() && $this->hasFinallyCallbacks()) { $this->invokeCallbacks('finally'); } } /** * Decrement the pending jobs for the batch. * * @param string $jobId * @return \Illuminate\Bus\UpdatedBatchJobCounts */ public function decrementPendingJobs(string $jobId) { return $this->repository->decrementPendingJobs($this->id, $jobId); } /** * Invoke the callbacks of the given type. */ protected function invokeCallbacks(string $type, ?Throwable $e = null): void { $batch = $this->fresh(); foreach ($this->options[$type] ?? [] as $handler) { $this->invokeHandlerCallback($handler, $batch, $e); } } /** * Determine if the batch has finished executing. * * @return bool */ public function finished() { return ! is_null($this->finishedAt); } /** * Determine if the batch has "progress" callbacks. * * @return bool */ public function hasProgressCallbacks() { return isset($this->options['progress']) && ! empty($this->options['progress']); } /** * Determine if the batch has "success" callbacks. * * @return bool */ public function hasThenCallbacks() { return isset($this->options['then']) && ! empty($this->options['then']); } /** * Determine if the batch allows jobs to fail without cancelling the batch. * * @return bool */ public function allowsFailures() { return Arr::get($this->options, 'allowFailures', false) === true; } /** * Determine if the batch has job failures. * * @return bool */ public function hasFailures() { return $this->failedJobs > 0; } /** * Record that a job within the batch failed to finish successfully, executing any callbacks if necessary. * * @param string $jobId * @param \Throwable $e * @return void */ public function recordFailedJob(string $jobId, $e) { $counts = $this->incrementFailedJobs($jobId); if ($counts->failedJobs === 1 && ! $this->allowsFailures()) { $this->cancel(); } if ($this->allowsFailures()) { if ($this->hasProgressCallbacks()) { $this->invokeCallbacks('progress', $e); } if ($this->hasFailureCallbacks()) { $this->invokeCallbacks('failure', $e); } } if ($counts->failedJobs === 1 && $this->hasCatchCallbacks()) { $this->invokeCallbacks('catch', $e); } if ($counts->allJobsHaveRanExactlyOnce() && $this->hasFinallyCallbacks()) { $this->invokeCallbacks('finally'); } } /** * Increment the failed jobs for the batch. * * @param string $jobId * @return \Illuminate\Bus\UpdatedBatchJobCounts */ public function incrementFailedJobs(string $jobId) { return $this->repository->incrementFailedJobs($this->id, $jobId); } /** * Determine if the batch has "catch" callbacks. * * @return bool */ public function hasCatchCallbacks() { return isset($this->options['catch']) && ! empty($this->options['catch']); } /** * Determine if the batch has "failure" callbacks. */ public function hasFailureCallbacks(): bool { return isset($this->options['failure']) && ! empty($this->options['failure']); } /** * Determine if the batch has "finally" callbacks. * * @return bool */ public function hasFinallyCallbacks() { return isset($this->options['finally']) && ! empty($this->options['finally']); } /** * Cancel the batch. * * @return void */ public function cancel() { $this->repository->cancel($this->id); } /** * Determine if the batch has been cancelled. * * @return bool */ public function canceled() { return $this->cancelled(); } /** * Determine if the batch has been cancelled. * * @return bool */ public function cancelled() { return ! is_null($this->cancelledAt); } /** * Delete the batch from storage. * * @return void */ public function delete() { $this->repository->delete($this->id); } /** * Invoke a batch callback handler. * * @param callable $handler * @param \Illuminate\Bus\Batch $batch * @param \Throwable|null $e * @return void */ protected function invokeHandlerCallback($handler, Batch $batch, ?Throwable $e = null) { try { $handler($batch, $e); } catch (Throwable $e) { if (function_exists('report')) { report($e); } } } /** * Convert the batch to an array. * * @return array */ public function toArray() { return [ 'id' => $this->id, 'name' => $this->name, 'totalJobs' => $this->totalJobs, 'pendingJobs' => $this->pendingJobs, 'processedJobs' => $this->processedJobs(), 'progress' => $this->progress(), 'failedJobs' => $this->failedJobs, 'options' => $this->options, 'createdAt' => $this->createdAt, 'cancelledAt' => $this->cancelledAt, 'finishedAt' => $this->finishedAt, ]; } /** * Get the JSON serializable representation of the object. * * @return array */ public function jsonSerialize(): array { return $this->toArray(); } /** * Dynamically access the batch's "options" via properties. * * @param string $key * @return mixed */ public function __get($key) { return $this->options[$key] ?? null; } } ChainedBatch.php000066400000007452000000000000007525 0ustar00jobs = static::prepareNestedBatches($batch->jobs); $this->name = $batch->name; $this->options = $batch->options; $this->queue = $batch->queue(); $this->connection = $batch->connection(); } /** * Prepare any nested batches within the given collection of jobs. * * @param \Illuminate\Support\Collection $jobs * @return \Illuminate\Support\Collection */ public static function prepareNestedBatches(Collection $jobs): Collection { return $jobs->map(fn ($job) => match (true) { is_array($job) => static::prepareNestedBatches(new Collection($job))->all(), $job instanceof Collection => static::prepareNestedBatches($job), $job instanceof PendingBatch => new ChainedBatch($job), default => $job, }); } /** * Handle the job. * * @return void */ public function handle() { $this->attachRemainderOfChainToEndOfBatch( $this->toPendingBatch() )->dispatch(); } /** * Convert the chained batch instance into a pending batch. * * @return \Illuminate\Bus\PendingBatch */ public function toPendingBatch() { $batch = Container::getInstance()->make(Dispatcher::class)->batch($this->jobs); $batch->name = $this->name; $batch->options = $this->options; if ($this->queue) { $batch->onQueue($this->queue); } if ($this->connection) { $batch->onConnection($this->connection); } foreach ($this->chainCatchCallbacks ?? [] as $callback) { $batch->catch(function (Batch $batch, ?Throwable $exception) use ($callback) { if (! $batch->allowsFailures()) { $callback($exception); } }); } return $batch; } /** * Move the remainder of the chain to a "finally" batch callback. * * @param \Illuminate\Bus\PendingBatch $batch * @return \Illuminate\Bus\PendingBatch */ protected function attachRemainderOfChainToEndOfBatch(PendingBatch $batch) { if (! empty($this->chained)) { $next = unserialize(array_shift($this->chained)); $next->chained = $this->chained; $next->onConnection($next->connection ?: $this->chainConnection); $next->onQueue($next->queue ?: $this->chainQueue); $next->chainConnection = $this->chainConnection; $next->chainQueue = $this->chainQueue; $next->chainCatchCallbacks = $this->chainCatchCallbacks; $batch->finally(function (Batch $batch) use ($next) { if (! $batch->cancelled()) { Container::getInstance()->make(Dispatcher::class)->dispatch($next); } }); $this->chained = []; } return $batch; } } UniqueLock.php000066400000003157000000000000007305 0ustar00cache = $cache; } /** * Attempt to acquire a lock for the given job. * * @param mixed $job * @return bool */ public function acquire($job) { $uniqueFor = method_exists($job, 'uniqueFor') ? $job->uniqueFor() : ($job->uniqueFor ?? 0); $cache = method_exists($job, 'uniqueVia') ? $job->uniqueVia() : $this->cache; return (bool) $cache->lock($this->getKey($job), $uniqueFor)->get(); } /** * Release the lock for the given job. * * @param mixed $job * @return void */ public function release($job) { $cache = method_exists($job, 'uniqueVia') ? $job->uniqueVia() : $this->cache; $cache->lock($this->getKey($job))->forceRelease(); } /** * Generate the lock key for the given job. * * @param mixed $job * @return string */ public static function getKey($job) { $uniqueId = method_exists($job, 'uniqueId') ? $job->uniqueId() : ($job->uniqueId ?? ''); return 'laravel_unique_job:'.get_class($job).':'.$uniqueId; } } BatchRepository.php000066400000004417000000000000010347 0ustar00connection = enum_value($connection); return $this; } /** * Set the desired queue for the job. * * @param \UnitEnum|string|null $queue * @return $this */ public function onQueue($queue) { $this->queue = enum_value($queue); return $this; } /** * Set the desired job "group". * * This feature is only supported by some queues, such as Amazon SQS. * * @param \UnitEnum|string $group * @return $this */ public function onGroup($group) { $this->messageGroup = enum_value($group); return $this; } /** * Set the desired job deduplicator callback. * * This feature is only supported by some queues, such as Amazon SQS FIFO. * * @param callable|null $deduplicator * @return $this */ public function withDeduplicator($deduplicator) { $this->deduplicator = $deduplicator instanceof Closure ? new SerializableClosure($deduplicator) : $deduplicator; return $this; } /** * Set the desired connection for the chain. * * @param \UnitEnum|string|null $connection * @return $this */ public function allOnConnection($connection) { $resolvedConnection = enum_value($connection); $this->chainConnection = $resolvedConnection; $this->connection = $resolvedConnection; return $this; } /** * Set the desired queue for the chain. * * @param \UnitEnum|string|null $queue * @return $this */ public function allOnQueue($queue) { $resolvedQueue = enum_value($queue); $this->chainQueue = $resolvedQueue; $this->queue = $resolvedQueue; return $this; } /** * Set the desired delay in seconds for the job. * * @param \DateTimeInterface|\DateInterval|array|int|null $delay * @return $this */ public function delay($delay) { $this->delay = $delay; return $this; } /** * Set the delay for the job to zero seconds. * * @return $this */ public function withoutDelay() { $this->delay = 0; return $this; } /** * Indicate that the job should be dispatched after all database transactions have committed. * * @return $this */ public function afterCommit() { $this->afterCommit = true; return $this; } /** * Indicate that the job should not wait until database transactions have been committed before dispatching. * * @return $this */ public function beforeCommit() { $this->afterCommit = false; return $this; } /** * Specify the middleware the job should be dispatched through. * * @param array|object $middleware * @return $this */ public function through($middleware) { $this->middleware = Arr::wrap($middleware); return $this; } /** * Set the jobs that should run if this job is successful. * * @param array $chain * @return $this */ public function chain($chain) { $this->chained = ChainedBatch::prepareNestedBatches(new Collection($chain)) ->map(fn ($job) => $this->serializeJob($job)) ->all(); return $this; } /** * Prepend a job to the current chain so that it is run after the currently running job. * * @param mixed $job * @return $this */ public function prependToChain($job) { $jobs = ChainedBatch::prepareNestedBatches(Collection::wrap($job)); foreach ($jobs->reverse() as $job) { $this->chained = Arr::prepend($this->chained, $this->serializeJob($job)); } return $this; } /** * Append a job to the end of the current chain. * * @param mixed $job * @return $this */ public function appendToChain($job) { $jobs = ChainedBatch::prepareNestedBatches(Collection::wrap($job)); foreach ($jobs as $job) { $this->chained = array_merge($this->chained, [$this->serializeJob($job)]); } return $this; } /** * Serialize a job for queuing. * * @param mixed $job * @return string * * @throws \RuntimeException */ protected function serializeJob($job) { if ($job instanceof Closure) { if (! class_exists(CallQueuedClosure::class)) { throw new RuntimeException( 'To enable support for closure jobs, please install the illuminate/queue package.' ); } $job = CallQueuedClosure::create($job); } return serialize($job); } /** * Dispatch the next job on the chain. * * @return void */ public function dispatchNextJobInChain() { if (! empty($this->chained)) { dispatch(tap(unserialize(array_shift($this->chained)), function ($next) { $next->chained = $this->chained; $next->onConnection($next->connection ?: $this->chainConnection); $next->onQueue($next->queue ?: $this->chainQueue); $next->chainConnection = $this->chainConnection; $next->chainQueue = $this->chainQueue; $next->chainCatchCallbacks = $this->chainCatchCallbacks; })); } } /** * Invoke all of the chain's failed job callbacks. * * @param \Throwable $e * @return void */ public function invokeChainCatchCallbacks($e) { (new Collection($this->chainCatchCallbacks))->each(function ($callback) use ($e) { $callback($e); }); } /** * Assert that the job has the given chain of jobs attached to it. * * @param array $expectedChain * @return void */ public function assertHasChain($expectedChain) { PHPUnit::assertTrue( (new Collection($expectedChain))->isNotEmpty(), 'The expected chain can not be empty.' ); if ((new Collection($expectedChain))->contains(fn ($job) => is_object($job))) { $expectedChain = (new Collection($expectedChain))->map(fn ($job) => serialize($job))->all(); } else { $chain = (new Collection($this->chained))->map(fn ($job) => get_class(unserialize($job)))->all(); } PHPUnit::assertTrue( $expectedChain === ($chain ?? $this->chained), 'The job does not have the expected chain.' ); } /** * Assert that the job has no remaining chained jobs. * * @return void */ public function assertDoesntHaveChain() { PHPUnit::assertEmpty($this->chained, 'The job has chained jobs.'); } } Batchable.php000066400000005417000000000000007074 0ustar00fakeBatch) { return $this->fakeBatch; } if ($this->batchId) { return Container::getInstance()->make(BatchRepository::class)?->find($this->batchId); } } /** * Determine if the batch is still active and processing. * * @return bool */ public function batching() { $batch = $this->batch(); return $batch && ! $batch->cancelled(); } /** * Set the batch ID on the job. * * @param string $batchId * @return $this */ public function withBatchId(string $batchId) { $this->batchId = $batchId; return $this; } /** * Indicate that the job should use a fake batch. * * @param string $id * @param string $name * @param int $totalJobs * @param int $pendingJobs * @param int $failedJobs * @param array $failedJobIds * @param array $options * @param \Carbon\CarbonImmutable|null $createdAt * @param \Carbon\CarbonImmutable|null $cancelledAt * @param \Carbon\CarbonImmutable|null $finishedAt * @return array{0: $this, 1: \Illuminate\Support\Testing\Fakes\BatchFake} */ public function withFakeBatch(string $id = '', string $name = '', int $totalJobs = 0, int $pendingJobs = 0, int $failedJobs = 0, array $failedJobIds = [], array $options = [], ?CarbonImmutable $createdAt = null, ?CarbonImmutable $cancelledAt = null, ?CarbonImmutable $finishedAt = null) { $this->fakeBatch = new BatchFake( empty($id) ? (string) Str::uuid() : $id, $name, $totalJobs, $pendingJobs, $failedJobs, $failedJobIds, $options, $createdAt ?? CarbonImmutable::now(), $cancelledAt, $finishedAt, ); return [$this, $this->fakeBatch]; } } UpdatedBatchJobCounts.php000066400000001504000000000000011377 0ustar00pendingJobs = $pendingJobs; $this->failedJobs = $failedJobs; } /** * Determine if all jobs have run exactly once. * * @return bool */ public function allJobsHaveRanExactlyOnce() { return ($this->pendingJobs - $this->failedJobs) === 0; } } PendingBatch.php000066400000025714000000000000007557 0ustar00 */ protected static $batchableClasses = []; /** * Create a new pending batch instance. * * @param \Illuminate\Contracts\Container\Container $container * @param \Illuminate\Support\Collection $jobs */ public function __construct(Container $container, Collection $jobs) { $this->container = $container; $this->jobs = $jobs->each(function (object|array $job) { $this->ensureJobIsBatchable($job); }); } /** * Add jobs to the batch. * * @param iterable|object|array $jobs * @return $this */ public function add($jobs) { $jobs = is_iterable($jobs) ? $jobs : Arr::wrap($jobs); foreach ($jobs as $job) { $this->ensureJobIsBatchable($job); $this->jobs->push($job); } return $this; } /** * Ensure the given job is batchable. * * @param object|array $job * @return void */ protected function ensureJobIsBatchable(object|array $job): void { foreach (Arr::wrap($job) as $job) { if ($job instanceof PendingBatch || $job instanceof Closure) { return; } if (! (static::$batchableClasses[$job::class] ?? false) && ! in_array(Batchable::class, class_uses_recursive($job))) { static::$batchableClasses[$job::class] = false; throw new RuntimeException(sprintf('Attempted to batch job [%s], but it does not use the Batchable trait.', $job::class)); } static::$batchableClasses[$job::class] = true; } } /** * Add a callback to be executed when the batch is stored. * * @param callable $callback * @return $this */ public function before($callback) { $this->registerCallback('before', $callback); return $this; } /** * Get the "before" callbacks that have been registered with the pending batch. * * @return array */ public function beforeCallbacks() { return $this->options['before'] ?? []; } /** * Add a callback to be executed after a job in the batch have executed successfully. * * @param callable $callback * @return $this */ public function progress($callback) { $this->registerCallback('progress', $callback); return $this; } /** * Get the "progress" callbacks that have been registered with the pending batch. * * @return array */ public function progressCallbacks() { return $this->options['progress'] ?? []; } /** * Add a callback to be executed after all jobs in the batch have executed successfully. * * @param callable $callback * @return $this */ public function then($callback) { $this->registerCallback('then', $callback); return $this; } /** * Get the "then" callbacks that have been registered with the pending batch. * * @return array */ public function thenCallbacks() { return $this->options['then'] ?? []; } /** * Add a callback to be executed after the first failing job in the batch. * * @param callable $callback * @return $this */ public function catch($callback) { $this->registerCallback('catch', $callback); return $this; } /** * Get the "catch" callbacks that have been registered with the pending batch. * * @return array */ public function catchCallbacks() { return $this->options['catch'] ?? []; } /** * Add a callback to be executed after the batch has finished executing. * * @param callable $callback * @return $this */ public function finally($callback) { $this->registerCallback('finally', $callback); return $this; } /** * Get the "finally" callbacks that have been registered with the pending batch. * * @return array */ public function finallyCallbacks() { return $this->options['finally'] ?? []; } /** * Indicate that the batch should not be canceled when a job within the batch fails. * * Optionally, add callbacks to be executed upon each job failure. * * @template TParam of (Closure(\Illuminate\Bus\Batch, \Throwable|null): mixed)|(callable(\Illuminate\Bus\Batch, \Throwable|null): mixed) * * @param bool|TParam|array $param * @return $this */ public function allowFailures($param = true) { if (! is_bool($param)) { $param = Arr::wrap($param); foreach ($param as $callback) { if (is_callable($callback)) { $this->registerCallback('failure', $callback); } } } $this->options['allowFailures'] = ! ($param === false); return $this; } /** * Determine if the pending batch allows jobs to fail without cancelling the batch. * * @return bool */ public function allowsFailures() { return Arr::get($this->options, 'allowFailures', false) === true; } /** * Get the "failure" callbacks that have been registered with the pending batch. * * @return array */ public function failureCallbacks(): array { return $this->options['failure'] ?? []; } /** * Register a callback with proper serialization. */ private function registerCallback(string $type, Closure|callable $callback): void { $this->options[$type][] = $callback instanceof Closure ? new SerializableClosure($callback) : $callback; } /** * Set the name for the batch. * * @param string $name * @return $this */ public function name(string $name) { $this->name = $name; return $this; } /** * Specify the queue connection that the batched jobs should run on. * * @param string $connection * @return $this */ public function onConnection(string $connection) { $this->options['connection'] = $connection; return $this; } /** * Get the connection used by the pending batch. * * @return string|null */ public function connection() { return $this->options['connection'] ?? null; } /** * Specify the queue that the batched jobs should run on. * * @param \UnitEnum|string|null $queue * @return $this */ public function onQueue($queue) { $this->options['queue'] = enum_value($queue); return $this; } /** * Get the queue used by the pending batch. * * @return string|null */ public function queue() { return $this->options['queue'] ?? null; } /** * Add additional data into the batch's options array. * * @param string $key * @param mixed $value * @return $this */ public function withOption(string $key, $value) { $this->options[$key] = $value; return $this; } /** * Dispatch the batch. * * @return \Illuminate\Bus\Batch * * @throws \Throwable */ public function dispatch() { $repository = $this->container->make(BatchRepository::class); try { $batch = $this->store($repository); $batch = $batch->add($this->jobs); } catch (Throwable $e) { if (isset($batch)) { $repository->delete($batch->id); } throw $e; } $this->container->make(EventDispatcher::class)->dispatch( new BatchDispatched($batch) ); return $batch; } /** * Dispatch the batch after the response is sent to the browser. * * @return \Illuminate\Bus\Batch */ public function dispatchAfterResponse() { $repository = $this->container->make(BatchRepository::class); $batch = $this->store($repository); if ($batch) { $this->container->terminating(function () use ($batch) { $this->dispatchExistingBatch($batch); }); } return $batch; } /** * Dispatch an existing batch. * * @param \Illuminate\Bus\Batch $batch * @return void * * @throws \Throwable */ protected function dispatchExistingBatch($batch) { try { $batch = $batch->add($this->jobs); } catch (Throwable $e) { $batch->delete(); throw $e; } $this->container->make(EventDispatcher::class)->dispatch( new BatchDispatched($batch) ); } /** * Dispatch the batch if the given truth test passes. * * @param bool|\Closure $boolean * @return \Illuminate\Bus\Batch|null */ public function dispatchIf($boolean) { return value($boolean) ? $this->dispatch() : null; } /** * Dispatch the batch unless the given truth test passes. * * @param bool|\Closure $boolean * @return \Illuminate\Bus\Batch|null */ public function dispatchUnless($boolean) { return ! value($boolean) ? $this->dispatch() : null; } /** * Store the batch using the given repository. * * @param \Illuminate\Bus\BatchRepository $repository * @return \Illuminate\Bus\Batch */ protected function store($repository) { $batch = $repository->store($this); (new Collection($this->beforeCallbacks()))->each(function ($handler) use ($batch) { try { return $handler($batch); } catch (Throwable $e) { if (function_exists('report')) { report($e); } } }); return $batch; } } LICENSE.md000066400000002063000000000000006114 0ustar00The MIT License (MIT) Copyright (c) Taylor Otwell Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. DynamoBatchRepository.php000066400000035706000000000000011524 0ustar00factory = $factory; $this->dynamoDbClient = $dynamoDbClient; $this->applicationName = $applicationName; $this->table = $table; $this->ttl = $ttl; $this->ttlAttribute = $ttlAttribute; $this->marshaler = new Marshaler; } /** * Retrieve a list of batches. * * @param int $limit * @param mixed $before * @return \Illuminate\Bus\Batch[] */ public function get($limit = 50, $before = null) { $condition = 'application = :application'; if ($before) { $condition = 'application = :application AND id < :id'; } $result = $this->dynamoDbClient->query([ 'TableName' => $this->table, 'KeyConditionExpression' => $condition, 'ExpressionAttributeValues' => array_filter([ ':application' => ['S' => $this->applicationName], ':id' => array_filter(['S' => $before]), ]), 'Limit' => $limit, 'ScanIndexForward' => false, ]); return array_map( fn ($b) => $this->toBatch($this->marshaler->unmarshalItem($b, mapAsObject: true)), $result['Items'] ); } /** * Retrieve information about an existing batch. * * @param string $batchId * @return \Illuminate\Bus\Batch|null */ public function find(string $batchId) { if (trim($batchId) === '') { return null; } $b = $this->dynamoDbClient->getItem([ 'TableName' => $this->table, 'Key' => [ 'application' => ['S' => $this->applicationName], 'id' => ['S' => $batchId], ], ]); if (! isset($b['Item'])) { // If we didn't find it via a standard read, attempt consistent read... $b = $this->dynamoDbClient->getItem([ 'TableName' => $this->table, 'Key' => [ 'application' => ['S' => $this->applicationName], 'id' => ['S' => $batchId], ], 'ConsistentRead' => true, ]); if (! isset($b['Item'])) { return null; } } $batch = $this->marshaler->unmarshalItem($b['Item'], mapAsObject: true); if ($batch) { return $this->toBatch($batch); } } /** * Store a new pending batch. * * @param \Illuminate\Bus\PendingBatch $batch * @return \Illuminate\Bus\Batch */ public function store(PendingBatch $batch) { $id = (string) Str::orderedUuid(); $batch = [ 'id' => $id, 'name' => $batch->name, 'total_jobs' => 0, 'pending_jobs' => 0, 'failed_jobs' => 0, 'failed_job_ids' => [], 'options' => $this->serialize($batch->options ?? []), 'created_at' => time(), 'cancelled_at' => null, 'finished_at' => null, ]; if (! is_null($this->ttl)) { $batch[$this->ttlAttribute] = time() + $this->ttl; } $this->dynamoDbClient->putItem([ 'TableName' => $this->table, 'Item' => $this->marshaler->marshalItem( array_merge(['application' => $this->applicationName], $batch) ), ]); return $this->find($id); } /** * Increment the total number of jobs within the batch. * * @param string $batchId * @param int $amount * @return void */ public function incrementTotalJobs(string $batchId, int $amount) { $update = 'SET total_jobs = total_jobs + :val, pending_jobs = pending_jobs + :val'; if ($this->ttl) { $update = "SET total_jobs = total_jobs + :val, pending_jobs = pending_jobs + :val, #{$this->ttlAttribute} = :ttl"; } $this->dynamoDbClient->updateItem(array_filter([ 'TableName' => $this->table, 'Key' => [ 'application' => ['S' => $this->applicationName], 'id' => ['S' => $batchId], ], 'UpdateExpression' => $update, 'ExpressionAttributeValues' => array_filter([ ':val' => ['N' => "$amount"], ':ttl' => array_filter(['N' => $this->getExpiryTime()]), ]), 'ExpressionAttributeNames' => $this->ttlExpressionAttributeName(), 'ReturnValues' => 'ALL_NEW', ])); } /** * Decrement the total number of pending jobs for the batch. * * @param string $batchId * @param string $jobId * @return \Illuminate\Bus\UpdatedBatchJobCounts */ public function decrementPendingJobs(string $batchId, string $jobId) { $update = 'SET pending_jobs = pending_jobs - :inc'; if ($this->ttl !== null) { $update = "SET pending_jobs = pending_jobs - :inc, #{$this->ttlAttribute} = :ttl"; } $batch = $this->dynamoDbClient->updateItem(array_filter([ 'TableName' => $this->table, 'Key' => [ 'application' => ['S' => $this->applicationName], 'id' => ['S' => $batchId], ], 'UpdateExpression' => $update, 'ExpressionAttributeValues' => array_filter([ ':inc' => ['N' => '1'], ':ttl' => array_filter(['N' => $this->getExpiryTime()]), ]), 'ExpressionAttributeNames' => $this->ttlExpressionAttributeName(), 'ReturnValues' => 'ALL_NEW', ])); $values = $this->marshaler->unmarshalItem($batch['Attributes']); return new UpdatedBatchJobCounts( $values['pending_jobs'], $values['failed_jobs'] ); } /** * Increment the total number of failed jobs for the batch. * * @param string $batchId * @param string $jobId * @return \Illuminate\Bus\UpdatedBatchJobCounts */ public function incrementFailedJobs(string $batchId, string $jobId) { $update = 'SET failed_jobs = failed_jobs + :inc, failed_job_ids = list_append(failed_job_ids, :jobId)'; if ($this->ttl !== null) { $update = "SET failed_jobs = failed_jobs + :inc, failed_job_ids = list_append(failed_job_ids, :jobId), #{$this->ttlAttribute} = :ttl"; } $batch = $this->dynamoDbClient->updateItem(array_filter([ 'TableName' => $this->table, 'Key' => [ 'application' => ['S' => $this->applicationName], 'id' => ['S' => $batchId], ], 'UpdateExpression' => $update, 'ExpressionAttributeValues' => array_filter([ ':jobId' => $this->marshaler->marshalValue([$jobId]), ':inc' => ['N' => '1'], ':ttl' => array_filter(['N' => $this->getExpiryTime()]), ]), 'ExpressionAttributeNames' => $this->ttlExpressionAttributeName(), 'ReturnValues' => 'ALL_NEW', ])); $values = $this->marshaler->unmarshalItem($batch['Attributes']); return new UpdatedBatchJobCounts( $values['pending_jobs'], $values['failed_jobs'] ); } /** * Mark the batch that has the given ID as finished. * * @param string $batchId * @return void */ public function markAsFinished(string $batchId) { $update = 'SET finished_at = :timestamp'; if ($this->ttl !== null) { $update = "SET finished_at = :timestamp, #{$this->ttlAttribute} = :ttl"; } $this->dynamoDbClient->updateItem(array_filter([ 'TableName' => $this->table, 'Key' => [ 'application' => ['S' => $this->applicationName], 'id' => ['S' => $batchId], ], 'UpdateExpression' => $update, 'ExpressionAttributeValues' => array_filter([ ':timestamp' => ['N' => (string) time()], ':ttl' => array_filter(['N' => $this->getExpiryTime()]), ]), 'ExpressionAttributeNames' => $this->ttlExpressionAttributeName(), ])); } /** * Cancel the batch that has the given ID. * * @param string $batchId * @return void */ public function cancel(string $batchId) { $update = 'SET cancelled_at = :timestamp, finished_at = :timestamp'; if ($this->ttl !== null) { $update = "SET cancelled_at = :timestamp, finished_at = :timestamp, #{$this->ttlAttribute} = :ttl"; } $this->dynamoDbClient->updateItem(array_filter([ 'TableName' => $this->table, 'Key' => [ 'application' => ['S' => $this->applicationName], 'id' => ['S' => $batchId], ], 'UpdateExpression' => $update, 'ExpressionAttributeValues' => array_filter([ ':timestamp' => ['N' => (string) time()], ':ttl' => array_filter(['N' => $this->getExpiryTime()]), ]), 'ExpressionAttributeNames' => $this->ttlExpressionAttributeName(), ])); } /** * Delete the batch that has the given ID. * * @param string $batchId * @return void */ public function delete(string $batchId) { $this->dynamoDbClient->deleteItem([ 'TableName' => $this->table, 'Key' => [ 'application' => ['S' => $this->applicationName], 'id' => ['S' => $batchId], ], ]); } /** * Execute the given Closure within a storage specific transaction. * * @param \Closure $callback * @return mixed */ public function transaction(Closure $callback) { return $callback(); } /** * Rollback the last database transaction for the connection. * * @return void */ public function rollBack() { } /** * Convert the given raw batch to a Batch object. * * @param object $batch * @return \Illuminate\Bus\Batch */ protected function toBatch($batch) { return $this->factory->make( $this, $batch->id, $batch->name, (int) $batch->total_jobs, (int) $batch->pending_jobs, (int) $batch->failed_jobs, $batch->failed_job_ids, $this->unserialize($batch->options) ?? [], CarbonImmutable::createFromTimestamp($batch->created_at, date_default_timezone_get()), $batch->cancelled_at ? CarbonImmutable::createFromTimestamp($batch->cancelled_at, date_default_timezone_get()) : $batch->cancelled_at, $batch->finished_at ? CarbonImmutable::createFromTimestamp($batch->finished_at, date_default_timezone_get()) : $batch->finished_at ); } /** * Create the underlying DynamoDB table. * * @return void */ public function createAwsDynamoTable(): void { $definition = [ 'TableName' => $this->table, 'AttributeDefinitions' => [ [ 'AttributeName' => 'application', 'AttributeType' => 'S', ], [ 'AttributeName' => 'id', 'AttributeType' => 'S', ], ], 'KeySchema' => [ [ 'AttributeName' => 'application', 'KeyType' => 'HASH', ], [ 'AttributeName' => 'id', 'KeyType' => 'RANGE', ], ], 'BillingMode' => 'PAY_PER_REQUEST', ]; $this->dynamoDbClient->createTable($definition); if (! is_null($this->ttl)) { $this->dynamoDbClient->updateTimeToLive([ 'TableName' => $this->table, 'TimeToLiveSpecification' => [ 'AttributeName' => $this->ttlAttribute, 'Enabled' => true, ], ]); } } /** * Delete the underlying DynamoDB table. */ public function deleteAwsDynamoTable(): void { $this->dynamoDbClient->deleteTable([ 'TableName' => $this->table, ]); } /** * Get the expiry time based on the configured time-to-live. * * @return string|null */ protected function getExpiryTime(): ?string { return is_null($this->ttl) ? null : (string) (time() + $this->ttl); } /** * Get the expression attribute name for the time-to-live attribute. * * @return array */ protected function ttlExpressionAttributeName(): array { return is_null($this->ttl) ? [] : ["#{$this->ttlAttribute}" => $this->ttlAttribute]; } /** * Serialize the given value. * * @param mixed $value * @return string */ protected function serialize($value) { return serialize($value); } /** * Unserialize the given value. * * @param string $serialized * @return mixed */ protected function unserialize($serialized) { return unserialize($serialized); } /** * Get the underlying DynamoDB client instance. * * @return \Aws\DynamoDb\DynamoDbClient */ public function getDynamoClient(): DynamoDbClient { return $this->dynamoDbClient; } /** * The name of the table that contains the batch records. * * @return string */ public function getTable(): string { return $this->table; } } composer.json000066400000001747000000000000007242 0ustar00{ "name": "illuminate/bus", "description": "The Illuminate Bus package.", "license": "MIT", "homepage": "https://laravel.com", "support": { "issues": "https://github.com/laravel/framework/issues", "source": "https://github.com/laravel/framework" }, "authors": [ { "name": "Taylor Otwell", "email": "taylor@laravel.com" } ], "require": { "php": "^8.2", "illuminate/collections": "^12.0", "illuminate/contracts": "^12.0", "illuminate/pipeline": "^12.0", "illuminate/support": "^12.0" }, "autoload": { "psr-4": { "Illuminate\\Bus\\": "" } }, "extra": { "branch-alias": { "dev-master": "12.x-dev" } }, "suggest": { "illuminate/queue": "Required to use closures when chaining jobs (^12.0)." }, "config": { "sort-packages": true }, "minimum-stability": "dev" } DatabaseBatchRepository.php000066400000025457000000000000012003 0ustar00factory = $factory; $this->connection = $connection; $this->table = $table; } /** * Retrieve a list of batches. * * @param int $limit * @param mixed $before * @return \Illuminate\Bus\Batch[] */ public function get($limit = 50, $before = null) { return $this->connection->table($this->table) ->orderByDesc('id') ->limit($limit) ->when($before, fn ($q) => $q->where('id', '<', $before)) ->get() ->map(function ($batch) { return $this->toBatch($batch); }) ->all(); } /** * Retrieve information about an existing batch. * * @param string $batchId * @return \Illuminate\Bus\Batch|null */ public function find(string $batchId) { $batch = $this->connection->table($this->table) ->useWritePdo() ->where('id', $batchId) ->first(); if ($batch) { return $this->toBatch($batch); } } /** * Store a new pending batch. * * @param \Illuminate\Bus\PendingBatch $batch * @return \Illuminate\Bus\Batch */ public function store(PendingBatch $batch) { $id = (string) Str::orderedUuid(); $this->connection->table($this->table)->insert([ 'id' => $id, 'name' => $batch->name, 'total_jobs' => 0, 'pending_jobs' => 0, 'failed_jobs' => 0, 'failed_job_ids' => '[]', 'options' => $this->serialize($batch->options), 'created_at' => time(), 'cancelled_at' => null, 'finished_at' => null, ]); return $this->find($id); } /** * Increment the total number of jobs within the batch. * * @param string $batchId * @param int $amount * @return void */ public function incrementTotalJobs(string $batchId, int $amount) { $this->connection->table($this->table)->where('id', $batchId)->update([ 'total_jobs' => new Expression('total_jobs + '.$amount), 'pending_jobs' => new Expression('pending_jobs + '.$amount), 'finished_at' => null, ]); } /** * Decrement the total number of pending jobs for the batch. * * @param string $batchId * @param string $jobId * @return \Illuminate\Bus\UpdatedBatchJobCounts */ public function decrementPendingJobs(string $batchId, string $jobId) { $values = $this->updateAtomicValues($batchId, function ($batch) use ($jobId) { return [ 'pending_jobs' => $batch->pending_jobs - 1, 'failed_jobs' => $batch->failed_jobs, 'failed_job_ids' => json_encode(array_values(array_diff((array) json_decode($batch->failed_job_ids, true), [$jobId]))), ]; }); return new UpdatedBatchJobCounts( $values['pending_jobs'], $values['failed_jobs'] ); } /** * Increment the total number of failed jobs for the batch. * * @param string $batchId * @param string $jobId * @return \Illuminate\Bus\UpdatedBatchJobCounts */ public function incrementFailedJobs(string $batchId, string $jobId) { $values = $this->updateAtomicValues($batchId, function ($batch) use ($jobId) { return [ 'pending_jobs' => $batch->pending_jobs, 'failed_jobs' => $batch->failed_jobs + 1, 'failed_job_ids' => json_encode(array_values(array_unique(array_merge((array) json_decode($batch->failed_job_ids, true), [$jobId])))), ]; }); return new UpdatedBatchJobCounts( $values['pending_jobs'], $values['failed_jobs'] ); } /** * Update an atomic value within the batch. * * @param string $batchId * @param \Closure $callback * @return int|null */ protected function updateAtomicValues(string $batchId, Closure $callback) { return $this->connection->transaction(function () use ($batchId, $callback) { $batch = $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); }); }); } /** * Mark the batch that has the given ID as finished. * * @param string $batchId * @return void */ public function markAsFinished(string $batchId) { $this->connection->table($this->table)->where('id', $batchId)->update([ 'finished_at' => time(), ]); } /** * Cancel the batch that has the given ID. * * @param string $batchId * @return void */ public function cancel(string $batchId) { $this->connection->table($this->table)->where('id', $batchId)->update([ 'cancelled_at' => time(), 'finished_at' => time(), ]); } /** * Delete the batch that has the given ID. * * @param string $batchId * @return void */ public function delete(string $batchId) { $this->connection->table($this->table)->where('id', $batchId)->delete(); } /** * Prune all of the entries older than the given date. * * @param \DateTimeInterface $before * @return int */ public function prune(DateTimeInterface $before) { $query = $this->connection->table($this->table) ->whereNotNull('finished_at') ->where('finished_at', '<', $before->getTimestamp()); $totalDeleted = 0; do { $deleted = $query->limit(1000)->delete(); $totalDeleted += $deleted; } while ($deleted !== 0); return $totalDeleted; } /** * Prune all of the unfinished entries older than the given date. * * @param \DateTimeInterface $before * @return int */ public function pruneUnfinished(DateTimeInterface $before) { $query = $this->connection->table($this->table) ->whereNull('finished_at') ->where('created_at', '<', $before->getTimestamp()); $totalDeleted = 0; do { $deleted = $query->limit(1000)->delete(); $totalDeleted += $deleted; } while ($deleted !== 0); return $totalDeleted; } /** * Prune all of the cancelled entries older than the given date. * * @param \DateTimeInterface $before * @return int */ public function pruneCancelled(DateTimeInterface $before) { $query = $this->connection->table($this->table) ->whereNotNull('cancelled_at') ->where('created_at', '<', $before->getTimestamp()); $totalDeleted = 0; do { $deleted = $query->limit(1000)->delete(); $totalDeleted += $deleted; } while ($deleted !== 0); return $totalDeleted; } /** * Execute the given Closure within a storage specific transaction. * * @param \Closure $callback * @return mixed */ public function transaction(Closure $callback) { return $this->connection->transaction(fn () => $callback()); } /** * Rollback the last database transaction for the connection. * * @return void */ public function rollBack() { $this->connection->rollBack(toLevel: 0); } /** * Serialize the given value. * * @param mixed $value * @return string */ protected function serialize($value) { $serialized = serialize($value); return $this->connection instanceof PostgresConnection ? base64_encode($serialized) : $serialized; } /** * Unserialize the given value. * * @param string $serialized * @return mixed */ protected function unserialize($serialized) { if ($this->connection instanceof PostgresConnection && ! Str::contains($serialized, [':', ';'])) { $serialized = base64_decode($serialized); } try { return unserialize($serialized); } catch (Throwable) { return []; } } /** * Convert the given raw batch to a Batch object. * * @param object $batch * @return \Illuminate\Bus\Batch */ protected function toBatch($batch) { return $this->factory->make( $this, $batch->id, $batch->name, (int) $batch->total_jobs, (int) $batch->pending_jobs, (int) $batch->failed_jobs, (array) json_decode($batch->failed_job_ids, true), $this->unserialize($batch->options), CarbonImmutable::createFromTimestamp($batch->created_at, date_default_timezone_get()), $batch->cancelled_at ? CarbonImmutable::createFromTimestamp($batch->cancelled_at, date_default_timezone_get()) : $batch->cancelled_at, $batch->finished_at ? CarbonImmutable::createFromTimestamp($batch->finished_at, date_default_timezone_get()) : $batch->finished_at ); } /** * Get the underlying database connection. * * @return \Illuminate\Database\Connection */ public function getConnection() { return $this->connection; } /** * Set the underlying database connection. * * @param \Illuminate\Database\Connection $connection * @return void */ public function setConnection(Connection $connection) { $this->connection = $connection; } } Events/BatchDispatched.php000066400000000447000000000000011503 0ustar00