# Commands & Queries (Request-Reply) (/integrations/watermill/how-to/commands-queries)
Commands and Queries are KubeMQ's request-reply patterns — see [RPC](/learn/rpc) for what they guarantee at the broker level. This page documents how the plugin exposes them to a Watermill application.
Most of the `watermill-kubemq` plugin maps Watermill's `message.Publisher` and `message.Subscriber` onto KubeMQ's pub/sub patterns. Request-reply is different: Watermill's core Pub/Sub interface has no built-in concept of a synchronous reply, so the plugin offers **two** ways to do it, with different trade-offs.
* **Watermill `requestreply` component** — standard Publisher/Subscriber wired through Watermill's `requestreply` package. Every Watermill middleware applies, but the request and the reply each travel through a queue (two hops), which adds latency.
* **Native `CQPublisher`** — a separate API that wraps KubeMQ's own Commands and Queries APIs. A single hop, lower latency, with execution confirmation and (for queries) server-side caching — but it bypasses the Watermill Router and middleware entirely.
This page covers both, starting with the high-level choice and then drilling into the `CQPublisher` API.
The Watermill plugin is a native gRPC client — it dials the broker's gRPC port (`50000`) directly. There is **no HTTP connector and no enable flag** for any pattern, including Commands and Queries. As long as the broker is up and `50000` is reachable, request-reply works. See [Getting Started](/integrations/watermill/tutorials/getting-started) for the broker setup.
## Two Approaches to Request-Reply [#two-approaches-to-request-reply]
The two approaches answer different questions. If you already have a Watermill Router with middleware (retry, correlation, poison queue, tracing) and want request-reply to participate in that pipeline, use the `requestreply` component. If you want the lowest possible latency for a synchronous call and are willing to step outside the Router, use `CQPublisher`.
| | `requestreply` component | Native `CQPublisher` |
| ------------------------------- | ---------------------------------------- | ------------------------------------- |
| Transport | KubeMQ Queues (or Events) | KubeMQ Commands / Queries |
| Hops | Two (request queue + reply queue) | One (native request-reply) |
| Latency | Higher | Lower |
| Watermill middleware | Applies | Bypassed |
| Implements `message.Publisher`? | Yes (uses standard Publisher/Subscriber) | **No** — separate API |
| Execution confirmation | Via handler result | Built in (`CommandResponse.Executed`) |
| Query caching | No | Yes (`CacheKey` / `CacheTTL`) |
## Approach 1: requestreply Component over Queues [#approach-1-requestreply-component-over-queues]
The `requestreply` component coordinates a request and its reply over a normal Watermill Publisher/Subscriber pair. With KubeMQ you back it with the **Queues** pattern for reliable delivery on both legs. A `PubSubBackend` is built from a reply publisher and a subscriber constructor, plus the topics it publishes replies to and listens on.
```go title="request-reply-watermill.go"
import (
"github.com/ThreeDotsLabs/watermill/components/cqrs"
"github.com/ThreeDotsLabs/watermill/components/requestreply"
"github.com/ThreeDotsLabs/watermill/message"
kubemq "github.com/kubemq-io/watermill-kubemq/pkg/kubemq"
)
// OrderResult is the typed reply returned by the handler.
type OrderResult struct {
OrderID string `json:"order_id"`
Status string `json:"status"`
Total float64 `json:"total"`
}
// replyPub publishes replies back over Queues.
replyPub, _ := kubemq.NewPublisher(kubemq.PublisherConfig{
Address: "localhost:50000",
Pattern: kubemq.PatternQueues,
Logger: logger,
})
replyTimeout := 10 * time.Second
backend, err := requestreply.NewPubSubBackend[OrderResult](
requestreply.PubSubBackendConfig{
Publisher: replyPub,
SubscriberConstructor: func(params requestreply.PubSubBackendSubscribeParams) (message.Subscriber, error) {
return kubemq.NewSubscriber(kubemq.SubscriberConfig{
Address: "localhost:50000",
Pattern: kubemq.PatternQueues,
MaxItems: 1,
WaitTimeoutSeconds: 5,
Logger: logger,
})
},
GeneratePublishTopic: func(params requestreply.PubSubBackendPublishParams) (string, error) {
return "watermill-adv.order-replies", nil
},
GenerateSubscribeTopic: func(params requestreply.PubSubBackendSubscribeParams) (string, error) {
return "watermill-adv.order-replies", nil
},
Logger: logger,
ListenForReplyTimeout: &replyTimeout,
},
requestreply.BackendPubsubJSONMarshaler[OrderResult]{},
)
```
The handler is registered on a `CommandProcessor` with `requestreply.NewCommandHandlerWithResult`. The backend automatically publishes the returned `OrderResult` to the reply topic, and the sender retrieves it with `SendWithReply`:
```go title="request-reply-watermill.go"
// Register a command handler that returns a typed result.
cp.AddHandlers(
requestreply.NewCommandHandlerWithResult[OrderCommand, OrderResult](
"order-handler",
backend,
func(ctx context.Context, cmd *OrderCommand) (OrderResult, error) {
return OrderResult{
OrderID: cmd.OrderID,
Status: "confirmed",
Total: float64(cmd.Qty) * 9.99,
}, nil
},
),
)
// Send a command and block until the typed reply arrives.
reply, err := requestreply.SendWithReply[OrderResult](ctx, commandBus, backend, cmd)
if err != nil {
log.Fatalf("SendWithReply error: %v", err)
}
if reply.Error != nil {
log.Fatalf("Reply error: %v", reply.Error)
}
fmt.Printf("Reply: %+v\n", reply.HandlerResult)
```
Because this path runs through the Router and `CommandProcessor`, every middleware you add to the Router (retry, recoverer, correlation ID, tracing) wraps both the request handling and the reply. The cost is the extra hop: the request goes through `watermill-adv.order-commands` and the reply comes back through `watermill-adv.order-replies`. The full program is in [`examples/advanced/request-reply-watermill`](https://github.com/kubemq-io/kubemq-watermill/blob/main/examples/advanced/request-reply-watermill/main.go).
## Approach 2: Native CQPublisher [#approach-2-native-cqpublisher]
`CQPublisher` is a standalone type that wraps KubeMQ's native Commands and Queries APIs. It does **not** implement `message.Publisher` — it is a separate API with its own `SendCommand` / `SendQuery` methods. Construct it with `NewCQPublisher` and a `CQConfig`:
```go title="cqpublisher.go"
import (
"time"
"github.com/ThreeDotsLabs/watermill"
"github.com/ThreeDotsLabs/watermill/message"
kubemq "github.com/kubemq-io/watermill-kubemq/pkg/kubemq"
)
cq, err := kubemq.NewCQPublisher(kubemq.CQConfig{
Address: "localhost:50000",
DefaultTimeout: 5 * time.Second,
})
if err != nil {
log.Fatal(err)
}
defer cq.Close()
```
`CQConfig` accepts the same connection fields as the other configs (`Address`, `ClientID`, `AuthToken`, `TLS`, or an `ExistingClient`), plus `DefaultTimeout` and the cache fields described below. Like the rest of the plugin it dials the broker over gRPC on port `50000`.
On the responder side there is no Watermill equivalent — Watermill's Pub/Sub interface has no request-reply server. The runnable examples answer commands and queries with the `kubemq-go/v2` SDK directly (`SubscribeToCommands` / `SubscribeToQueries` and `SendCommandResponse` / `SendQueryResponse`). `CQPublisher` is the **client** side that bridges Watermill messages to KubeMQ requests.
### SendQuery [#sendquery]
`SendQuery` sends a Watermill message as a KubeMQ Query and returns the reply as a Watermill `*message.Message` — the response body becomes the `Payload` and the response tags become `Metadata`.
```go
func (p *CQPublisher) SendQuery(
ctx context.Context,
channel string,
msg *message.Message,
timeout time.Duration,
) (*message.Message, error)
```
```go title="send-query.go"
msg := message.NewMessage(watermill.NewUUID(), []byte("get-order-123"))
replyMsg, err := cq.SendQuery(ctx, "watermill-cq.send-query", msg, 10*time.Second)
if err != nil {
log.Fatalf("SendQuery failed: %v", err)
}
// Response body -> Payload, response tags -> Metadata.
fmt.Printf("Query reply payload: %s\n", string(replyMsg.Payload))
fmt.Printf("content-type=%s, source=%s\n",
replyMsg.Metadata.Get("content-type"),
replyMsg.Metadata.Get("source"))
```
A query is only considered successful if the responder marked it executed. If the underlying response has `Executed == false`, `SendQuery` returns an error (`watermill-kubemq: query not executed: ...`) carrying the responder's error message, so a query that reaches a responder but fails server-side surfaces as a Go error rather than an empty reply. See [`examples/queries/send-query`](https://github.com/kubemq-io/kubemq-watermill/blob/main/examples/queries/send-query/main.go).
### SendCommand [#sendcommand]
`SendCommand` sends a Watermill message as a KubeMQ Command. Commands are fire-and-forget **with execution confirmation**: instead of a reply message you get a `kubemq-go` `*CommandResponse`, whose `Executed` field tells you whether the responder ran successfully.
```go
func (p *CQPublisher) SendCommand(
ctx context.Context,
channel string,
msg *message.Message,
timeout time.Duration,
) (*kubemqSDK.CommandResponse, error)
```
```go title="send-command.go"
msg := message.NewMessage(watermill.NewUUID(), []byte("process-order-123"))
resp, err := cq.SendCommand(ctx, "watermill-cq.send-command", msg, 10*time.Second)
if err != nil {
log.Fatalf("SendCommand failed: %v", err)
}
fmt.Printf("Command response: Executed=%v\n", resp.Executed)
```
Use a command when you need to know the action was carried out but do not need a data payload back; use a query when you need the responder to return data. See [`examples/commands/send-command`](https://github.com/kubemq-io/kubemq-watermill/blob/main/examples/commands/send-command/main.go).
### Timeouts: Per-Call vs DefaultTimeout [#timeouts-per-call-vs-defaulttimeout]
Both `SendCommand` and `SendQuery` take a `timeout` argument. When it is `<= 0`, the call falls back to `CQConfig.DefaultTimeout`. And `CQConfig.Validate` sets `DefaultTimeout` to **5 seconds** if you leave it unset — so a `CQPublisher` always has an effective timeout even if you never configure one.
```go title="command-timeout.go"
cq, _ := kubemq.NewCQPublisher(kubemq.CQConfig{
Address: "localhost:50000",
DefaultTimeout: 5 * time.Second,
})
// Per-call override: time out after 2s regardless of DefaultTimeout.
_, err := cq.SendCommand(ctx, "watermill-cq.command-timeout", msg1, 2*time.Second)
// Pass 0 to fall back to DefaultTimeout (5s here).
_, err = cq.SendCommand(ctx, "watermill-cq.command-timeout", msg2, 0)
```
If no responder is listening, the call blocks until the effective timeout elapses and then returns a `send command error` / `send query error`. The [`examples/commands/command-timeout`](https://github.com/kubemq-io/kubemq-watermill/blob/main/examples/commands/command-timeout/main.go) example measures both the per-call (2s) and default (5s) paths against an empty channel.
`DefaultTimeout` defaulting to 5s is **source behavior** (`CQConfig.Validate`). The README's `CQConfig` table lists `DefaultTimeout` as required and does not document the fallback, so do not rely on the table alone — set `DefaultTimeout` explicitly when 5s is not the value you want.
### Query Caching [#query-caching]
KubeMQ Queries support a server-side cache: when a query carries a cache key, the broker caches the first response and serves subsequent matching queries from cache until the TTL expires, without re-invoking the responder. The plugin exposes this two ways.
**Config-level (applies to every query).** Set `CacheKey` and `CacheTTL` on `CQConfig`. When `CacheKey` is non-empty, `SendQuery` attaches both to each query it sends:
```go title="config-cache.go"
cq, _ := kubemq.NewCQPublisher(kubemq.CQConfig{
Address: "localhost:50000",
DefaultTimeout: 5 * time.Second,
CacheKey: "orders-cache",
CacheTTL: 30 * time.Second,
})
```
**Per-query.** `SendQueryWithCache` sets the cache key and TTL for a single call, overriding the config-level settings:
```go
func (p *CQPublisher) SendQueryWithCache(
ctx context.Context,
channel string,
msg *message.Message,
timeout time.Duration,
cacheKey string,
cacheTTL time.Duration,
) (*message.Message, error)
```
```go title="cached-query.go"
// First call: cache miss — the responder is invoked.
reply1, _ := cq.SendQueryWithCache(ctx, "watermill-cq.cached-query", msg1,
10*time.Second, // timeout
"price-key-ABC", // cacheKey
30*time.Second, // cacheTTL
)
// Second call with the same key: cache hit — the responder is NOT invoked.
reply2, _ := cq.SendQueryWithCache(ctx, "watermill-cq.cached-query", msg2,
10*time.Second,
"price-key-ABC",
30*time.Second,
)
```
In [`examples/queries/cached-query`](https://github.com/kubemq-io/kubemq-watermill/blob/main/examples/queries/cached-query/main.go) the responder counts its invocations and confirms it ran exactly once across two identical cached queries.
The `CacheKey` and `CacheTTL` fields exist on `CQConfig` in the source, but they are **not** listed in the README's `CQConfig` configuration table. Treat the source as the authority for these fields.
## CQRS Component [#cqrs-component]
For full Command Query Responsibility Segregation, Watermill ships a `cqrs` component that you wire with KubeMQ transports: **Queues** for commands (reliable delivery) and **EventsStore** for domain events (persistence and replay). Commands and events are routed to channels by name using a topic naming convention such as `commands.` and `events.` (the example below uses a `watermill-cqrs.` prefix).
```go title="cqrs-facade.go"
facade, err := cqrs.NewFacade(cqrs.FacadeConfig{
// Commands via Queues (reliable delivery).
GenerateCommandsTopic: func(commandName string) string {
return fmt.Sprintf("watermill-cqrs.%s", commandName)
},
CommandsPublisher: cmdPub, // PatternQueues publisher
CommandsSubscriberConstructor: func(handlerName string) (message.Subscriber, error) {
return kubemq.NewSubscriber(kubemq.SubscriberConfig{
Address: "localhost:50000",
Pattern: kubemq.PatternQueues,
MaxItems: 1,
WaitTimeoutSeconds: 5,
Logger: logger,
})
},
CommandHandlers: func(cb *cqrs.CommandBus, eb *cqrs.EventBus) []cqrs.CommandHandler {
return []cqrs.CommandHandler{
cqrs.NewCommandHandler[CreateOrderCmd]("CreateOrderHandler",
func(ctx context.Context, cmd *CreateOrderCmd) error {
// ... handle command, then publish a domain event:
return eb.Publish(ctx, &OrderCreatedEvt{OrderID: cmd.OrderID})
},
),
}
},
// Events via EventsStore (persistent, replayable).
GenerateEventsTopic: func(eventName string) string {
return fmt.Sprintf("watermill-cqrs.%s", eventName)
},
EventsPublisher: evtPub, // PatternEventsStore publisher
EventsSubscriberConstructor: func(handlerName string) (message.Subscriber, error) {
return kubemq.NewSubscriber(kubemq.SubscriberConfig{
Address: "localhost:50000",
Pattern: kubemq.PatternEventsStore,
EventsStoreStartOption: kubemq.StartFromNew,
Logger: logger,
})
},
EventHandlers: func(cb *cqrs.CommandBus, eb *cqrs.EventBus) []cqrs.EventHandler {
return []cqrs.EventHandler{
cqrs.NewEventHandler[OrderCreatedEvt]("OrderCreatedHandler",
func(ctx context.Context, evt *OrderCreatedEvt) error {
fmt.Printf("Order %s created\n", evt.OrderID)
return nil
},
),
}
},
Router: router,
CommandEventMarshaler: cqrs.JSONMarshaler{GenerateName: cqrs.StructName},
Logger: logger,
})
```
A command sent through `facade.CommandBus().Send(ctx, cmd)` is delivered over Queues to its handler, which publishes a domain event over EventsStore that the event handler then processes — a complete command-to-event flow on KubeMQ transports.
`cqrs.Facade` is **deprecated in Watermill v1.5.1** (the version this plugin pins). The modern approach wires `CommandProcessor` and `EventProcessor` (and their bus counterparts) directly. The Facade is shown here for reference and backward compatibility; the [`examples/cqrs/facade`](https://github.com/kubemq-io/kubemq-watermill/blob/main/examples/cqrs/facade/main.go) program notes the same.
## Lifecycle [#lifecycle]
`CQPublisher` owns a gRPC client when you construct it with an `Address` (rather than passing your own `ExistingClient`). Always `defer cq.Close()` so the connection is released. `Close` is idempotent — calling it more than once is safe — and it only closes the underlying client when the `CQPublisher` created it.
After `Close`, the publisher is unusable: `SendCommand`, `SendQuery`, and `SendQueryWithCache` all check the closed flag first and return `watermill-kubemq: CQPublisher is closed` before touching the broker.
```go title="lifecycle.go"
cq, err := kubemq.NewCQPublisher(kubemq.CQConfig{
Address: "localhost:50000",
DefaultTimeout: 5 * time.Second,
})
if err != nil {
log.Fatal(err)
}
defer cq.Close() // release the gRPC client on exit
// ... SendQuery / SendCommand ...
// Any Send after Close returns:
// watermill-kubemq: CQPublisher is closed
```
## Runnable Examples [#runnable-examples]
With a broker running on `localhost:50000`, run any of these from the repository root. (The README refers to an `_examples/` path, but the actual directory is `examples/`.)
```bash
go run ./examples/queries/send-query/main.go
go run ./examples/queries/cached-query/main.go
go run ./examples/commands/send-command/main.go
go run ./examples/commands/command-timeout/main.go
go run ./examples/advanced/request-reply-cq/main.go
```
The `advanced/request-reply-cq` example exercises both `SendCommand` and `SendQuery` against a single `kubemq-go` responder; the `queries` and `commands` examples isolate each method. For the middleware-compatible path, run `./examples/advanced/request-reply-watermill/main.go`.
## Next Steps [#next-steps]