PHP code example of tuxweb-it / amp-sql-queue

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

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

    

tuxweb-it / amp-sql-queue example snippets

use Amp\Postgres\PostgresConfig;
use Amp\Postgres\PostgresConnectionPool;
use AmpSqlQueue\DispatchOptions;
use AmpSqlQueue\Queue;
use AmpSqlQueue\Storage\PostgresQueueStorage;
use function Amp\Future\await;

$pool = new PostgresConnectionPool(PostgresConfig::fromString(
    'host=127.0.0.1 port=5432 user=amp_queue password=amp_queue dbname=amp_queue',
));

$queue = new Queue(new PostgresQueueStorage($pool));

$result = $queue->dispatch(
    queue: 'ocr',
    type: 'document.ocr',
    payload: [
        'project_file_id' => 123,
    ],
    options: new DispatchOptions(
        delaySeconds: 0,
        priority: 10,
        maxAttempts: 5,
        idempotencyKey: 'ocr:123',
    ),
)->await();

var_dump($result->jobId, $result->inserted);

use Amp\Mysql\MysqlConfig;
use Amp\Mysql\MysqlConnectionPool;
use AmpSqlQueue\Queue;
use AmpSqlQueue\Storage\MysqlQueueStorage;

$pool = new MysqlConnectionPool(MysqlConfig::fromString(
    'host=127.0.0.1 user=amp_queue password=amp_queue db=amp_queue',
));

$queue = new Queue(new MysqlQueueStorage($pool));

use Amp\Future;
use AmpSqlQueue\Job;
use AmpSqlQueue\Worker;
use function Amp\async;

$worker = new Worker(
    queue: $queue,
    queueName: 'ocr',
    concurrency: 4,
    leaseSeconds: 300,
    handler: static function (Job $job): Future {
        return async(static function () use ($job): void {
            // Do real async work here.
            // Use AMPHP-compatible clients for I/O.
        });
    },
);

$worker->run();

use AmpSqlQueue\WorkerHookEvents;

$worker->addHook(static function (string $event, array $context): void {
    if ($event === WorkerHookEvents::JobFailed) {
        error_log('Job failed: ' . $context['job']->id);
    }
});

$queue->storage()->extendLease($job, 300)->await();

new DispatchOptions(
    maxAttempts: 5,
);

use AmpSqlQueue\Backoff\ExponentialBackoffStrategy;

$worker = new Worker(
    queue: $queue,
    queueName: 'ocr',
    concurrency: 4,
    leaseSeconds: 300,
    handler: $handler,
    backoffStrategy: new ExponentialBackoffStrategy(
        baseDelaySeconds: 2,
        maxDelaySeconds: 300,
        multiplier: 2.0,
        jitterRatio: 0.1,
    ),
);

new DispatchOptions(
    idempotencyKey: 'ocr:123',
);

bash
export AMP_SQL_QUEUE_POSTGRES_DSN='host=127.0.0.1 port=5432 user=amp_queue password=amp_queue dbname=amp_queue'
php examples/dispatch.php
php examples/worker.php