Building a PHP client for Faktory, Part 6: Higher-level usage
Tags: queues, php
Published: 2023-11-25 23:43:51

We have the low\-level tools for connecting to the Faktory server, executing commands, fetching data, logging, and so on. Now, I'll work on the high\-level API this Faktory library will expose. I plan to make it flexible enough for a user to use directly, or maybe plug it into their framework.

We need to provide a producer API \(enqueuing jobs\) and a consumer API \(fetching and executing jobs\).

## Producer API

For the producer, we can expose a base `Job` class and a `Dispatcher` which enqueues jobs.

My proposed usage looks like this:

```php
// Create a dispatcher
$dispatcher = Dispatcher::make(
    logLevel: \Monolog\Level::Debug,
    hostname: 'tcp://localhost',
);
// Dispatch a job now
$dispatcher->dispatch(SendWelcomeEmail::class, [$userId]);
// Schedule a job for later
$dispatcher->dispatch(SendWelcomeEmail::class, [$userId], delaySeconds: 60);
// Enqueue multiple instances in bulk
$dispatcher->dispatchMany(
  SendWelcomeEmail::class, [[$user1, $user2]], 
  delaySeconds: 60
);
```

I believe in convenience, so I'll also provide a global dispatcher:

```php
Dispatcher::configure(
    logLevel: \Monolog\Level::Debug,
    hostname: 'tcp://dreamatorium',
);
SendWelcomeEmail::dispatch($userId);
SendWelcomeEmail::dispatchIn(seconds: 60, $userId);
SendWelcomeEmail::dispatchMany(
  inSeconds: 60, [[$user1], [$user2]]
);
```

Okay, implementing. The job class allows a user to specify options for the the job payload sent to Faktory:

```php
abstract class Job
{
  public static ?string $queue = null;

  public static int $retry = 25;

  public static int $reserveFor = 1800;

  public static array $custom = []; // From the Faktory spec; allows users to add custom data

  public static function dispatch(...$args)
  {
    Dispatcher::instance()->dispatch(static::class, $args);
  }

  public static function dispatchIn(?int $seconds, array $args = [])
  {
    Dispatcher::instance()->dispatch(static::class, $args, delaySeconds: $seconds);
  }

  public static function dispatchMany(array $args, ?int $inSeconds = null)
  {
    Dispatcher::instance()->dispatchMany(static::class, $args, delaySeconds: $inSeconds);
  }
}
```

The Dispatcher creates and passes configuration to the Faktory client, and transforms job operations \(enqueueing/scheduling\) to the appropriate Faktory operation:

```php
class Dispatcher
{
  public function dispatch(string $jobClass, array $args = [], int $delaySeconds = null)
  {
    $jobPayload = static::toJobPayload($jobClass, $args, $delaySeconds);
    $this->client->push($jobPayload);
  }

  public function dispatchMany(string $jobClass, array $argumentsListing, int $delaySeconds = null)
  {
    $basicPayload = static::toJobPayload($jobClass, [], $delaySeconds);

    $jobPayloads = [];
    foreach ($argumentsListing as $index => $arguments) {
      $jobPayloads[] = array_merge($basicPayload, [
        "jid" => "{$basicPayload['jid']}_{$index}",
        "args" => $arguments,
      ]);
    }

    return $this->client->pushBulk($jobPayloads);
  }

  protected static function toJobPayload(string $jobClass, array $args, int $delaySeconds = null)
  {
    return PayloadBuilder::build(
      jobType: $jobClass,
      args: $args,
      queue: $jobClass::$queue,
      retry: $jobClass::$retry,
      reserveFor: $jobClass::$reserveFor,
      delaySeconds: $delaySeconds
    );
  }
}
```

In the above snippet, I've omitted some code from the dispatcher, mainly boilerplate like constructors and managing the underlying client. You can view the full code [on GitHub](https://github.com/shalvah/faktory-php/commit/006c684ff8cfb9991d32d63ad7105f8ec97e03f6).

