PHP code example of popphp / pop-queue

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

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

    

popphp / pop-queue example snippets

use Pop\Queue\Queue;
use Pop\Queue\Adapter\File;
use Pop\Queue\Process\Job;

// Create a job and add it to a queue
$job = Job::create(function() {
    echo 'This is job' . PHP_EOL;
});

$queue = new Queue('pop-queue', new File(__DIR__ . '/queue'));
$queue->addJob($job);

use Pop\Queue\Queue;
use Pop\Queue\Worker;
use Pop\Queue\Adapter\File;

// Call up the queue and pass it to a worker object
$queue  = new Queue('pop-queue', new File(__DIR__ . '/queue'));
$worker = Worker::create($queue);

// Trigger the worker to work the next job across its queues
$worker->workAll();

use Pop\Queue\Queue;
use Pop\Queue\Adapter\File;
use Pop\Queue\Process\Task;

$task = Task::create(function() {
    echo 'This is a scheduled task' . PHP_EOL;
})->every30Minutes();

// Add to a queue
$queue = new Queue('pop-queue', new File(__DIR__ . '/queue'));
$queue->addTask($task);

use Pop\Queue\Queue;
use Pop\Queue\Worker;
use Pop\Queue\Adapter\File;

// Call up the queue and pass it to a worker object
$queue  = new Queue('pop-queue', new File(__DIR__ . '/queue'));
$worker = Worker::create($queue);

// Trigger the worker to run the next scheduled task across its queues
$worker->runAll();

var_dump($job->hasJobId());
$id = $job->getJobId();

var_dump($job->hasStarted()); // Has a started timestamp
var_dump($job->hasNotRun());  // No started timestamp and no completed timestamp
var_dump($job->isRunning());  // Has a started timestamp, but not a completed/failed
var_dump($job->isComplete()); // Has a completed timestamp
var_dump($job->hasFailed());  // Has a failed timestamp
var_dump($job->getStarted());
var_dump($job->getCompleted());
var_dump($job->getFailed());

use Pop\Queue\Process\Job;

// Create a job from a closure
$job1 = Job::create(function() {
    echo 'This is job #1' . PHP_EOL;
});

// Create a job from a static class method
$job2 = Job::create('MyApp\Service\SomeService::doSomething');

// Create a job with parameters
$closure = function($num) {
    echo 'This is job ' . $num . PHP_EOL;
};

// The second argument passed is the callable's parameter(s) 
$job3 = Job::create($closure, 1);

use Pop\Queue\Queue;
use Pop\Queue\Worker;
use Pop\Queue\Adapter\File;
use Pop\Queue\Process\Job;

// Create a job that needs the application object
$job = Job::create(function($application) {
    // Do something with the application
});

$queue = new Queue('pop-queue', new File(__DIR__ . '/queue'));
$queue->addJob($job);

use Pop\Queue\Queue;
use Pop\Queue\Worker;
use Pop\Queue\Adapter\File;
use Pop\Application;

$application = new Application();

$queue  = new Queue('pop-queue', new File(__DIR__ . '/queue'));
$worker = Worker::create($queue, $application);

// When the worker works the job, it will push the application object to the job
$worker->workAll();

$queue = new Queue('pop-queue', new File(__DIR__ . '/queue'));
$queue->work($application);

use Pop\Queue\Process\Job;

// Call an instance method - the class is instantiated when the job is worked
$job1 = Job::create('MyApp\Service\DigestService->handle', ['[email protected]']);

// Call a static method
$job2 = Job::create('MyApp\Service\DigestService::handleStatic', ['[email protected]']);

// Instantiate a class, passing the parameters to its constructor - the
// constructed object becomes the job's results
$job3 = Job::create('MyApp\Service\DigestService', ['[email protected]']);

namespace MyApp\Service;

class DigestService
{
    public function handle($application, string $to): string
    {
        // Do the work; the returned value becomes the job's results
        return 'sent to ' . $to;
    }
}

use Pop\Queue\Queue;
use Pop\Queue\Adapter\File;
use Pop\Queue\Process\Job;

// Create a job from an application command and add to the queue
$job   = Job::command('hello world');
$queue = new Queue('pop-queue', new File(__DIR__ . '/queue'));
$queue->addJob($job);

// For the route 'greet <name>'
$job = Job::command('greet Nick');

$job = Job::command(['notify', 'Hello there, world']);

$queue  = new Queue('pop-queue', new File(__DIR__ . '/queue'));
$worker = Worker::create($queue, $application);

use Pop\Queue\Process\Job;

// Create a job from an executable CLI command
$job = Job::exec('ls -la');

$job = Job::exec(['ls', '-la']);

use Pop\Queue\Process\Job;

$job = Job::create(function() {
    echo 'This is job #1' . PHP_EOL;
});
$job->setMaxAttempts(10);

