mle86/wq-beanstalkd

A Beanstalkd module for mle86/wq using the pda/pheanstalk connector

v1.0.3 2020-06-18 13:53 UTC

Requires

• php: >=7.1
• mle86/wq: ^v1.0.0 || ^v0
• pda/pheanstalk: ^3.0.2

Requires (Dev)

• phpunit/phpunit: ^7.4.0
• roave/security-advisories: dev-master

MIT

• Maximilian Eul

README

This package contains the PHP class mle86\WQ\WorkServerAdapter\BeanstalkdWorkServer.

It supplements the mle86/wq package by implementing its WorkServerAdapter interface.

It connects to a Beanstalkd server using the pda/pheanstalk package by Paul Annesley.

Version and Compatibility

This is version 1.0.2 of mle86/wq-beanstalkd.

It was developed for version 1.0.0 of mle86/wq and should be compatible with all of its future 1.x versions as well.

Installation and Dependencies

$ composer require mle86/wq-beanstalkd

It requires PHP 7.1, mle86/wq, and pda/pheanstalk.

Class reference

class mle86\WQ\WorkServerAdapter\BeanstalkdWorkServer implements WorkServerAdapter

getNextQueueEntry() uses the RESERVE command, buryEntry() uses the BURY command, storeJob() and requeueEntry() use the PUT command, and deleteEntry() uses the DELETE command.

Work Queues are Beanstalkd's “tubes”.

• public function __construct (Pheanstalk $pheanstalk)
  Constructor. Takes an already-configured Pheanstalk instance to work with. Does not attempt to establish a connection itself – use the connect() factory method for that instead.
• public static function connect (string $host = "localhost", int $port = PheanstalkInterface::DEFAULT_PORT, int $connectTimeout = null)
  Factory method. See Pheanstalk::__construct for the parameter descriptions.

Interface methods which are documented in the WorkServerAdapter interface:

• public function storeJob (string $workQueue, Job $job, int $delay = 0)
• public function getNextQueueEntry ($workQueue, int $timeout = DEFAULT_TIMEOUT) : ?QueueEntry
• public function buryEntry (QueueEntry $entry)
• public function requeueEntry (QueueEntry $entry, int $delay, string $workQueue = null)
• public function deleteEntry (QueueEntry $entry)

Usage example

<?php
use mle86\WQ\WorkServerAdapter\BeanstalkdWorkServer;
use mle86\WQ\WorkProcessor;
use mle86\WQ\Job\Job;

$processor = new WorkProcessor( BeanstalkdWorkServer::connect("localhost") );

while (true) {
    $processor->processNextJob("mail", function(Job $job) {
        $job->...;
    });
}

This executes all jobs available in the local Beanstalkd server's “mail” tube, forever. It will however abort if one of the jobs throws an exception – you might want to add a logging try-catch block around the processNextJob() call as shown in WQ's “Quick Start” example.