Middleware

Middleware omhult taakuitvoering in een pijplijn in uienstijl. Elke middleware kan logica uitvoeren vóór en na de taak, de uitvoering vroegtijdig beëindigen of uitzonderingen opvangen. Queuety bevat vijf ingebouwde middlewarecomponenten en ondersteunt aangepaste middleware via de interface Contracts\Middleware.

Middleware is beschikbaar voor verzendbare taakklassen (klassen die Contracts\Job implementeren met de trait Dispatchable).

De interface Contracts\Middleware

use Queuety\Contracts\Middleware;

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

De parameter $job is de taakinstantie die wordt verwerkt. Roep $next( $job ) aan om de uitvoering door te geven aan de volgende middleware in de pijplijn (of aan de methode handle() van de taak als dit de laatste middleware is).

Middleware aan een taak toevoegen

Definieer een methode middleware() in uw taakklasse die een array met middleware-instanties retourneert:

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

Volgorde van pijplijnuitvoering

Middleware wordt toegepast in de volgorde van de array, waarbij de eerste middleware de buitenste laag vormt. Als uw methode middleware() [A, B, C] retourneert, is de uitvoeringsvolgorde:

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

Elke middleware omhult de volgende. A wordt vóór de taak als eerste uitgevoerd en erna als laatste, waardoor deze ideaal is voor timing- of try/catch-wrappers.

Aangepaste middleware schrijven

Implementeer de 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" );
        }
    }
}

Gebruik deze voor een taak:

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

Uitvoering vroegtijdig beëindigen

Middleware kan de taak volledig overslaan door $next() niet aan te roepen:

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

        $next( $job );
    }
}

Ingebouwde middleware

ThrottlesExceptions

Beperkt het aantal uitzonderingen van een taakhandler om stormen van taken te voorkomen. Als een handler binnen een tijdsvenster te veel uitzonderingen heeft veroorzaakt, worden volgende taken voor die handler afgewezen met een RuntimeException in plaats van verwerkt. Zo voorkomt u dat een externe service die niet werkt alle nieuwe pogingen voor veel taken tegelijk verbruikt.

use Queuety\Middleware\ThrottlesExceptions;

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

Uitzonderingstellingen worden per handlerklasse bijgehouden in de lockstabel van Queuety en gedeeld tussen alle workerprocessen. Zodra de drempel is bereikt, genereert de middleware een RuntimeException zonder de taak aan te roepen. De worker behandelt dit als elke andere mislukking: de poging telt mee voor max_attempts, de taak wordt opnieuw geprobeerd met de normale wachttijd en wordt in quarantaine geplaatst zodra alle pogingen zijn verbruikt.

Voorbeeld: als een externe API niet beschikbaar is, worden de eerste 10 mislukkingen normaal verwerkt (waarbij de uitzonderingsteller telkens wordt verhoogd). Verdere pogingen mislukken direct zonder API aan te roepen, totdat er 10 minuten verstrijken zonder nieuwe uitzondering; bij elke geregistreerde uitzondering begint het venster opnieuw.

RateLimited

Dwingt een maximumaantal uitvoeringen binnen een tijdsvenster af. Als de limiet binnen de middleware wordt overschreden, wordt er een RateLimitExceededException gegenereerd. De worker behandelt dit als een gewone mislukte poging: deze wordt opnieuw geprobeerd met de normale wachttijd en telt mee voor max_attempts. Nadat de middleware de limiet bij de snelheidsbegrenzer van de worker heeft geregistreerd, worden latere taken voor de handler bij het bereiken van de limiet ook teruggezet naar pending voordat ze worden uitgevoerd (zie Snelheidsbeperking).

use Queuety\Middleware\RateLimited;

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

De snelheidsbegrenzer gebruikt de interne RateLimiter-instantie van Queuety, zodat de limieten voor alle workerprocessen gelden.

Timeout

Stelt een time-out voor de uitvoering per taak in met pcntl_alarm en is bedoeld om een TimeoutException te genereren wanneer de taak te lang duurt. De time-out wordt momenteel niet afgedwongen (#10): Queuety schakelt asynchrone signalen niet in, waardoor de alarmhandler niet wordt uitgevoerd zolang de taak bezig is. De middleware reset het alarm ook wanneer deze terugkeert, waardoor het eigen alarm voor de time-out van de taak van de worker wordt geannuleerd.

use Queuety\Middleware\Timeout;

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

Vereist de extensie pcntl PHP. Als pcntl niet beschikbaar is, wordt de middleware stilzwijgend overgeslagen (er wordt geen time-out afgedwongen).

UniqueJob

Voorkomt gelijktijdige uitvoering van taken met dezelfde sleutel. Gebruikt een databasetabel voor vergrendelingen om ervoor te zorgen dat er slechts één instantie tegelijk actief is. Als er al een vergrendeling is, wordt de taak stilzwijgend overgeslagen (zonder nieuwe poging).

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

Voorkomt overlappende uitvoering van taken met dezelfde logische sleutel, met automatische vrijgave van de vergrendeling om te voorkomen dat workercrashes permanente vergrendelingen veroorzaken. Vergelijkbaar met UniqueJob, maar ontworpen voor langlopende bewerkingen waarbij de vergrendeling automatisch moet worden vrijgegeven na een time-out.

use Queuety\Middleware\WithoutOverlapping;

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

Als er al een vergrendeling is, wordt de taak stilzwijgend overgeslagen. Verlopen vergrendelingen worden automatisch opgeruimd vóór elke poging om een vergrendeling te verkrijgen.

Voorbeeld: middleware combineren

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

Deze taak wordt:

1. Beëindigd als deze langer dan 5 minuten duurt (Timeout)
2. Overgeslagen als er al een import voor dezelfde leverancier actief is (WithoutOverlapping)
3. Beperkt tot 5 imports per minuut voor alle leveranciers samen (RateLimited)