PHP code example of shawty81 / multiprocessor

    
        
<?php
require_once('vendor/autoload.php');

/* Start to develop here. Best regards https://php-download.com/ */

    

shawty81 / multiprocessor example snippets

declare(strict_types=1);

or\ChildProcessorInterface;
use MultiProcessor\Iterator\ArrayIterator;
use MultiProcessor\Log\CommandLineLogger;
use MultiProcessor\MultiProcessor;
use MultiProcessor\Queue\Chunk;
use MultiProcessor\Settings;
use Psr\Log\LoggerAwareTrait;

final class GenerateThumbnails implements ChildProcessorInterface
{
    use LoggerAwareTrait;

    #[Override]
    public function init(): void {}

    #[Override]
    public function process(Chunk $chunk): void
    {
        foreach ($chunk->data as $image) {
            // Stands in for the real work: reading the file, resizing it and writing
            // the result back. A quarter of a second of it, without the dependencies.
            usleep(250_000);

            $this->logger?->info('pid {pid}: resized {image}', [
                'pid' => posix_getpid(),
                'image' => $image,
            ]);
        }
    }

    #[Override]
    public function finish(): void {}
}

$logger = new CommandLineLogger();

$iterator = new ArrayIterator();
$iterator->setArray(array_map(
    static fn (int $number): string => sprintf('photo-%02d.jpg', $number),
    range(1, 20),
));

$childProcessor = new GenerateThumbnails();
$childProcessor->setLogger($logger);

$settings = new Settings(
    iterator: $iterator,
    childProcessor: $childProcessor,
    logger: $logger,
    maxChildren: 4,
    chunkSize: 5,
);

new MultiProcessor($settings)->run();

declare(strict_types=1);

namespace App\Import;

use MultiProcessor\Iterator\IteratorInterface;
use MultiProcessor\Queue\Chunk;
use Override;
use PDO;

final class UnprocessedOrderIterator implements IteratorInterface
{
    private ?PDO $connection = null;
    private int $lastId = 0;

    #[Override]
    public function init(): void
    {
        $this->connection = new PDO('mysql:host=localhost;dbname=shop', 'user', 'secret');
    }

    #[Override]
    public function getChunk(int $size): Chunk
    {
        $statement = $this->connection->prepare(
            'SELECT id FROM orders WHERE processed = 0 AND id > :lastId ORDER BY id LIMIT :size'
        );
        $statement->bindValue('lastId', $this->lastId, PDO::PARAM_INT);
        $statement->bindValue('size', $size, PDO::PARAM_INT);
        $statement->execute();

        $ids = $statement->fetchAll(PDO::FETCH_COLUMN);

        if ($ids !== []) {
            $this->lastId = (int) $ids[array_key_last($ids)];
        }

        return new Chunk($ids);
    }

    #[Override]
    public function getNumberOfChunks(int $chunkSize): int
    {
        $total = (int) $this->connection->query('SELECT COUNT(*) FROM orders WHERE processed = 0')->fetchColumn();

        return (int) ceil($total / $chunkSize);
    }

    #[Override]
    public function dropConnections(): void
    {
        // The child inherited the parent's socket. Let go of it rather than share it.
        $this->connection = null;
    }

    #[Override]
    public function finish(): void
    {
        $this->connection = null;
    }
}

use MultiProcessor\Settings;

$settings = new Settings(
    iterator: $iterator,
    childProcessor: $childProcessor,
    logger: $logger,
    maxChildren: 8,
    chunkSize: 250,
    retryOnFatal: true,
    maxRetries: 2,
    exitOnFatal: false,
);

use MultiProcessor\MultiProcessor;

$multiProcessor = new MultiProcessor($settings);
$multiProcessor->setLogger($logger);
$multiProcessor->run();

use MultiProcessor\Exception\ExceptionInterface;

try {
    new MultiProcessor($settings)->run();
} catch (ExceptionInterface $exception) {
    // Anything MultiProcessor itself failed on.
}