Queue System

Nexphant includes a native, high-performance job queue system built directly into the managed runtime. Jobs can run asynchronously within worker processes, fully integrated with the event loop, ownership tree, and cancellation mechanisms.

Defining Jobs

To create a queue job, extend the Nexphant\Queue\Job class. The properties you pass to the constructor are automatically serialized.

<?php

namespace App\Jobs;

use Nexphant\Queue\Job;

class ProcessVideo extends Job
{
    public function __construct(
        private string $path,
        private string $format
    ) {}

    public function handle(): void
    {
        // Long running task handled inside a worker fiber
        video_processor()->convert($this->path, $this->format);
    }
}

Dispatching Jobs

You can dispatch jobs using the global Queue facade:

use App\Jobs\ProcessVideo;
use Nexphant\Queue\Queue;

// Dispatch immediately to the default queue
Queue::dispatch(new ProcessVideo('/tmp/video.mp4', 'webm'));

// Dispatch with a delay (in seconds)
Queue::dispatch(new ProcessVideo('/tmp/video.mp4', 'webm'), delay: 60);

// Dispatch to a specific queue
Queue::dispatch(new ProcessVideo('/tmp/video.mp4', 'webm'))->onQueue('heavy-tasks');

Worker Configuration

To run a queue worker, create a worker instance and start the loop. A single worker runs continuously, pulling and processing jobs inside fibers:

use Nexphant\Queue\Worker;

$worker = Worker::create(
    queue: 'default',
    rate: 100 // rate-limit: max 100 jobs per minute
);

// Start processing jobs
$worker->work();

Worker Options

Option	Type	Default	Description
queue	string	'default'	Name of the queue to listen to
rate	int	null	Rate limit (maximum jobs per minute)
memory	int	128	Memory limit in megabytes before graceful restart
timeout	int	60	Maximum job execution time in seconds

Resiliency and Retries

You can control failure behavior directly on your job class:

class ProcessVideo extends Job
{
    // The number of times the job may be attempted
    public int $tries = 3;

    // The number of seconds to wait before retrying a failed job
    public int $retryAfter = 30;

    // Automatically release resources if the job times out
    public int $timeout = 120;
}

Graceful Drain

During a server deployment or shutdown, sending a termination signal (SIGTERM) to the workers triggers a graceful drain.

• Workers will stop accepting new jobs from the queue.
• The job currently running will be allowed to complete.
• Once the current job finishes, resources are safely released, and the worker process terminates.