# API Reference (/integrations/watermill/reference/api)



This page documents the public API surface of the `watermill-kubemq` plugin: the three
constructors and the methods on each returned type. The configuration fields those
constructors accept — and the enums, validation rules, and marshaling contract — are in the
[Configuration reference](/integrations/watermill/reference/configuration). Every entry is drawn from the plugin source.

## Constructors [#constructors]

The package exposes three constructors. Each validates its config (mutating it in place to
apply defaults), opens or reuses a `kubemq-go/v2` client, and returns a ready-to-use instance.

| Constructor      | Signature                                | Returns                 |
| ---------------- | ---------------------------------------- | ----------------------- |
| `NewPublisher`   | `NewPublisher(config PublisherConfig)`   | `(*Publisher, error)`   |
| `NewSubscriber`  | `NewSubscriber(config SubscriberConfig)` | `(*Subscriber, error)`  |
| `NewCQPublisher` | `NewCQPublisher(config CQConfig)`        | `(*CQPublisher, error)` |

```go
pub, err := kubemq.NewPublisher(kubemq.PublisherConfig{
    Address: "localhost:50000",
    Pattern: kubemq.PatternEvents,
})

sub, err := kubemq.NewSubscriber(kubemq.SubscriberConfig{
    Address: "localhost:50000",
    Pattern: kubemq.PatternEvents,
})

cq, err := kubemq.NewCQPublisher(kubemq.CQConfig{
    Address:        "localhost:50000",
    DefaultTimeout: 5 * time.Second,
})
```

<Callout type="info">
  When `ExistingClient` is set on any config, the constructor reuses that client and ignores
  `Address`, `ClientID`, `AuthToken`, and `TLS`. In that case the caller owns the client and is
  responsible for closing it; calling `Close()` on the Publisher, Subscriber, or CQPublisher will
  **not** close a client it does not own.
</Callout>

## Publisher API [#publisher-api]

`Publisher` implements Watermill's `message.Publisher` interface and is safe for concurrent
use. Each instance serves a single pattern (Events, EventsStore, or Queues), selected at
construction.

| Method        | Signature                                                   | Description                                                                                  |
| ------------- | ----------------------------------------------------------- | -------------------------------------------------------------------------------------------- |
| `Publish`     | `Publish(topic string, messages ...*message.Message) error` | Publishes one or more messages to `topic` using the configured pattern.                      |
| `Close`       | `Close() error`                                             | Closes the active streaming handle and, if the client is owned, the gRPC client. Idempotent. |
| `HealthCheck` | `HealthCheck(ctx context.Context) error`                    | Pings the broker to verify connectivity.                                                     |

```go
msg := message.NewMessage(watermill.NewUUID(), []byte("hello events"))
if err := pub.Publish("my-topic", msg); err != nil {
    log.Printf("publish error: %v", err)
}
defer pub.Close()
```

A few behaviors worth noting:

* `Publish` returns an error if the publisher has already been closed, or if `topic` is empty.
* By default the publisher uses KubeMQ's streaming send path. Setting `DisableStreaming: true` switches to synchronous per-message sends.
* `Close` is guarded by an atomic compare-and-swap, so calling it more than once is safe and only the first call has any effect.

## Subscriber API [#subscriber-api]

`Subscriber` implements Watermill's `message.Subscriber` interface. A single subscriber serves
one pattern but supports **multiple concurrent `Subscribe` calls** on different topics — each
gets its own goroutine and output channel.

| Method        | Signature                                                                       | Description                                                                                                        |
| ------------- | ------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------ |
| `Subscribe`   | `Subscribe(ctx context.Context, topic string) (<-chan *message.Message, error)` | Subscribes to `topic` and returns a channel of Watermill messages.                                                 |
| `Close`       | `Close() error`                                                                 | Cancels all active subscriptions, waits for their goroutines to drain, and closes the client if owned. Idempotent. |
| `HealthCheck` | `HealthCheck(ctx context.Context) error`                                        | Pings the broker to verify connectivity.                                                                           |

```go
messages, err := sub.Subscribe(context.Background(), "my-topic")
if err != nil {
    log.Fatal(err)
}
for msg := range messages {
    log.Printf("received: %s", string(msg.Payload))
    msg.Ack() // only meaningful for the Queues pattern
}
```

