Improved messaging

[brendt]

We want to add significant improvements to our event and command busses. A first step is @aidan-casey drafting a proper outline of what these features should look like.

[aidan-casey]

RFC is in progress and I will have this published here this week. Just a few internals I want to flesh out a bit.

[aidan-casey]

Tempest CQRS

Warning

The following is not intended to be an exhaustive RFC with final naming and definitions. It is intended as a discussion point to ensure that we are aligned on the underlying manner in which the buses will work.

Definitions

Term	Definition
Message	A serializable data object dispatched through a bus
Command	A specific type of message expressing intent to change state
Query	A specific type of message requested data
Event	A specific type of message describing that something has occured
Handler	A callable responsible for processing a message
Transport	A message delivery mechanism (local, RabbitMQ, Kafka, etc.)
Envelope	A wrapper containing a message payload and metadata
Metadata	Structured headers attached to a message

Out-of-the-Box Developer Experience

The out-of-the-box CQRS experience should be much like the existing one.

Command Bus

// Defining a command...
#[Message(name: 'shelter.pet.adopt', version: 1)]
final readonly class AdoptPetCommand
{
    public function __construct(
        public Uuid $id,
    ) {}
}

// Dispatching a command...
$commandBus = get(CommandBus::class);

$commandBus->dispatch(new AdoptPetCommand(
    id: new Uuid(),
));

// Handling a command...
final class AdoptPetCommandHandler
{
    #[CommandHandler]
    public function __invoke(AdoptPetCommand $command, Metadata $metadata)
    {
        // ...
    }
}

Query Bus

// Defining a query...
#[Message(name: 'shelter.pet.list-available-pets', version: 1)]
final readonly class ListAvailablePetsQuery
{
    public function __construct(
        public int $page,
        public int $count,
        public string $sortField,
        public string $sortDirection,
    ) {}
}

// Request a query result...
$queryBus = get(QueryBus::class);

/** @var ListAvailablePetsResult $users */
$pets = $queryBus->execute(
    new ListAvailablePetsQuery(
        page: 1,
        count: 10,
        sortField: 'name',
        sortDirection: 'asc',
    )
);

// Handle a query...
final class ListAvailablePetsQueryHandler
{
    #[QueryHandler]
    public function __invoke(ListAvailablePetsQuery $query): ListAvailablePetsResult
    {
        // ...
    }
}

Event Bus

// Defining an event...
#[Message(name: 'shelter.pet.was-adopted', version: 2)]
final readonly class PetWasAdoptedV2
{
    public function __construct(
        public string $id,
    ) {}
}

// Dispatching an event...
$eventBus = get(EventBus::class);

$eventBus->publish(new PetWasAdoptedV2(
    id: new Uuid(),
));

// Handling an event...
final class SendThankYouEmail
{
    #[EventHandler]
    public function __invoke(PetWasAdoptedV2 $event, Metadata $metadata)
    {
        // ...
    }
}

Internals

Message Bus

All buses are abstractions of the core message bus, which involves a lightweight router. This lightweight router enables the developer to utilize other transports (e.g., RabbitMQ, Kafka, etc.) on a case-by-case basis while also starting simple with the built-in functionality Tempest provides.

A message bus only accepts an Envelope object, relying on higher level abstractions to wrap POPO objects into an envelope.

Configuration

return new MessageBusConfig(
    transports: [
        // Create a new "local" transport (default).
        new LocalTransport()->addRule(
            // Route all messages to the local "messages" channel.
            new MatchAll()->to('messages'),
        ),
        
        // Create a new "RabbitMQ" transport.
        new RabbitMQTransport(host: 'localhost')->addRules(
            // Route commands to the commands channel.
            new MessageTypeRule(MessageType::COMMAND)->to('commands'),
            
            // Route events to the events channel.
            new MessageTypeRule(MessageType::EVENT)->to('events'),
            
            // Route queries to the queries channel.
            new MessageTypeRule(MessageType::QUERY)->to('queries'),
        ),
    ],
    
    messageRegistry: new MetadataAttributeRegistry(),
    
    serializer: new DefaultMessageSerializer(),
);

API

interface MessageBus
{
    public function dispatch(Envelope $message): MessageResult;
}

Message Registry

A message registry tracks a string form of the message name to the message object within the codebase. While many kinds of message registries can exist, the default Tempest registry is the MessageAttributeRegistry which relies on discovery to determine these details from the Message attribute.

For example:

#[Message(name: 'shelter.pet.was-adopted', version: 2)]
final class PetWasAdoptedV2

Deserializes all messages with the name of shelter.pet.adopted.v2 to the PetWasAdoptedV2 object.

Envelopes & Metadata

