[dgafka]

Why is this change proposed?

Why

A service that sends a message through the Distributed Bus can now get the receiving handler's result back on a routing key of its own choice.

Before this change, DistributedBus::sendMessage() could only send a message one way. If the sender needed the result, the receiving service had to know about the sender and send a second message back itself. That puts the sender's routing inside the receiver.

Before / after

Before

sequenceDiagram
    participant User as user_service
    participant Ticket as ticket_service
    User->>Ticket: sendMessage('ticket.create')
    Note over Ticket: handler result is discarded

Loading

After

sequenceDiagram
    participant User as user_service
    participant Ticket as ticket_service
    User->>Ticket: sendMessage('ticket.create', replyRoutingKey: 'ticket.created')
    Ticket-->>User: handler result, through ticket_service's Service Map
    Note over User: 'ticket.created' handler runs

Loading

Example

In user_service, which sends the request and handles the reply:

$distributedBus->sendMessage('ticket_service', 'ticket.create', 'Printer is broken', replyRoutingKey: 'ticket.created');

final class TicketNotifications
{
    #[Distributed]
    #[InternalHandler('ticket.created')]
    public function whenCreated(string $ticket): void
    {
    }
}

In ticket_service, which handles the request and returns the result:

final class Tickets
{
    #[Distributed]
    #[InternalHandler('ticket.create')]
    public function create(string $description): string
    {
        return 'ticket for: ' . $description;
    }
}

Description of Changes

• Reply routing: the receiving service picks the channel back to the sender from its own DistributedServiceMap. A reply is refused with a message that names how to fix it in three cases: the request names no source service, no Service Map maps the sender, or two Service Maps map the sender to different channels. Start reading at DistributedReplyRouter.
• What a reply carries: the request's metadata, its correlation id, and the sender's trace. The reply's parent id is the request. Nothing is sent for a null result, and a reply is never answered again.
• Asynchronous targets: the reply is sent once the asynchronous work has been consumed. It goes straight to the Service Map channel and skips the receiver's asynchronous bus channel. Over DBAL, a reply that cannot be sent rolls back the handler, and the retried request replies once.
• Renamed parameter: sendMessage()'s targetChannelName is now routingKey, the name the other four DistributedBus methods use. Positional calls are unchanged. A named targetChannelName: argument fails, and upgrade-2.0.md records the rename.
• Message flow graph: a #[Distributed] #[InternalHandler] is drawn with its reply edge. The reply flow is part of the shared flow catalogue, so tracing, OpenTelemetry, flow tracking and the graph are all checked against it.

The Distributed Bus with a Service Map is Enterprise, so replies are Enterprise too.

Pull Request Contribution Terms

• I have read and agree to the contribution terms outlined in CONTRIBUTING.