Middleware

El middleware envuelve la ejecución de trabajos en una canalización con estructura de cebolla. Cada middleware puede ejecutar lógica antes y después del trabajo, omitir su ejecución o capturar excepciones. Queuety incluye cinco middleware integrados y admite middleware personalizados mediante la interfaz Contracts\Middleware.

El middleware está disponible para clases de trabajo que se pueden despachar (clases que implementan Contracts\Job con el trait Dispatchable).

La interfaz Contracts\Middleware

use Queuety\Contracts\Middleware;

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

El parámetro $job es la instancia del trabajo que se está procesando. Llame a $next( $job ) para pasar la ejecución al siguiente middleware de la canalización (o al método handle() del trabajo si este es el último middleware).

Añadir middleware a un trabajo

Defina un método middleware() en la clase del trabajo que devuelva una matriz de instancias de 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 ),
        ];
    }
}

Orden de ejecución de la canalización

El middleware se aplica en el orden de la matriz, y el primero forma la capa más externa. Si su método middleware() devuelve [A, B, C], el orden de ejecución es:

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

Cada middleware envuelve al siguiente. A se ejecuta primero antes del trabajo y al final después del trabajo, por lo que es ideal para medir el tiempo o envolverlo en un bloque try/catch.

Escribir middleware personalizado

Implemente la interfaz 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" );
        }
    }
}

Úselo en un trabajo:

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

Omitir la ejecución

Un middleware puede omitir por completo el trabajo si no llama 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 integrado

ThrottlesExceptions

Limita las excepciones de un controlador de trabajo para evitar avalanchas de trabajos. Si un controlador ha producido demasiadas excepciones durante un intervalo de tiempo, los trabajos posteriores de ese controlador se rechazan con un RuntimeException en lugar de procesarse. Así se evita que un servicio externo con fallos agote todos los intentos de reintento de muchos trabajos a la vez.

use Queuety\Middleware\ThrottlesExceptions;

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

Los recuentos de excepciones se registran por clase de controlador en la tabla de bloqueos de Queuety y se comparten entre todos los procesos de trabajo. Al alcanzar el umbral, el middleware lanza un RuntimeException sin llamar al trabajo, y el trabajador lo trata como cualquier otro fallo: el intento cuenta para max_attempts, el trabajo se reintenta con la espera habitual y queda en la cola de trabajos fallidos cuando se agotan los intentos.

Ejemplo: si una API externa está caída, los primeros 10 fallos se procesan con normalidad (cada uno incrementa el contador de excepciones). Los intentos posteriores fallan rápidamente sin llamar a API hasta que transcurran 10 minutos sin una nueva excepción; el intervalo se reinicia con cada excepción registrada.

RateLimited

Impone un número máximo de ejecuciones durante un intervalo de tiempo. Si se supera el límite dentro del middleware, se lanza un RateLimitExceededException; el trabajador lo gestiona como un intento fallido normal, lo reintenta con la espera habitual y lo cuenta para max_attempts. Una vez que el middleware registra el límite en el limitador de tasa del trabajador, los trabajos posteriores de ese controlador también se devuelven a pending antes de ejecutarse mientras se alcance el límite (consulte Limitación de tasa).

use Queuety\Middleware\RateLimited;

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

El limitador de tasa usa la instancia interna RateLimiter de Queuety, por lo que los límites se aplican a todos los procesos de trabajo.

Timeout

Activa un tiempo de espera de ejecución por trabajo con pcntl_alarm y está diseñado para lanzar un TimeoutException si el trabajo tarda demasiado. Actualmente, el tiempo de espera no se aplica (#10): Queuety no habilita las señales asíncronas, por lo que el controlador de la alarma no se ejecuta mientras el trabajo está en curso. El middleware también restablece la alarma al finalizar, lo que cancela la alarma de tiempo de espera del trabajo del propio trabajador.

use Queuety\Middleware\Timeout;

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

Requiere la extensión pcntl PHP. Si pcntl no está disponible, el middleware continúa sin avisar (no se aplica ningún tiempo de espera).

UniqueJob

Evita la ejecución simultánea de trabajos con la misma clave. Usa una tabla de bloqueos en la base de datos para garantizar que solo se ejecute una instancia a la vez. Si ya hay un bloqueo, el trabajo se omite silenciosamente (sin reintento).

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

Evita que se solapen las ejecuciones de trabajos que comparten una clave lógica, con una caducidad automática del bloqueo para evitar bloqueos permanentes si los trabajadores se bloquean. Es similar a UniqueJob, pero está diseñado para operaciones de larga duración en las que el bloqueo debe liberarse automáticamente tras un tiempo de espera.

use Queuety\Middleware\WithoutOverlapping;

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

Si ya hay un bloqueo, el trabajo se omite silenciosamente. Los bloqueos caducados se limpian automáticamente antes de cada intento de adquisición.

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

Este trabajo:

1. Se cancelará si tarda más de 5 minutos (Timeout)
2. Se omitirá si ya hay otra importación en curso para el mismo proveedor (WithoutOverlapping)
3. Tendrá un límite de tasa de 5 importaciones por minuto entre todos los proveedores (RateLimited)