The `PayloadBuilder` class is a helper class to generate the needed payload shape for the Faktory server:

```php
class PayloadBuilder
{
  public static function build(
    string $jobType,
    array $args,
    ?string $queue,
    ?int $retry,
    ?int $reserveFor,
    ?int $delaySeconds,
  ): array
  {
    $payload = [
      'jid' => 'job_' . bin2hex(random_bytes(12)),
      'jobtype' => $jobType,
      'args' => $args,
    ];

    if ($queue) {
      $payload['queue'] = $queue;
    }
    if (is_null($retry)) { // 0 is a possible value, meaning no retries
      $payload['retry'] = $retry;
    }
    if ($reserveFor) {
      $payload['reserve_for'] = $reserveFor;
    }

    if ($delaySeconds) {
      $executionTime = (new \DateTimeImmutable('@'.(time() + $delaySeconds)));
      $payload['at'] = $executionTime->format(\DateTimeInterface::ATOM);
    }

    return $payload;
  }
}
```

That's it for the dispatcher. Next up: retrieving and executing jobs. For this, I'm creating an `Executor` class. It's quite similar to the Dispatcher in terms of managing the client and having a global instance, so I'll omit those parts. Here's the key logic::

```php
class Executor
{
  public function start(array $queues = []): void
  {
    $this->client->connect();

    $queuesListening = empty($queues)
      ? "all queues" : ("queues" . implode(",", $queues));
    $this->logger->info(sprintf("Listening on %s", $queuesListening));

    while (true) {
      $jobPayload = $this->getNextJob($queues);
      $this->processAndReport($jobPayload);
    }
  }

  /**
   * Process a retrieved job, and ACK or FAIL it to Faktory.
   */
  public function processAndReport(array $jobPayload): bool
  {
    try {
      $this->process($jobPayload);
      $this->reportSuccess($jobPayload);
      return true;
    } catch (Throwable $e) {
      $this->reportFailure($jobPayload, $e);
      return false;
    }
  }

  /**
   * Process a retrieved job. This merely instantiates the job and executes it.
   * Any exceptions thrown by the job are not handled.
   */
  public function process(array $jobPayload): void
  {
    $jobInstance = $this->instantiateJob($jobPayload);
    $jobInstance->process(...$jobPayload['args']);
  }

  /**
   * Manually fetch the next job to be executed from the specified queues.
   * Faktory will block for a few seconds if no job available, then return null.
   * The $retryUntilAvailable parameter forces the executor to try again when this happens.
   */
  public function getNextJob(array $queues = ["default"], bool $retryUntilAvailable = true): array|null
  {
    while (true) {
      $job = $this->client->fetch(...$queues);
      if ($job || !$retryUntilAvailable) return $job;
    }
  }

  protected function reportSuccess(array $jobPayload)
  {
    $this->logger->info(sprintf("Processed job result=success class=%s id=%s", $jobPayload['jobtype'], $jobPayload['jid']));
    $this->client->ack(["jid" => $originalJobPayload["jid"]]);
  }

  protected function reportFailure(array $jobPayload, Throwable $exception)
  {
    $this->logger->info(sprintf("Processed job result=failure class=%s id=%s", $jobPayload['jobtype'], $jobPayload['jid']));
    $this->client->fail([
      "jid" => $originalJobPayload["jid"],
      "errtype" => $exception::class,
      "message" => $exception->getMessage(),
      "backtrace" => explode("\n", $exception->getTraceAsString()),
    ]));
  }

  protected function instantiateJob(array $jobPayload): Job
  {
    $class = $jobPayload['jobtype'];
    return new $class;
  }
}
```

I'll bring this article to a close here, because I got interrupted multiple times and ended up spending several months on it. 😅 I'm off to tackle the next stages; [here's](https://github.com/shalvah/faktory-php/commit/006c684ff8cfb9991d32d63ad7105f8ec97e03f6) the code thus far.

Thus far, we have a basic working API. I'm not yet sure how it would fit into a production app, but it's good for a start. In the next phase, I'd like to address some limitations of the current setup.