* `Subscribe` returns an error if the subscriber is closed or `topic` is empty.
* For the **Queues** pattern, each delivered message is bridged to KubeMQ settlement: calling `msg.Ack()` acknowledges and removes it, while `msg.Nack()` returns it to the queue for redelivery. For **Events** and **EventsStore**, ack/nack on the Watermill message has no effect on the broker.
* `Close` cancels every active subscription's context and blocks until all subscription goroutines (and per-message ack/nack bridge goroutines) have finished.

## CQPublisher API [#cqpublisher-api]

`CQPublisher` is a **separate API** — it does not implement `message.Publisher`. It wraps
KubeMQ's native Commands and Queries for low-latency request-reply.

| Method               | Signature                                                                                                                                                 | Description                                                     |
| -------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------- |
| `SendCommand`        | `SendCommand(ctx, channel string, msg *message.Message, timeout time.Duration) (*kubemqSDK.CommandResponse, error)`                                       | Sends a command and returns the native KubeMQ command response. |
| `SendQuery`          | `SendQuery(ctx, channel string, msg *message.Message, timeout time.Duration) (*message.Message, error)`                                                   | Sends a query and returns the reply as a Watermill message.     |
| `SendQueryWithCache` | `SendQueryWithCache(ctx, channel string, msg *message.Message, timeout time.Duration, cacheKey string, cacheTTL time.Duration) (*message.Message, error)` | Like `SendQuery`, but with explicit per-call cache key and TTL. |
| `Close`              | `Close() error`                                                                                                                                           | Closes the client if owned. Idempotent.                         |

```go
import "time"

cq, _ := kubemq.NewCQPublisher(kubemq.CQConfig{
    Address:        "localhost:50000",
    DefaultTimeout: 5 * time.Second,
})
defer cq.Close()

// Query (request-reply with a returned payload)
q := message.NewMessage(watermill.NewUUID(), []byte(`{"prompt":"classify this text"}`))
reply, err := cq.SendQuery(ctx, "ml.classify", q, 10*time.Second)
if err != nil {
    log.Fatalf("query failed: %v", err)
}
log.Printf("reply: %s", string(reply.Payload))

// Command (request-reply with execution confirmation)
c := message.NewMessage(watermill.NewUUID(), []byte(`{"action":"deploy"}`))
resp, err := cq.SendCommand(ctx, "ops.deploy", c, 30*time.Second)
```

Behavior notes:

* If `timeout` is `<= 0` on any send call, the `DefaultTimeout` from `CQConfig` is used.
* `SendQuery` and `SendQueryWithCache` return an error when the responder reports the query as not executed.
* When `CacheKey` is set on `CQConfig`, `SendQuery` applies it (and `CacheTTL`) to every query. `SendQueryWithCache` instead takes the cache key and TTL per call.

For the full request-reply walkthrough — including the Watermill `requestreply` component over
Queues — see [Commands & Queries](/integrations/watermill/how-to/commands-queries).

## Make targets [#make-targets]

The repository `Makefile` provides the following targets:

| Target               | Command                                                         | Purpose                                                  |
| -------------------- | --------------------------------------------------------------- | -------------------------------------------------------- |
| `test`               | `go test -v -count=1 ./pkg/kubemq/...`                          | Unit tests.                                              |
| `test-race`          | `go test -v -race -count=1 ./pkg/kubemq/...`                    | Unit tests with the race detector.                       |
| `test-integration`   | `go test ... -tags=integration -timeout=5m ./pkg/kubemq/...`    | Integration tests (require a running broker).            |
| `test-compatibility` | `go test ... -tags=compatibility -timeout=10m ./pkg/kubemq/...` | Watermill `pubsub/tests.TestPubSub` compatibility suite. |
| `lint`               | `golangci-lint run ./...`                                       | Lint.                                                    |
| `coverage`           | `go test ... -coverprofile=coverage.out ./pkg/kubemq/...`       | Coverage profile and total summary.                      |

```bash
make test
make test-race
docker-compose up -d
make test-integration
make test-compatibility
```

## Related [#related]

<Cards>
  <Card title="Configuration Reference" href="/integrations/watermill/reference/configuration" description="Every config field, enum, validation rule, and the marshaling contract." />

  <Card title="Getting Started" href="/integrations/watermill/tutorials/getting-started" description="Install the plugin, start a broker, and run a first end-to-end Events round-trip." />

  <Card title="Commands & Queries" href="/integrations/watermill/how-to/commands-queries" description="Using the native CQPublisher for low-latency request-reply." />
</Cards>