Middleware

Il middleware avvolge l'esecuzione dei job in una pipeline a strati. Ogni componente può eseguire logica prima e dopo il job, interromperne l'esecuzione o intercettare eccezioni. Queuety include cinque middleware integrati e supporta middleware personalizzati tramite l'interfaccia Contracts\Middleware.

Il middleware è disponibile per le classi di job che possono essere distribuite (classi che implementano Contracts\Job con il trait Dispatchable).

L'interfaccia Contracts\Middleware

use Queuety\Contracts\Middleware;

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

Il parametro $job è l'istanza del job in elaborazione. Chiamate $next( $job ) per passare l'esecuzione al middleware successivo nella pipeline (o al metodo handle() del job, se questo è l'ultimo middleware).

Aggiungere middleware a un job

Definite un metodo middleware() nella classe del job che restituisca un array di istanze di middleware:

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 ),
        ];
    }
}

Ordine di esecuzione della pipeline

I middleware vengono applicati nell'ordine dell'array; il primo è il livello più esterno. Se il metodo middleware() restituisce [A, B, C], l'ordine di esecuzione è:

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

Ogni middleware avvolge quello successivo. A viene eseguito per primo prima del job e per ultimo dopo il job, il che lo rende ideale per misurare i tempi o racchiudere l'esecuzione in un blocco try/catch.

Scrivere middleware personalizzato

Implementate l'interfaccia 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" );
        }
    }
}

Usatelo in un job:

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

Interruzione anticipata

Un middleware può saltare completamente il job omettendo la chiamata a $next():

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

        $next( $job );
    }
}

Middleware integrato

ThrottlesExceptions

Limita le eccezioni generate dal gestore di un job per prevenire raffiche di job. Se un gestore ha generato troppe eccezioni in un intervallo di tempo, i job successivi per quel gestore vengono rifiutati con un RuntimeException invece di essere elaborati. In questo modo si evita che un servizio esterno in errore esaurisca tutti i tentativi di ripetizione di molti job contemporaneamente.

use Queuety\Middleware\ThrottlesExceptions;

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

I conteggi delle eccezioni vengono registrati per classe del gestore nella tabella dei lock di Queuety e condivisi tra tutti i processi worker. Al raggiungimento della soglia, il middleware genera un RuntimeException senza chiamare il job; il worker lo tratta come qualsiasi altro errore: il tentativo viene conteggiato in max_attempts, il job viene ripetuto con il normale backoff e finisce in coda morta quando esaurisce i tentativi.

Esempio: se un API esterno è inattivo, i primi 10 errori vengono elaborati normalmente (ognuno incrementa il contatore delle eccezioni). I tentativi successivi falliscono subito, senza chiamare API, finché non trascorrono 10 minuti senza nuove eccezioni; l'intervallo riparte a ogni eccezione registrata.

RateLimited

Impone un numero massimo di esecuzioni in un intervallo di tempo. Se il limite viene superato all'interno del middleware, viene generato un RateLimitExceededException; il worker lo gestisce come un normale tentativo non riuscito, lo ripete con il backoff abituale e lo conteggia in max_attempts. Dopo che il middleware ha registrato il limite nel limitatore di frequenza del worker, i job successivi per quel gestore vengono rimessi in pending prima dell'esecuzione finché il limite resta raggiunto (vedere Limitazione della frequenza).

use Queuety\Middleware\RateLimited;

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

Il limitatore di frequenza usa l'istanza interna RateLimiter di Queuety, quindi i limiti vengono applicati a tutti i processi worker.

Timeout

Imposta un timeout di esecuzione per job con pcntl_alarm e dovrebbe generare un TimeoutException se il job impiega troppo tempo. Al momento il timeout non viene applicato (#10): Queuety non abilita i segnali asincroni, quindi il gestore dell'allarme non viene eseguito durante l'esecuzione del job. Inoltre, quando il middleware termina, reimposta l'allarme, annullando il timeout del job impostato dal worker.

use Queuety\Middleware\Timeout;

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

Richiede l'estensione pcntl PHP. Se pcntl non è disponibile, il middleware prosegue senza segnalazioni (il timeout non viene applicato).

UniqueJob

Impedisce l'esecuzione simultanea di job con la stessa chiave. Usa una tabella di lock nel database per assicurarsi che venga eseguita una sola istanza alla volta. Se un lock è già attivo, il job viene saltato senza segnalazioni (e non viene ripetuto).

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

Impedisce che si sovrappongano le esecuzioni di job con la stessa chiave logica, usando una scadenza automatica del lock per evitare che resti attivo dopo un arresto anomalo del worker. È simile a UniqueJob, ma pensato per operazioni di lunga durata in cui il lock deve essere rilasciato automaticamente dopo un timeout.

use Queuety\Middleware\WithoutOverlapping;

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

Se un lock è già attivo, il job viene saltato senza segnalazioni. I lock scaduti vengono eliminati automaticamente prima di ogni tentativo di acquisizione.

Esempio: combinare middleware

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 ),
        ];
    }
}

Questo job:

1. Viene interrotto se dura più di 5 minuti (Timeout)
2. Viene saltato se è già in corso un'altra importazione per lo stesso fornitore (WithoutOverlapping)
3. È soggetto a un limite di 5 importazioni al minuto tra tutti i fornitori (RateLimited)