Ask for an answer¶

A request needs two sides: a subscription that answers it, and a caller that asks it. One side declares a Responder instead of a Handler, and the other calls Bus.Request with a deadline. Exactly one member of the responder's group answers, and the caller gets that reply or an error saying why it didn't come.

Request and reply works on a backend that declares Capabilities.RequestReply. The in-memory backend does.

Answer requests¶

messaging.SubscriptionSpec{
    Name:    "table-setup",
    Pattern: "tables.setup",
    Shape:   messaging.Competing,
    Group:   "ingress",
    Responder: func(ctx context.Context, d messaging.Delivery) (cloudevents.Event, error) {
        var req SetupRequest
        if err := d.Decode(&req); err != nil {
            return cloudevents.Event{}, err
        }

        outcome := setUp(ctx, req) // "set up", or a reason it was refused
        return replyEvent(outcome), nil
    },
}

The event the Responder returns is the reply. A responder must be Competing: a request goes to one member of the group, and only that member answers. The bus also needs a Settings.Source.

A refusal is a reply, not an error. "No slot left" or "that name is taken" is your application's answer, and it belongs in the reply's payload. Return an error only when the responder failed, not when its answer is "no".

When the responder does fail, by returning an error, panicking, or returning an event that cannot be sent, the bus answers for it with a failure reply. The caller gets ErrResponderFailed straight away, not a timeout. The error's text stays in the responder's process, where it is logged and counted, because it can carry detail the asking process has no business seeing.

Ask¶

ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()

reply, err := bus.RequestValue(ctx, "tables.setup", "scoutdm.tables.setup.v1", SetupRequest{Name: "orcs"})
if err != nil {
    return err
}

var outcome SetupOutcome
if err := reply.Decode(&outcome); err != nil {
    return err
}

The context must carry a deadline. A request without one is refused with ErrNoDeadline before anything is sent, because a request with no deadline, sent to a responder that has died, waits for ever. How long an answer is worth waiting for is your decision, and the bus picks no default.

RequestValue encodes through the bus's codec and needs Settings.Source, as PublishValue does. To set the event's ID or schema yourself, build the event and call Request.

What each error means¶

Refused before anything is sent, and not counted: ErrNoDeadline; ErrNoReply for a context that had already ended; ErrNotReady when the bus has no live activation; ErrNotSupported on a backend that does not answer requests; and ErrReservedAttribute for an event carrying messagingfailed, the attribute that marks a failure reply.

ErrNoResponders means nothing is subscribed, not that no responder is. If only observers or plain handlers subscribe to the subject, the request is delivered, nobody answers, and you get ErrNoReply at the deadline.

The bus never retries for you. A retry inside the bus would hide the count you need, and is only safe when the responder is idempotent, which only you know.

Count requests by conversation¶

messaging.Settings{
    RequestNames: []string{"table-setup", "audit"},
}

reply, err := bus.Request(ctx, "tables.setup", event, messaging.WithRequestName("table-setup"))

bus.RequestStatsFor("table-setup") is that conversation's row, RequestStatsFor("") is requests sent without a name, and bus.RequestStats() is the total. Names are declared up front, like subscription names, so the number of rows is fixed by your code and not by your traffic. A name you didn't declare is counted with the unnamed requests, and logged once; RequestStatsFor with such a name reads as zero.

Once nothing is in flight, Sent equals Answered + NoResponders + NoReply + Failed. Unclaimed is replies that arrived after their request had finished: a responder slower than its caller's deadline, or a second group answering the same subject. It is a bus-wide count and appears only in the total.

On the answering side, ConsumerStats.Replies says what became of each reply: Replied, Refused by the backend, or Unaddressed.

Moving from publish to request¶

A responder still runs for a plain publish on its subject. The reply has nowhere to go, so it is discarded and counted Unaddressed. That means you can turn a subscription into a responder while publishers are still publishing to it: their work still gets done, and you can move them to Request one at a time.

Rules the bus enforces¶

Each of these is refused when the bus is built, and again by a late Subscribe:

• a subscription with both a Handler and a Responder;
• a responder that is Fanout, or AtLeastOnce: a redelivered request would answer nobody;
• two responders in different groups whose patterns can select the same subject, because both would answer it;
• a Handler in a responder's group, which would take requests it cannot answer;
• a responder on a backend without Capabilities.RequestReply, or on a bus without Source.

The bus can't see other processes' subscriptions. Run one responder group per subject across your deployment. A second group's reply is counted Unclaimed rather than delivered.