PHP Oro\Component\MessageQueue\Job JobRunner Code Examples

                                                    public function postpone(JobRunner $jobRunner, Job $currentJob, string $fileName, array $body, array &$result)
    {
        $attempts = isset($body['attempts']) ? (int) $body['attempts'] + 1 : 1;

        if ($attempts > self::MAX_ATTEMPTS) {
            if (array_key_exists('postponedRows', $result) && !empty($result['postponedRows'])) {
                $result['errors'][] = $this->translator->trans(
                    'oro.importexport.import.postponed_rows',
                    ['%postponedRows%' => count($result['postponedRows'])]
                );
                $result['counts']['errors']++;
            }

            return;
        }

        $delay = $result['postponedDelay'] ?? self::DELAY_SECONDS;
        $jobRunner->createDelayed(
            $this->getDelayedJobName($currentJob, $attempts),
            function (JobRunner $jobRunner, Job $child) use ($body, $fileName, $attempts, $delay) {
                $body = array_merge($body, [
                    'jobId' => $child->getId(),
                    'attempts' => $attempts,
                    'fileName' => $fileName,
                ]);

                $body['options']['attempts'] = $attempts;
                $body['options']['max_attempts'] = self::MAX_ATTEMPTS;

                if (!array_key_exists('incremented_read', $body['options'])) {
                    $body['options']['incremented_read'] = false;
                }
                $message = new Message();
                if ($delay > 0) {
                    $message->setDelay($delay);
                }
                $message->setBody($body);
                $this->messageProducer->send(Topics::IMPORT, $message);
            }
        );
    }
                                            

                                                    private function scheduleUpdatingVisibilities(JobRunner $jobRunner, string $jobName, int $organizationId): void
    {
        $jobRunner->createDelayed(
            sprintf('%s:%d', $jobName, $organizationId),
            function (JobRunner $jobRunner, Job $childJob) use ($organizationId) {
                $this->producer->send(
                    UpdateVisibilitiesForOrganizationTopic::getName(),
                    ['jobId' => $childJob->getId(), 'organizationId' => $organizationId]
                );
            }
        );
    }
                                            

                                                    private function scheduleSettingVisibilities(
        JobRunner $jobRunner,
        string $jobName,
        int $organizationId,
        int $chunkNumber,
        int $firstEmailId,
        ?int $lastEmailId
    ): void {
        $jobRunner->createDelayed(
            sprintf('%s:%d', $jobName, $chunkNumber),
            function (JobRunner $jobRunner, Job $childJob) use ($organizationId, $firstEmailId, $lastEmailId) {
                $chunkMessageBody = [
                    'jobId'          => $childJob->getId(),
                    'organizationId' => $organizationId,
                    'firstEmailId'   => $firstEmailId,
                ];
                if (null !== $lastEmailId) {
                    $chunkMessageBody['lastEmailId'] = $lastEmailId;
                }
                $this->producer->send(
                    UpdateEmailVisibilitiesForOrganizationChunkTopic::getName(),
                    $chunkMessageBody
                );
            }
        );
    }
                                            

                                                    private function createChunkJob(
        JobRunner $jobRunner,
        string $jobName,
        array $parentBody,
        ChunkFile $chunkFile
    ): void {
        $jobRunner->createDelayed(
            $jobName,
            function (JobRunner $jobRunner, Job $job) use ($parentBody, $chunkFile) {
                $this->processingHelper->sendProcessChunkMessage($parentBody, $job, $chunkFile, true);

                return true;
            }
        );
    }
                                            

                                                    private function retryChunk(
        JobRunner $jobRunner,
        Job $job,
        array $body,
        ChunkFile $chunkFile
    ): bool {
        $chunkCount = $this->updateChunkCount($body['operationId'], 1);
        if (null === $chunkCount) {
            return false;
        }

        $jobRunner->createDelayed(
            $this->getRetryChunkJobName($job->getName()),
            function (JobRunner $jobRunner, Job $job) use ($body, $chunkFile) {
                $this->processingHelper->sendProcessChunkMessage(
                    $body,
                    $job,
                    $chunkFile,
                    $body['extra_chunk'] ?? false
                );

                return true;
            }
        );

        return true;
    }
                                            

                                                    private function scheduleCacheRecalculationForCustomer(JobRunner $jobRunner, string $entityClass, int $entityId)
    {
        $jobRunner->createDelayed(
            sprintf('%s:%s:%s', Topics::CALCULATE_OWNER_TREE_CACHE, $entityClass, $entityId),
            function (JobRunner $jobRunner, Job $child) use ($entityClass, $entityId) {
                $messageData = $this->businessUnitMessageFactory
                    ->createMessage($child->getId(), $entityClass, $entityId);

                $this->producer->send(Topics::CALCULATE_BUSINESS_UNIT_OWNER_TREE_CACHE, $messageData);
            }
        );
    }