Function Descriptions

<?php

public function process($pulse): void
{
    if ($this->lockManager->isLocked(self::HEARTBEAT_LOCK_KEY)) {
        throw new LocalizedException(__('Process is locked! Could not execute heartbeat.'));
    }

    $this->lockManager->lock(self::HEARTBEAT_LOCK_KEY);
    try {
        if ($this->isMaintenanceModeEnabled->execute()) {
            $this->heartbeatLogger->info('Maintenance mode is active for heartbeat.');
        } else {
            $this->spawnPipelines->setPulse((string)$pulse);
            $this->spawnPipelines->execute();
            $this->executePendingSteps->execute();
        }
        $this->cancelExpiredPipelines->execute();
        $this->updatePipelineStatus->execute();
        $this->lockManager->unlock(self::HEARTBEAT_LOCK_KEY);
    } catch (Throwable $throwable) {
        $this->lockManager->unlock(self::HEARTBEAT_LOCK_KEY);
        throw $throwable;
    }
}

<?php

/**
 * Create new pipeline instance
 * @return void
 */
public function execute()
{
    $this->logger->info("Initiating new pipelines");
    $pipelineList = $this->config->getList();
    foreach ($pipelineList as $pipeline) {
        $this->logger->debug            (
            sprintf(
                'Checking pipeline "%s"',
                $pipeline[SpawnPipelinesCommand::PIPELINE_CONFIG_NAME]
            )
        );
        if ($this->isPipelineReady($pipeline) === false) {
            $this->stepArguments = [];
            continue;
        }
        $this->logger->debug
            sprintf(
                ' ==> Initiate pipeline "%s"',
                $pipeline[SpawnPipelinesCommand::PIPELINE_CONFIG_NAME]
            )
        );
        $this->pipelineBuilder->setStepArguments($this->stepArguments)
        $this->pipelineBuilder->createPipelineByConfig($pipeline)
    }
}

<?php

/**
 * Publishes a step message by given StepId.
 *
 * @param StepInterface $step
 */
public function publishStepMessage(StepInterface $step)
{
    try {
        $data = json_encode(['step_id' => $step->getId()]);
        $this->updateStepStatus($
        $this->publisher->publish(StepPublisher::TOPIC_NAME, $data);
    } catch (\Throwable $exception) {
        $this->processException($step, $exception);
    }
}

<publisher topic="pipeline.step.process">
    <connection name="amqp" exchange="techdivision-process-pipelines"
                disabled="false"
    />
    <connection name="db" exchange="techdivision-process-pipelines-db"
                disabled="true"
    />
</publisher>

<publisher topic="pipeline.step.process">
    <connection name="amqp" exchange="techdivision-process-pipelines"
                disabled="false"
    />
    <connection name="db" exchange="techdivision-process-pipelines-db"
                disabled="true"
    />
</publisher>

<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="urn:magento:framework:Communication/etc/communication.xsd">
    <topic name="pipeline.step.process" is_synchronous="false" request="string" />
</config>

<?php

public function process(int $pipelineId)
{
    if (!$this->isMiniHeartbeatAllowed->execute()) {
        $this->heartbeatLogger->info('Mini heartbeat is not allowed.');
        return;
    }

    if ($this->lockManager->isLocked(PipelineHeartbeat::HEARTBEAT_LOCK_KEY)) {
        ...
        return;
    }

    $lockName = $this->getLockName($pipelineId);
    if ($this->lockManager->isLocked($lockName)) {
        ...
        return;
    }

    $this->lockManager->lock($lockName);
    try {
        $pipeline = $this->getPipelineById->execute($pipelineId);
        if ($this->isMaintenanceModeEnabled->execute()) {
            $this->heartbeatLogger->info('Maintenance mode is active for mini heartbeat.');
        } else {
            $this->executeAllPendingStepsByPipelineCommand->execute($pipeline);
            $this->cancelPipelineIfExpiredCommand->execute($pipeline);
        }
        $this->updatePipelineStatusCommand->execute($pipeline);
        $this->lockManager->unlock($lockName);
    } catch (Throwable $throwable) {
        $this->lockManager->unlock($lockName);
        throw $throwable;
    }
}

bin/magento queue:consumers:start [--max-messages=<value>] [--batch-size=<value>] [--single-thread] [--area-code=<value>] <consumer_name>

bin/magento queue:consumer:start pipeline.runner.db --max-messages=1

<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="urn:magento:framework-message-queue:etc/topology.xsd">
    <exchange name="techdivision-process-pipelines" type="topic" connection="amqp">
        <binding id="TechDivisionProcessPipelinesBinding" topic="pipeline.step.process" destinationType="queue" destination="pipeline_process_steps"/>
    </exchange>
    <exchange name="techdivision-process-pipelines-db" type="topic" connection="db">
        <binding id="TechDivisionProcessPipelinesBindingDb" topic="pipeline.step.process" destinationType="queue" destination="pipeline_process_steps"/>
    </exchange>
</config>

<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="urn:magento:framework-message-queue:etc/consumer.xsd">
    <!-- Deprecated -->
    <consumer name="pipelineRunner" queue="pipeline_process_steps" connection="amqp" handler="TechDivision\ProcessPipelines\Model\Runner::process"/>

    <consumer name="pipeline.runner.amqp" queue="pipeline_process_steps" connection="amqp" handler="TechDivision\ProcessPipelines\Model\Runner::process"/>
    <consumer name="pipeline.runner.db" queue="pipeline_process_steps" connection="db" handler="TechDivision\ProcessPipelines\Model\Runner::process"/>
</config>

<step name="delete_pipeline_data" executorType="TechDivision\ProcessPipelines\Model\Executor\DeleteExecutor" sortOrder="20" description="" >
    ...
</step>

<?php

public function process(StepInterface $step)
{
    $compressTime = $this->getCompareTime(
        ConfigInterface::CONFIG_PATH_CLEAN_UP_COMPRESS
    );
    $fileListToCompress = $this->getDirListOlderThenCompareTime($compressTime);
    ...
    foreach ($fileListToCompress as $folderToCompress) {
        if ($this->hasFiles($folderToCompress) === false) {
            ...
            $this->delete($folderToCompress);
            continue;
        }
        ...
        $this->archive($folderToCompress);
    }
}