Calling other services (outbound clients + cross-cutting decorators)

A handler often needs to call another service. In Benzene that goes through one port — a MessageSender (async send_message(topic, message, headers) -> Result) — so a handler never couples to a transport, and cross-cutting concerns (retry, correlation ids, trace propagation) are decorators over that one interface. The same wrapping works whether the underlying client is HTTP, gRPC, or a cloud queue.

1. Pick a transport client

Every transport ships a MessageSender. Over HTTP:

from benzene.http import HttpMessageSender

client = HttpMessageSender("https://orders.svc")          # topic -> {base}/{topic}

Over gRPC (pip install 'benzene-grpc[transport]'):

import grpc
from benzene.grpc import GrpcMessageSender

client = GrpcMessageSender(grpc.insecure_channel("orders.svc:50051"))

…or a cloud client (SnsMessageSender, PubSubMessageSender, ServiceBusMessageSender). The handler code below does not change when you swap one for another.

2. Add cross-cutting behaviour by wrapping

The decorators live in benzene.core (retry, correlation id) and benzene.mesh (trace propagation). Each wraps a MessageSender and is one, so they compose — order matters: put with_correlation_id outside with_retry so a retried request keeps one correlation id across all its attempts.

from benzene.core import with_retry, with_correlation_id
from benzene.mesh import with_trace_propagation

outbound = with_correlation_id(              # inject x-correlation-id ONCE (stable across retries)
    with_retry(                              # re-send transient failures (service-unavailable, …)
        with_trace_propagation(client),      # forward the current traceparent to the callee
        attempts=5,
    )
)

• with_correlation_id lets a request be followed across services without each call site remembering. Outermost, so every retry attempt shares the one id (an id the caller already set is left untouched).
• with_retry returns at once on success or a real failure (a not-found won't get better by retrying); pass a backoff hook to sleep between attempts. A peer that never answers is retried too: every client returns a call that got no reply as a timeout (its deadline elapsed) or service-unavailable (refused, reset, unreachable) Result rather than raising.
• with_trace_propagation forwards the trace of the current invocation (set by trace_middleware) — traceparent, and the inbound tracestate unchanged — so the callee joins the same trace. Wrap the sender you publish events through too: a meshed service forwards trace context on every outbound message, sends and published events alike (mesh.md §7.1, M2). The joined trace lets the mesh collector attribute invocation/error stats to the provider edge this service already declares — the sends/events of the Contract Document it registers (mesh.md §2, §4) — the trace never creates the edge itself.

3. Inject it through the composition root

Register the composed client as the MessageSender in your BenzeneStartUp, so handlers receive it and a test can swap it for a fake — the testing seam:

from benzene.core import MessageSender

class OrdersStartUp(BenzeneStartUp):
    def configure_services(self, services, config):
        services.try_add_singleton(MessageSender, lambda scope: outbound)

A handler then just publishes — no transport, no retry logic, no header plumbing in sight:

async def place_order(request: PlaceOrder) -> Result:
    order = store.place(request.sku, request.quantity)
    await sender.send_message("orders:created", OrderCreated(id=order.id, sku=order.sku))
    return Result.created(order)

4. Test it with a fake

The FakeMessageSender records what a handler published — assert on the egress, no network:

from benzene.testing import FakeMessageSender

fake = FakeMessageSender()
host = create_test_host(OrdersStartUp).with_services(
    lambda s: s.add_instance(MessageSender, fake)
).build_aws()

host.send_http("POST", "/orders", body={"sku": "ABC"})
assert fake.last_topic == "orders:created"

See transport-bindings §"Outbound clients" for the reverse-direction contract.

Edit this page on GitHub