Middleware

O middleware envolve a execução de jobs em um pipeline em camadas. Cada middleware pode executar lógica antes e depois do job, interromper a execução ou capturar exceções. Queuety inclui cinco middlewares integrados e aceita middlewares personalizados por meio da interface Contracts\Middleware.

O middleware está disponível para classes de job que podem ser despachadas (classes que implementam Contracts\Job com o trait Dispatchable).

A interface Contracts\Middleware

use Queuety\Contracts\Middleware;

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

O parâmetro $job é a instância do job que está sendo processada. Chame $next( $job ) para passar a execução ao próximo middleware no pipeline (ou ao método handle() do job, se este for o último middleware).

Adicionar middleware a um job

Defina um método middleware() na classe do job que retorne um array de instâncias 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 ),
        ];
    }
}

Ordem de execução do pipeline

Os middlewares são aplicados na ordem do array, sendo o primeiro a camada mais externa. Se o método middleware() retornar [A, B, C], a ordem de execução será:

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

Cada middleware envolve o próximo. A é executado primeiro, antes do job, e por último, após o job, sendo ideal para medir o tempo ou envolver a execução com try/catch.

Criar middleware personalizado

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

Use-o em um job:

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

Interrupção antecipada

Um middleware pode ignorar completamente o job se não chamar $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 as exceções de um handler de job para evitar uma onda de jobs. Se um handler lançar exceções demais em um intervalo de tempo, os jobs seguintes desse handler serão rejeitados com um RuntimeException em vez de serem processados. Isso impede que um serviço externo com falhas consuma todas as tentativas de repetição de vários jobs de uma só vez.

use Queuety\Middleware\ThrottlesExceptions;

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

As contagens de exceções são acompanhadas por classe de handler na tabela de bloqueios de Queuety e compartilhadas entre todos os processos de worker. Quando o limite é atingido, o middleware lança um RuntimeException sem chamar o job, e o worker trata isso como qualquer outra falha: a tentativa conta para max_attempts, o job é repetido com o intervalo normal entre tentativas e é colocado em espera quando todas as tentativas se esgotam.

Exemplo: se um API externo estiver fora do ar, as primeiras 10 falhas são processadas normalmente (cada uma incrementa o contador de exceções). As tentativas seguintes falham rapidamente, sem chamar API, até que se passem 10 minutos sem uma nova exceção; a janela reinicia a cada exceção registrada.

RateLimited

Impõe um número máximo de execuções em um intervalo de tempo. Se o limite for excedido dentro do middleware, será lançado um RateLimitExceededException; o worker o trata como uma tentativa com falha comum, que será repetida com o intervalo normal entre tentativas e contabilizada em max_attempts. Depois que o middleware registra o limite no limitador de taxa do worker, os jobs seguintes desse handler também voltam para pending antes de serem executados enquanto o limite estiver atingido (consulte Limitação de taxa).

use Queuety\Middleware\RateLimited;

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

O limitador de taxa usa a instância interna de Queuety RateLimiter, então os limites são aplicados a todos os processos de worker.

Timeout

Ativa um tempo limite de execução por job com pcntl_alarm e deve lançar um TimeoutException quando o job demorar demais. Atualmente, o tempo limite não é aplicado (#10): Queuety não ativa sinais assíncronos, então o handler do alarme não é executado enquanto o job está em execução. O middleware também redefine o alarme ao retornar, cancelando o próprio alarme de tempo limite de job do worker.

use Queuety\Middleware\Timeout;

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

Requer a extensão pcntl PHP. Se pcntl não estiver disponível, o middleware não fará nada (nenhum tempo limite será aplicado).

UniqueJob

Impede a execução simultânea de jobs com a mesma chave. Usa uma tabela de bloqueios no banco de dados para garantir que apenas uma instância seja executada por vez. Se um bloqueio já estiver ativo, o job será ignorado silenciosamente (sem nova tentativa).

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

Impede a execução simultânea de jobs que compartilham uma chave lógica, com expiração automática do bloqueio para evitar bloqueios permanentes causados por workers que falham. É semelhante a UniqueJob, mas foi projetado para operações longas, nas quais o bloqueio deve ser liberado automaticamente após um tempo limite.

use Queuety\Middleware\WithoutOverlapping;

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

Se um bloqueio já estiver ativo, o job será ignorado silenciosamente. Os bloqueios expirados são limpos automaticamente antes de cada tentativa de aquisição.

Exemplo: combinar middlewares

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 job será:

1. Encerrado à força se durar mais de 5 minutos (Timeout)
2. Ignorado se outra importação do mesmo fornecedor já estiver em andamento (WithoutOverlapping)
3. Limitado a 5 importações por minuto entre todos os fornecedores (RateLimited)