Allow invoking Messages to topics dynamically

[bielu]

Feature Request: Dynamic Topic/Destination Routing for InvokeAsync

Summary

InvokeAsync currently has no way to specify a dynamic topic or destination at the call site without manually constructing a transport-specific URI. This breaks the transport-agnostic design that Wolverine otherwise provides well.

Current Behavior

To route InvokeAsync to a dynamic topic, you must use EndpointFor with a manually constructed URI:

var result = await bus
    .EndpointFor(new Uri($"kafka://topic/{dynamicTopicName}"))
    .InvokeAsync<TResponse>(new MyRequest(), timeout: TimeSpan.FromMinutes(2));

This has several problems:

• Requires the caller to know the transport scheme (kafka://topic/, rabbitmq://, etc.)
• Tight coupling between business logic and transport implementation details
• No equivalent to BroadcastToTopicAsync for request/reply scenarios
• URI construction is error-prone (special characters, scheme differences between transports)

Expected Behavior

Similar to how BroadcastToTopicAsync allows dynamic topic publishing, InvokeAsync should support equivalent dynamic routing:

// Option A — topic overload directly on InvokeAsync
var result = await bus.InvokeAsync<TResponse>(
    new MyRequest(),
    topic: dynamicTopicName,
    timeout: TimeSpan.FromMinutes(2)
);

// Option B — fluent ForTopic() mirroring BroadcastToTopicAsync
var result = await bus
    .ForTopic(dynamicTopicName)
    .InvokeAsync<TResponse>(new MyRequest(), timeout: TimeSpan.FromMinutes(2));

Use Case

A common pattern is dynamically started workers — each worker instance listens on its own topic derived at runtime (e.g. per-tenant, per-device, per-session). The orchestrator needs to invoke a request/reply against a specific worker's topic without hardcoding transport URIs.

Currently BroadcastToTopicAsync covers the fire-and-forget side of this pattern but there is no equivalent for request/reply via InvokeAsync, forcing a transport-specific workaround.

Current Workaround

// Works but transport-specific and fragile
bus.EndpointFor(new Uri($"kafka://topic/{dynamicTopic}"))
   .InvokeAsync<TResponse>(request);

Proposed API

public interface IMessageBus
{
    // Existing
    Task BroadcastToTopicAsync(string topicName, object message, DeliveryOptions? options = null);

    // Proposed — mirrors BroadcastToTopicAsync for request/reply
    Task<T> InvokeForTopicAsync<T>(
        string topicName,
        object message,
        CancellationToken cancellation = default,
        TimeSpan? timeout = default
    ) where T : class;
}

Environment

• Wolverine version: latest
• Transport: Kafka (but issue applies to all transports)
• .NET version: 10

[jeremydmiller]

You can use the endpoint name, and by default, that endpoint name already is the topic name for Kafka, Pulsar, et al. And you already have BroadcastToTopicAsync() as well.

I'm not seeing anything that's really new here, but feel free to go for a pull request I guess. Otherwise, I'm kicking this over to a discussion.

[bielu]

Can you maybe show example as tried few different ways and it didn't worked 😅