Intergiciel

L’intergiciel encadre l’exécution des tâches dans un pipeline en oignon. Chaque intergiciel peut exécuter une logique avant et après la tâche, interrompre l’exécution ou intercepter les exceptions. Queuety fournit cinq intergiciels intégrés et prend en charge les intergiciels personnalisés via l’interface Contracts\Middleware.

L’intergiciel est disponible pour les classes de tâches distribuables (classes qui implémentent Contracts\Job avec le trait Dispatchable).

L’interface Contracts\Middleware

use Queuety\Contracts\Middleware;

interface Middleware {
    public function handle( object $job, \Closure $next ): void;
}

Le paramètre $job est l’instance de la tâche en cours de traitement. Appelez $next( $job ) pour transmettre l’exécution à l’intergiciel suivant du pipeline (ou à la méthode handle() de la tâche s’il s’agit du dernier intergiciel).

Ajouter un intergiciel à une tâche

Définissez une méthode middleware() dans la classe de votre tâche, qui renvoie un tableau d’instances d’intergiciel :

use Queuety\Contracts\Job;
use Queuety\Dispatchable;
use Queuety\Middleware\RateLimited;
use Queuety\Middleware\Timeout;

class SyncInventoryJob implements Job {
    use Dispatchable;

    public function __construct(
        public readonly string $sku,
    ) {}

    public function handle(): void {
        // Sync inventory for this SKU...
    }

    public function middleware(): array {
        return [
            new Timeout( seconds: 30 ),
            new RateLimited( max: 100, window: 60 ),
        ];
    }
}

Ordre d’exécution du pipeline

Les intergiciels sont appliqués dans l’ordre du tableau, le premier formant la couche extérieure. Si votre méthode middleware() renvoie [A, B, C], l’ordre d’exécution est le suivant :

A::handle() -> B::handle() -> C::handle() -> $job->handle()

Chaque intergiciel encadre le suivant. A s’exécute en premier avant la tâche et en dernier après celle-ci, ce qui en fait un choix idéal pour le chronométrage ou les blocs try/catch.

Écrire un intergiciel personnalisé

Implémentez l’interface Contracts\Middleware :

use Queuety\Contracts\Middleware;

class LogExecution implements Middleware {
    public function handle( object $job, \Closure $next ): void {
        $class = get_class( $job );
        error_log( "[Queuety] Starting: {$class}" );

        $start = microtime( true );

        try {
            $next( $job );
        } finally {
            $elapsed = round( microtime( true ) - $start, 3 );
            error_log( "[Queuety] Finished: {$class} in {$elapsed}s" );
        }
    }
}

Utilisez-le sur une tâche :

public function middleware(): array {
    return [
        new LogExecution(),
    ];
}

Interruption anticipée

Un intergiciel peut ignorer entièrement la tâche en n’appelant pas $next() :

class SkipOnMaintenanceMode implements Middleware {
    public function handle( object $job, \Closure $next ): void {
        if ( wp_is_maintenance_mode() ) {
            return; // Job silently skipped.
        }

        $next( $job );
    }
}

Intergiciels intégrés

ThrottlesExceptions

Limite les exceptions d’un gestionnaire de tâches afin d’éviter les tempêtes de tâches. Si un gestionnaire a levé trop d’exceptions pendant une période donnée, les tâches suivantes de ce gestionnaire sont rejetées avec un RuntimeException au lieu d’être traitées. Cela empêche un service externe défaillant d’épuiser en même temps toutes les tentatives de relance de nombreuses tâches.

use Queuety\Middleware\ThrottlesExceptions;

public function middleware(): array {
    return [
        new ThrottlesExceptions( max_attempts: 10, decay_minutes: 10 ),
    ];
}

Le nombre d’exceptions est suivi par classe de gestionnaire dans la table des verrous de Queuety et partagé entre tous les processus workers. Lorsque le seuil est atteint, l’intergiciel lève un RuntimeException sans appeler la tâche. Le worker traite alors cela comme tout autre échec : la tentative est comptabilisée dans max_attempts, la tâche est relancée avec le délai d’attente habituel, puis est mise en attente définitive une fois toutes les tentatives épuisées.

