Middleware

Middleware-ul învelește execuția sarcinilor într-un flux în straturi. Fiecare middleware poate rula logică înainte și după sarcină, poate opri execuția înainte de termen sau poate intercepta excepții. Queuety include cinci middleware integrate și acceptă middleware personalizat prin interfața Contracts\Middleware.

Middleware-ul este disponibil pentru clasele de sarcini care pot fi lansate (clasele care implementează Contracts\Job cu traitul Dispatchable).

Interfața Contracts\Middleware

use Queuety\Contracts\Middleware;

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

Parametrul $job este instanța sarcinii procesate. Apelați $next( $job ) pentru a transmite execuția următorului middleware din flux (sau metodei handle() a sarcinii, dacă acesta este ultimul middleware).

Adăugarea middleware-ului la o sarcină

Definiți o metodă middleware() în clasa sarcinii, care returnează o matrice cu instanțe 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 ),
        ];
    }
}

Ordinea de execuție a fluxului

Middleware-urile se aplică în ordinea matricei, primul fiind stratul exterior. Dacă metoda middleware() returnează [A, B, C], ordinea de execuție este:

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

Fiecare middleware îl învelește pe următorul. A rulează primul înaintea sarcinii și ultimul după aceasta, fiind ideal pentru măsurarea timpului sau pentru blocuri try/catch.

Scrierea unui middleware personalizat

Implementați interfața 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" );
        }
    }
}

Folosiți-l pentru o sarcină:

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

Oprirea înainte de termen

Un middleware poate omite complet sarcina dacă nu apelează $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 încorporat

ThrottlesExceptions

Limitează excepțiile unui handler de sarcini pentru a preveni valuri de sarcini. Dacă un handler a generat prea multe excepții într-un interval de timp, sarcinile următoare pentru acel handler sunt respinse cu un RuntimeException în loc să fie procesate. Astfel, un serviciu extern defect nu consumă toate încercările de reluare pentru multe sarcini dintr-odată.

use Queuety\Middleware\ThrottlesExceptions;

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

Numărul excepțiilor este urmărit per clasă de handler în tabelul de blocări al Queuety și este partajat între toate procesele worker. Când se atinge pragul, middleware-ul generează un RuntimeException fără să apeleze sarcina, iar workerul tratează acest caz ca pe orice altă eroare: încercarea se contabilizează pentru max_attempts, sarcina este reluată cu intervalul normal de așteptare și ajunge în carantină când se epuizează încercările.

Exemplu: dacă un API extern nu funcționează, primele 10 eșecuri sunt procesate normal (fiecare mărește contorul de excepții). Încercările următoare eșuează imediat, fără a apela API, până când trec 10 minute fără o nouă excepție; intervalul repornește la fiecare excepție înregistrată.

RateLimited

Impune un număr maxim de execuții într-un interval de timp. Dacă limita este depășită în middleware, se generează un RateLimitExceededException; workerul îl tratează ca pe o încercare eșuată obișnuită, o reia după intervalul normal de așteptare și o contabilizează pentru max_attempts. După ce middleware-ul înregistrează limita în limitatorul de rată al workerului, sarcinile următoare ale handlerului sunt, de asemenea, puse înapoi în pending înainte de execuție, cât timp limita este atinsă (consultați Limitarea ratei).

use Queuety\Middleware\RateLimited;

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

Limitatorul de rată folosește instanța internă RateLimiter a Queuety, astfel încât limitele sunt impuse pentru toate procesele worker.

Timeout

Activează o limită de timp pentru execuția fiecărei sarcini cu pcntl_alarm, cu intenția de a genera un TimeoutException dacă sarcina durează prea mult. Limita de timp nu este aplicată în prezent (#10): Queuety nu activează semnalele asincrone, astfel că handlerul de alarmă nu rulează în timpul execuției sarcinii. Middleware-ul resetează alarma și la final, anulând alarma proprie de expirare a sarcinii a workerului.

use Queuety\Middleware\Timeout;

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

Necesită extensia pcntl PHP. Dacă pcntl nu este disponibil, middleware-ul trece mai departe fără să facă nimic (nu se aplică nicio limită de timp).

UniqueJob

Împiedică executarea simultană a sarcinilor cu aceeași cheie. Folosește un tabel de blocări în baza de date pentru a garanta că rulează o singură instanță odată. Dacă există deja o blocare, sarcina este omisă fără notificare (nu se reia).

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

Împiedică suprapunerea execuțiilor sarcinilor care au aceeași cheie logică, cu expirarea automată a blocării pentru a preveni blocările persistente cauzate de workerii opriți neașteptat. Similar cu UniqueJob, dar conceput pentru operațiuni de lungă durată, unde blocarea trebuie eliberată automat după expirarea unui interval.

use Queuety\Middleware\WithoutOverlapping;

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

Dacă există deja o blocare, sarcina este omisă fără notificare. Blocările expirate sunt eliminate automat înaintea fiecărei încercări de obținere a unei blocări.

Exemplu: combinarea middleware-urilor

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

Această sarcină va:

1. Fi oprită dacă rulează mai mult de 5 minute (Timeout)
2. Fi omisă dacă este deja în curs un alt import pentru același furnizor (WithoutOverlapping)
3. Fi limitată la 5 importuri pe minut pentru toți furnizorii (RateLimited)