Envelopes contain the POPO object as well as metadata/headers for the message. Throughout these examples, you may have noticed the Metadata object available to handlers. This parameter is optional, but useful for storing and retrieving peripheral information about the messages such as correlation IDs, etc.

Metadata may be enriched by middleware.

Envelope

new Envelope(
    message: 'test',
    metadata: new Metadata([
        'Tempest-Message-Type' => MessageType::COMMAND,    
    ]),
);

Envelope Factory

Envelope factories simply take a POPO and wrap it in an envelope. Envelope factories can be specified on each bus individually and come with their own middleware stack which can enrich metadata.

For example:

return new CommandBusConfig(
    envelopeFactory: new DefaultEnvelopeFactory(
        middleware: []
    )
);

Workers

Tip

If no worker is registered, but handlers are discovered, the handlers will be executed at the end of the HTTP request or CLI command.

Workers ensure that messages are handled outside of the primary application, preventing them from blocking the HTTP response or CLI result. Workers typically run as a CLI command and are encouraged for non-blocking execution.

new MessageBusWorker(channels: ['commands', 'events'])->run();

Questions

Should we add a MessageResult object?

Adding a message result object may allow us to ensure that certain types of messages are waited for.

For example, the following will execute within the primary request, avoiding any worker configured.

// Bypasses the worker and executes synchronously.
get(CommandBus::class)
    ->dispatch(
        new AdoptPetCommand(id: new Uuid())
    )
    ->await();

The primary benefit in my mind here is that queries could execute synchronously all the time.

Resources & Inspiration

• https://masstransit.io
• https://watermill.io
• https://wolverinefx.net
• https://github.com/KhronosPHP/khronos

---

CC: @brendt, @innocenzi, @xHeaven

[brendt]

Thanks! I have a couple of questions:

• Would the #[Message] attribute be required, or only needed if you wanted to add metadata or change defaults? (eg. the "default name" is the class name, the default version is "1").
• IDK how I feel about one interface to rule them all, because a message bus in the context of events is different than one in the context of queries (return vs no return). I think that if we try to have everything fit into one paradigm, it'll hurt DX
• What about middleware? Something I especially like with our current middleware design is that input/output is properly typed:

interface EventBusMiddleware
{
    public function __invoke(string|object $event, EventBusMiddlewareCallable $next): void;
}

interface CommandBusMiddleware
{
    public function __invoke(object $command, CommandBusMiddlewareCallable $next): void;
}

This again circles back to the previous point of trying to fit everything into one interface, while I don't think they can without hurting DX.

• Finally, I think routing is super powerful. I find the proposed config syntax a bit clunky, but I also don't have any alternatives at the moment. That's ok, we can finetune along the way.

[aidan-casey]

Would the #[Message] attribute be required, or only needed if you wanted to add metadata or change defaults? (eg. the "default name" is the class name, the default version is "1").

It's required. Class name has huge potential to break and is really a bad practice.

IDK how I feel about one interface to rule them all, because a message bus in the context of events is different than one in the context of queries

Maybe I didn't make this clear, but that's not what is happening here. Command, Event, and Query all have their own class which is calling the base message bus under the hood. Those interfaces would be unique to those bus types.

What about middleware? Something I especially like with our current middleware design is that input/output is properly typed

Unnecessary IMO. A message is a message is a message.

Finally, I think routing is super powerful. I find the proposed config syntax a bit clunky

Keep in mind, routing is really not likely to change often here unlike HTTP requests and that we definitely don't want that living with the message or handler.

All ears if there is a better suggestion though. 🙂

[brendt]

It's required. Class name has huge potential to break and is really a bad practice.

Hmm in simple apps that only want to trigger some basic events or async commands, that's quite a lot of overhead and I don't think it would cause issues for such simple setups. I want to make sure the classic use case isn't negatively impacted.

Maybe I didn't make this clear, but that's not what is happening here. Command, Event, and Query all have their own class which is calling the base message bus under the hood. Those interfaces would be unique to those bus types.

Ah, yes, that makes sense, thanks!

Unnecessary IMO. A message is a message is a message.

Could be, let's continue down this path for now.

Keep in mind, routing is really not likely to change often here unlike HTTP requests and that we definitely don't want that living with the message or handler.

True, and it's not a dealbreaker. But I do wanna think about it some more to see if we can come up with something a little nicer

[aidan-casey]

Hmm in simple apps that only want to trigger some basic events or async commands, that's quite a lot of overhead and I don't think it would cause issues for such simple setups. I want to make sure the classic use case isn't negatively impacted.

Technically we can add something like this. I'd just really hate defaulting to it because I've just seen this cause all sorts of problems in Laraland.

True, and it's not a dealbreaker. But I do wanna think about it some more to see if we can come up with something a little nicer

Fair enough!