Exemple : si un API externe est indisponible, les 10 premiers échecs sont traités normalement (chacun incrémente le compteur d’exceptions). Les tentatives suivantes échouent rapidement sans appeler API jusqu’à ce que 10 minutes s’écoulent sans nouvelle exception ; la fenêtre redémarre à chaque exception enregistrée.

RateLimited

Impose un nombre maximal d’exécutions pendant une période donnée. Si la limite est dépassée dans l’intergiciel, un RateLimitExceededException est levé ; le worker le traite comme une tentative échouée ordinaire, la relance avec le délai d’attente habituel et la comptabilise dans max_attempts. Une fois la limite enregistrée auprès du limiteur de débit du worker par l’intergiciel, les tâches suivantes de ce gestionnaire sont également remises en pending avant leur exécution tant que la limite est atteinte (voir Limitation du débit).

use Queuety\Middleware\RateLimited;

public function middleware(): array {
    return [
        new RateLimited( max: 60, window: 60 ), // 60 per minute
    ];
}

Le limiteur de débit utilise l’instance RateLimiter interne de Queuety ; les limites sont donc appliquées à tous les processus workers.

Timeout

Déclenche un délai d’expiration d’exécution par tâche avec pcntl_alarm et est censé lever un TimeoutException si la tâche dure trop longtemps. Le délai n’est pas appliqué actuellement (#10) : Queuety n’active pas les signaux asynchrones, donc le gestionnaire d’alarme ne s’exécute pas pendant l’exécution de la tâche. L’intergiciel réinitialise également l’alarme à son retour, ce qui annule l’alarme de délai d’expiration de tâche du worker.

use Queuety\Middleware\Timeout;

public function middleware(): array {
    return [
        new Timeout( seconds: 30 ),
    ];
}

Nécessite l’extension pcntl PHP. Si pcntl n’est pas disponible, l’intergiciel poursuit silencieusement son exécution (aucun délai d’expiration n’est appliqué).

UniqueJob

Empêche l’exécution simultanée de tâches ayant la même clé. Utilise une table de verrous en base de données pour garantir qu’une seule instance s’exécute à la fois. Si un verrou est déjà détenu, la tâche est ignorée silencieusement (sans nouvelle tentative).

use Queuety\Middleware\UniqueJob;

public function middleware(): array {
    return [
        new UniqueJob(),                         // Lock key defaults to the job class name
        new UniqueJob( key: 'sync_inventory' ),   // Custom lock key
    ];
}

WithoutOverlapping

Empêche le chevauchement de tâches partageant une clé logique, avec une expiration automatique du verrou pour éviter les verrous morts causés par des workers interrompus. Similaire à UniqueJob, mais conçu pour les opérations longues dont le verrou doit être libéré automatiquement après un délai.

use Queuety\Middleware\WithoutOverlapping;

public function middleware(): array {
    return [
        new WithoutOverlapping( key: 'import_products', release_after: 600 ),
    ];
}

Si un verrou est déjà détenu, la tâche est ignorée silencieusement. Les verrous expirés sont automatiquement nettoyés avant chaque tentative d’acquisition.

Exemple : combiner des intergiciels

use Queuety\Contracts\Job;
use Queuety\Dispatchable;
use Queuety\Middleware\RateLimited;
use Queuety\Middleware\Timeout;
use Queuety\Middleware\WithoutOverlapping;

class ImportProductsJob implements Job {
    use Dispatchable;

    public function __construct(
        public readonly int $supplier_id,
    ) {}

    public function handle(): void {
        // Fetch and import products from the supplier...
    }

    public function middleware(): array {
        return [
            new Timeout( seconds: 300 ),
            new WithoutOverlapping(
                key: "import_supplier_{$this->supplier_id}",
                release_after: 600,
            ),
            new RateLimited( max: 5, window: 60 ),
        ];
    }
}

Cette tâche :

1. Sera interrompue si elle dure plus de 5 minutes (Timeout)
2. Sera ignorée si un autre import du même fournisseur est déjà en cours (WithoutOverlapping)
3. Sera limitée à 5 imports par minute pour tous les fournisseurs (RateLimited)