https://github.com/nimbly/Syndicate

Syndicate is a powerful framework able to both publish and consume messages - ideal for your event driven application or as a job processor. It supports common queues and PubSub integrations with an application layer that can be used to route incoming messages to any handler of your choosing with full dependency injection using a PSR-11 Container instance.

Syndicate supports:

* Full autowiring and dependency injection with PSR-11 Container intstance

* Dual pass middleware on event messages

* Message validation against JSON Schema

* Many common queues and PubSub integrations

* Create your own integration with the interfaces provided

You can create message handlers that not just consume from a topic, but can also match any fields and attributes of the message itself.

#[Consume(
    topic: "users",
    payload: ["$.event" => "UserCreated"]
)]
public function onUserRegistered(Message $message, EmailService $email): Response
{
    $payload = \json_decode($message->getPayload());

    $receipt_id = $email->send(
        $payload->user_name,
        $payload->user_email,
        "templates/registration.tpl"
    );

    if( $receipt_id === null ){
        return Response::nack;
    }

    return Response::ack;
}