$job->setMaxAttempts(0);

var_dump($job->hasExceededMaxAttempts());

use Pop\Queue\Process\Job;

$job = Job::create(function() {
    echo 'This is job #1' . PHP_EOL;
});

// Wait 30 seconds before every retry
$job->setBackoff(30);

// Or a per-attempt schedule: 10s after the 1st failure, 30s after the 2nd,
// 60s after the 3rd and every failure after that
$job->setBackoff([10, 30, 60]);

// Available in 60 seconds
$job->delay(60);

// Or at an absolute time
$job->delay('2026-12-01 09:00:00');

// Interrupt the job if it runs longer than 30 seconds
$job->setTimeout(30);

use Pop\Queue\Queue;
use Pop\Queue\Adapter\File;
use Pop\Queue\Process\Task;

// Create a scheduled task and add to the queue
$task = Task::create(function() {
    echo 'This is job #1' . PHP_EOL;
})->every5Minutes();

$queue = new Queue('pop-queue', new File(__DIR__ . '/queue'));
$queue->addTask($task);

use Pop\Queue\Process\Task;

$task = Task::create(function() {
    echo 'This is job #1' . PHP_EOL;
});

// Submit a cron-formatted schedule string
$task->schedule('* */2 1,15 1-4 *')

// Submit a non-standard cron-formatted schedule string
// that 

use Pop\Queue\Process\Task;

$task = Task::create(function() {
    echo 'This is job #1' . PHP_EOL;
});
// Using a valid date/time string
$task->every30Minutes()->runUntil('2027-11-30 23:59:59');

// Using a valid UNIX timestamp
$task->every30Minutes()->runUntil(1827619199);

use Pop\Queue\Process\Task;

$task = Task::create(function() {
    echo 'This is job #1' . PHP_EOL;
})->every30Minutes();   // due for all of :00 and :30

$task->setGracePeriod(10);   // $task->gracePeriod(10) is an alias

$task->setGracePeriod(0);

use Pop\Queue\Adapter\Redis;

$adapter = new Redis();

$adapter = new Redis('my.redis.server.com', 6380, 'my-queue');

$adapter = new Redis('my.redis.server.com', 6380, 'my-queue', 'FILO', 30, 'my-password');

$adapter = new Redis('my.redis.server.com', 6380, 'my-queue', null, 60, 'my-password', [
    'stream' => [
        'ssl' => [
            'verify_peer'      => true,
            'cafile'           => '/path/to/ca.pem',
            'verify_peer_name' => true
        ]
    ]
]);

use Pop\Queue\Adapter\Database;
use Pop\Db\Db;

$db = Db::mysqlConnect([
    'database' => 'DATABASE',
    'username' => 'DB_USER',
    'password' => 'DB_PASS'
]);

$adapter = new Database($db); 

$adapter = new Database($db, 'my_queue_jobs'); 

$adapter = new Database($db, 'my_queue_jobs', 'FILO', 30); // 30-second lease, FILO priority

use Pop\Queue\Adapter\File;

$adapter = new File(__DIR__ . '/queues'); 

$adapter = new File(__DIR__ . '/queues', 'FILO', 30); // 30-second lease, FILO priority

use Pop\Queue\Adapter\Memory;

$adapter = new Memory();

$adapter = new Memory(30, 'FILO'); // 30-second lease, FILO priority

use Pop\Queue\Queue;

$queue = Queue::fake(); // same as Queue::create('pop-queue', new Memory())

$queue = Queue::fake('test-queue', 'FILO', 30);

use Pop\Queue\Adapter\Sqs;
use Aws\Sqs\SqsClient;

$client = new SqsClient([
        'key'    => 'AWS_KEY',
        'secret' => 'AWS_SECRET',
    ],
    'region'  => 'AWS_REGION',
    'version' => 'AWS_VERSION'
]);

$adapter = new Sqs($client, 'YOUR_AWS_QUEUE_URL');

use Pop\Queue\Queue;

$queue = Queue::create('pop-queue', $adapter); 

use Pop\Queue\Queue;
use Pop\Queue\Adapter\File;
use Pop\Queue\Process\Job;
use Pop\Queue\Process\Task;

$job1 = Job::create(function() {
    echo 'This is job #1' . PHP_EOL;
});

$job2 = Job::create(function() {
    echo 'This is job #2' . PHP_EOL;
});

$task1 = Task::create(function() {
    echo 'This is scheduled task #1' . PHP_EOL;
})->every30Minutes();

$task2 = Task::create(function() {
    echo 'This is scheduled task #2' . PHP_EOL;
})->sundays();

$queue = new Queue('pop-queue', new File(__DIR__ . '/queue'), Queue::FILO);
$queue->addJobs([$job1, $job2])
    ->addTasks([$task1, $task2]);

