Commands and Queries are KubeMQ's request-reply patterns — see 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 for the broker setup.

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

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.

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:

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.

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:

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 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.

func (p *CQPublisher) SendQuery(
    ctx context.Context,
    channel string,
    msg *message.Message,
    timeout time.Duration,
) (*message.Message, error)

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.

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.

func (p *CQPublisher) SendCommand(
    ctx context.Context,
    channel string,
    msg *message.Message,
    timeout time.Duration,
) (*kubemqSDK.CommandResponse, error)

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.

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.

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 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

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:

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:

func (p *CQPublisher) SendQueryWithCache(
    ctx context.Context,
    channel string,
    msg *message.Message,
    timeout time.Duration,
    cacheKey string,
    cacheTTL time.Duration,
) (*message.Message, error)

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 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

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.<Cmd> and events.<Evt> (the example below uses a watermill-cqrs.<name> prefix).

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 program notes the same.

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.

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

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/.)

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.

Commands & Queries (Request-Reply)

Two Approaches to Request-Reply

Approach 1: requestreply Component over Queues

Approach 2: Native CQPublisher

SendQuery

SendCommand

Timeouts: Per-Call vs DefaultTimeout

Query Caching

CQRS Component

Lifecycle

Runnable Examples

Next Steps

Queues

EventsStore

Reference

On this page