use Pop\Queue\Queue;
use Pop\Queue\Adapter\File;

$queue = new Queue('pop-queue', new File(__DIR__ . '/queue'), Queue::FILO);

// Or set it after the fact
$queue->setPriority(Queue::FILO);
$queue->setPriority('FILO'); // the constants are just these two strings

$queue->getPriority(); // 'FIFO' or 'FILO'
$queue->isFifo();      // bool
$queue->isFilo();      // bool

use Pop\Queue\Process\PayloadSigner;

PayloadSigner::setKey($_ENV['QUEUE_SIGNING_KEY']);

use Pop\Event\Manager;

$events = new Manager();
$events->on('queue.job.post', function($job, $queue) {
    echo 'Job ' . $job->getJobId() . ' completed on queue ' . $queue->getName() . PHP_EOL;
});

$queue->setEvents($events);

use Pop\Queue\Queue;
use Pop\Queue\Adapter\File;

// Call up the queue and pass it to a worker object
$queue1  = new Queue('pop-queue1', new File(__DIR__ . '/queue1'));
$queue2  = new Queue('pop-queue2', new File(__DIR__ . '/queue2'));
$worker = Worker::create([$queue1, $queue2]);

$worker->work('pop-queue1');

$worker->workAll();

$worker = Worker::create();
$worker->addQueue($queue1, 10); // high weight
$worker->addQueue($queue2, 1);  // low weight

$job = $worker->work(); // tries $queue1 first, then $queue2

$worker->run('pop-queue1');

$worker->runAll();

$worker->addQueue($queue1);
$worker->addQueue($queue2, 10);   // with a weight
$worker->addQueues([$queue3, $queue4]);

$worker->getQueue('pop-queue1');   // ?Queue - null if not registered
$worker->hasQueue('pop-queue1');   // bool
$worker->getWeight('pop-queue1');  // int - 0 if never weighted
$worker->getQueues();              // array of all queues, in weight order

// Array access, keyed by queue name
$worker['pop-queue1'] = $queue1;
$queue = $worker['pop-queue1'];
isset($worker['pop-queue1']);
unset($worker['pop-queue1']);

// Property access, same thing
$worker->{'pop-queue1'} = $queue1;
$queue = $worker->{'pop-queue1'};

// Countable and iterable
count($worker);
foreach ($worker as $name => $queue) {
    // ...
}

$queue = new Queue('pop-queue1', $adapter);

$worker['some-other-name'] = $queue;

isset($worker['some-other-name']); // false
isset($worker['pop-queue1']);      // true - keyed by the queue's own name

$worker->getApplication(); // ?Application ($worker->application() is an alias)
$worker->hasApplication(); // bool

$worker->setName('billing-worker-01'); // getName() / hasName()
$worker->setRegistry($registry);       // getRegistry() / registry() / hasRegistry()

$worker->workLoop(); // runs forever, servicing jobs as they arrive

use Pop\Event\Manager;

$events = new Manager();
$events->on('worker.work_loop.idle', function($worker) {
    echo 'No work found this pass' . PHP_EOL;
});

$worker->setEvents($events);

use Pop\Event\Manager;

$events = new Manager();

$events->on('queue.job.post', function($job, $queue) {
    StatsD::timing("queue.{$queue->getName()}.duration_seconds", $job->getDuration());
});

$events->on('queue.job.failed', function($job, $queue, $exception) {
    Log::warning("job {$job->getJobId()} failed: {$exception->getMessage()}");
});

$queue->setEvents($events);

use Pop\Queue\Registry\WorkerRegistry;
use Pop\Queue\Registry\Adapter\Redis as RegistryRedis;

$registry = new WorkerRegistry(new RegistryRedis());

$worker->setName('billing-worker-01'); // optional label
$worker->setRegistry($registry);

$worker->workLoop();

$registry = new WorkerRegistry(new RegistryRedis());

foreach ($registry->getWorkers() as $record) {
    printf("%s on %s (pid %d) - %d done, %d failed%s\n",
        $record->getName() ?? $record->getId(),
        $record->getHost(),
        $record->getPid(),
        $record->getJobsProcessed(),
        $record->getJobsFailed(),
        ($record->getCurrentJobId() !== null)
            ? " - running {$record->getCurrentJobId()} for {$record->getCurrentJobDuration()}s"
            : ' - idle'
    );
}

$registry->countWorkers();
$registry->getStaleWorkers();  // heartbeat has gone quiet
$registry->getStuckWorkers();  // quiet AND holding a job past its own timeout
$registry->prune();            // reap records untouched for an hour (pass seconds to change)

bash
composer 

bash
$ php worker.php workLoop &
$ php worker.php runLoop &

bash
composer