# Connection & Configuration (/integrations/watermill/how-to/configuration)



Every `watermill-kubemq` Publisher, Subscriber, and CQPublisher is created from a config struct (`PublisherConfig`, `SubscriberConfig`, or `CQConfig`). The connection-related fields are shared across all three, so once you understand how to point one component at a broker, the rest follow the same pattern. This page covers each connection and behavior knob, with runnable code drawn from the repository's `examples/` directory.

The plugin is a native gRPC client — it dials the broker's gRPC port directly. Start a local broker before running any of the snippets below:

<RunKubeMQ ports="[50000, 8080, 9090]" />

Port `50000` is the gRPC API used by this plugin; `8080` is the REST API and health endpoint; `9090` exposes Prometheus metrics. Unlike HTTP-based connectors, the Watermill plugin needs no enable flag — as long as `50000` is reachable, it can publish and subscribe.

## Prerequisites [#prerequisites]

* `github.com/kubemq-io/watermill-kubemq` installed (see [Getting Started with Watermill](/integrations/watermill/tutorials/getting-started))
* The broker above running and reachable

## Address and ClientID [#address-and-clientid]

`Address` is the KubeMQ server in `host:port` form. It is **required unless `ExistingClient` is set** — `Validate()` returns an error when both are empty:

```go title="config.go"
func (c *PublisherConfig) Validate() error {
    if c.ExistingClient == nil && c.Address == "" {
        return fmt.Errorf("watermill-kubemq: Address is required when ExistingClient is nil")
    }
    // ...
}
```

`ClientID` uniquely identifies the connection to the broker. If you leave it empty, the underlying `kubemq-go` client generates a random UUID — so you never have to set it, but a stable, descriptive ID makes broker-side diagnostics easier to read.

```go title="address_clientid.go"
pub, err := kubemq.NewPublisher(kubemq.PublisherConfig{
    Address:  "localhost:50000",   // required (host:port)
    ClientID: "order-publisher",    // optional; auto-UUID when empty
    Pattern:  kubemq.PatternEvents,
})
```

These two fields, along with `AuthToken` and `TLS`, appear on `PublisherConfig`, `SubscriberConfig`, and `CQConfig` with identical semantics.

<TypeTable
  type="{
  Address: {
    type: 'string',
    description: 'KubeMQ server host:port. Required unless ExistingClient is set.',
  },
  ClientID: {
    type: 'string',
    description: 'Unique identifier for this connection. A random UUID is generated when empty.',
    default: 'auto-generated UUID',
  },
}"
/>

## Auth Token [#auth-token]

If your broker enforces authentication, set `AuthToken` to a JWT or token string. The plugin passes it to `kubemq-go` via `WithAuthToken` when it builds the client:

```go title="publisher.go"
if config.AuthToken != "" {
    opts = append(opts, kubemqSDK.WithAuthToken(config.AuthToken))
}
```

Both ends of a conversation must present a valid token when the broker requires auth — configure the publisher and the subscriber the same way:

```go title="auth_token.go"
pub, err := kubemq.NewPublisher(kubemq.PublisherConfig{
    Address:   "localhost:50000",
    AuthToken: "your-jwt-token-here", // maps to WithAuthToken internally
    Pattern:   kubemq.PatternEvents,
    Logger:    logger,
})

sub, err := kubemq.NewSubscriber(kubemq.SubscriberConfig{
    Address:   "localhost:50000",
    AuthToken: "your-jwt-token-here",
    Pattern:   kubemq.PatternEvents,
    Logger:    logger,
})
```

<Callout type="info">
  If your broker does not require authentication, omit `AuthToken` entirely. The full program is in `examples/connection/auth-token`.
</Callout>

## TLS [#tls]

For a TLS-enabled broker, supply a `*TLSConfig`. It has four fields, and the plugin maps them onto distinct `kubemq-go` options in `buildTLSOptions`:

| `TLSConfig` field      | Maps to                                           | Purpose                                   |
| ---------------------- | ------------------------------------------------- | ----------------------------------------- |
| `CertFile`             | `WithCredentials(certFile, serverOverrideDomain)` | Path to a PEM CA cert file                |
| `CertData`             | `WithCertificate(certData, serverOverrideDomain)` | Inline PEM cert string                    |
| `ServerOverrideDomain` | (passed alongside cert)                           | Expected server name in the TLS handshake |
| `InsecureSkipVerify`   | `WithInsecureSkipVerify()`                        | Disables certificate verification         |

```go title="publisher.go — buildTLSOptions"
func buildTLSOptions(tls *TLSConfig) []kubemqSDK.Option {
    var opts []kubemqSDK.Option
    if tls.CertFile != "" {
        opts = append(opts, kubemqSDK.WithCredentials(tls.CertFile, tls.ServerOverrideDomain))
    } else if tls.CertData != "" {
        opts = append(opts, kubemqSDK.WithCertificate(tls.CertData, tls.ServerOverrideDomain))
    }
    if tls.InsecureSkipVerify {
        opts = append(opts, kubemqSDK.WithInsecureSkipVerify())
    }
    return opts
}
```

`CertFile` takes precedence over `CertData` — if both are set, the file path wins. Use `ServerOverrideDomain` when the broker certificate's CN or SAN does not match the connection address (for example, connecting to `localhost` against a cert issued for `kubemq.example.com`).

<Tabs items="['Cert File', 'Cert Data', 'Insecure Skip Verify']">
  <Tab value="Cert File">
    ```go title="tls_cert_file.go"
    tlsConfig := &kubemq.TLSConfig{
        CertFile:             "/path/to/cert.pem",
        ServerOverrideDomain: "kubemq.example.com",
    }

    pub, err := kubemq.NewPublisher(kubemq.PublisherConfig{
        Address: "localhost:50000",
        Pattern: kubemq.PatternEvents,
        TLS:     tlsConfig,
        Logger:  logger,
    })
    ```
  </Tab>

  <Tab value="Cert Data">
    ```go title="tls_cert_data.go"
    // Load PEM data from an env var or Kubernetes secret in production:
    //   certData := os.Getenv("KUBEMQ_TLS_CERT")
    tlsConfig := &kubemq.TLSConfig{
        CertData:             "-----BEGIN CERTIFICATE-----\n...(your PEM data)...\n-----END CERTIFICATE-----",
        ServerOverrideDomain: "kubemq.example.com",
    }

    pub, err := kubemq.NewPublisher(kubemq.PublisherConfig{
        Address: "localhost:50000",
        Pattern: kubemq.PatternEvents,
        TLS:     tlsConfig,
        Logger:  logger,
    })
    ```
  </Tab>

  <Tab value="Insecure Skip Verify">
    ```go title="tls_insecure.go"
    // WARNING: disables all certificate verification.
    // Use ONLY for local development with self-signed certs.
    tlsConfig := &kubemq.TLSConfig{
        InsecureSkipVerify: true,
    }

    pub, err := kubemq.NewPublisher(kubemq.PublisherConfig{
        Address: "localhost:50000",
        Pattern: kubemq.PatternEvents,
        TLS:     tlsConfig,
        Logger:  logger,
    })
    ```
  </Tab>
</Tabs>

<Callout type="warn">
  Never use `InsecureSkipVerify` in production. It disables certificate verification entirely, leaving the connection open to man-in-the-middle attacks. It exists only for local development and testing against self-signed certificates. Runnable versions of all three modes are in `examples/tls/cert-file`, `examples/tls/cert-data`, and `examples/tls/insecure-skip-verify`.
</Callout>

## Reusing an Existing Client [#reusing-an-existing-client]

In a service that runs several publishers and subscribers, opening a separate broker connection for each is wasteful. Set `ExistingClient` to a `*kubemq.Client` you created yourself, and the plugin reuses that one connection.

When `ExistingClient` is set, the plugin ignores `Address`, `ClientID`, `AuthToken`, and `TLS` — the shared client's own configuration is used instead. The plugin also records that it does **not** own the connection, so calling `Close()` on the Publisher or Subscriber leaves the shared client open. **You own the client's lifecycle and must close it yourself** after all components that share it are closed.

```go title="config.go"
// ExistingClient allows reusing an existing kubemq-go Client.
// When set, Address, ClientID, AuthToken, and TLS are ignored.
// The caller is responsible for closing the Client.
ExistingClient *kubemqSDK.Client
```

```go title="existing_client.go"
// Create one shared client with all connection parameters.
client, err := kubemqSDK.NewClient(ctx,
    kubemqSDK.WithAddress("localhost", 50000),
    kubemqSDK.WithClientId("shared-client"),
)
if err != nil {
    log.Fatal(err)
}

// Reuse it across a publisher and a subscriber.
pub, err := kubemq.NewPublisher(kubemq.PublisherConfig{
    ExistingClient: client,
    Pattern:        kubemq.PatternEvents,
})
sub, err := kubemq.NewSubscriber(kubemq.SubscriberConfig{
    ExistingClient: client,
    Pattern:        kubemq.PatternEvents,
})

// Cleanup order matters: close the components first (they do NOT
// close the shared client), then close the client explicitly.
pub.Close()
sub.Close()
if err := client.Close(); err != nil {
    log.Printf("Error closing shared client: %v", err)
}
```

The full program — including a CQPublisher sharing the same client — is in `examples/connection/existing-client`.

## Custom Marshaler [#custom-marshaler]

The marshaler converts between a Watermill `*message.Message` and the intermediate `MarshaledMessage`/`ReceivedMessage` types the plugin hands to KubeMQ. The default, `DefaultMarshaler`, stores the Watermill payload in the KubeMQ message `Body` and the Watermill metadata in KubeMQ `Tags`. To change that mapping — compress payloads, encrypt them, switch serialization formats — implement the `Marshaler` / `Unmarshaler` interfaces:

```go title="marshaler.go"
type Marshaler interface {
    Marshal(topic string, msg *message.Message) (*MarshaledMessage, error)
}

type Unmarshaler interface {
    Unmarshal(msg *ReceivedMessage) (*message.Message, error)
}

// MarshalerUnmarshaler combines both — implement this to use one type
// for both directions (and for CQConfig.Marshaler).
type MarshalerUnmarshaler interface {
    Marshaler
    Unmarshaler
}
```

Set your implementation on `PublisherConfig.Marshaler` and `SubscriberConfig.Unmarshaler` (or `CQConfig.Marshaler`, which expects the combined `MarshalerUnmarshaler`). The repository ships a `CompressedJSONMarshaler` that gzip-compresses the payload and records the encoding in a custom tag, so a subscriber knows to decompress:

```go title="compressed_marshaler.go"
type CompressedJSONMarshaler struct{}

func (m CompressedJSONMarshaler) Marshal(topic string, msg *message.Message) (*kubemq.MarshaledMessage, error) {
    if msg == nil {
        return nil, fmt.Errorf("compressed-marshaler: message is nil")
    }

    // Gzip-compress the payload.
    var buf bytes.Buffer
    gz := gzip.NewWriter(&buf)
    if _, err := gz.Write(msg.Payload); err != nil {
        return nil, fmt.Errorf("compressed-marshaler: gzip write error: %w", err)
    }
    if err := gz.Close(); err != nil {
        return nil, fmt.Errorf("compressed-marshaler: gzip close error: %w", err)
    }

    // Preserve the Watermill UUID and record the encoding in tags.
    tags := make(map[string]string, len(msg.Metadata)+2)
    tags["_watermill_uuid"] = msg.UUID
    tags["_content_encoding"] = "gzip"

    for k, v := range msg.Metadata {
        if k == "_watermill_uuid" || k == "_content_encoding" {
            return nil, fmt.Errorf("compressed-marshaler: metadata key %q is reserved", k)
        }
        tags[k] = v
    }

    return &kubemq.MarshaledMessage{Body: buf.Bytes(), Tags: tags}, nil
}
```

On the receive side, `Unmarshal` reads the `_content_encoding` tag, decompresses when it is `"gzip"`, restores the UUID, and rebuilds the metadata map — excluding the two internal keys. The result is a full metadata round-trip: a message published with `source` and `version` metadata arrives with exactly those keys intact.

```go title="wire_custom_marshaler.go"
customMarshaler := CompressedJSONMarshaler{}

pub, _ := kubemq.NewPublisher(kubemq.PublisherConfig{
    Address:   "localhost:50000",
    Pattern:   kubemq.PatternEvents,
    Marshaler: customMarshaler, // outbound
})

sub, _ := kubemq.NewSubscriber(kubemq.SubscriberConfig{
    Address:     "localhost:50000",
    Pattern:     kubemq.PatternEvents,
    Unmarshaler: customMarshaler, // inbound
})
```

See `examples/advanced/custom-marshaler` for the complete compress/decompress round-trip, and `examples/advanced/metadata-roundtrip` for a focused look at metadata preservation.

### Reserved Metadata Key [#reserved-metadata-key]

The plugin uses one reserved tag, `_watermill_uuid`, to carry the Watermill message UUID across the broker. &#x2A;*Do not use this key in your message metadata.** `DefaultMarshaler` rejects any message whose metadata contains it:

```go title="marshaler.go"
const WatermillUUIDTag = "_watermill_uuid"

// inside DefaultMarshaler.Marshal:
for k, v := range msg.Metadata {
    if k == WatermillUUIDTag {
        return nil, fmt.Errorf("watermill-kubemq: metadata key %q is reserved for Watermill UUID", WatermillUUIDTag)
    }
    tags[k] = v
}
```

A custom marshaler can reserve additional keys of its own — the `CompressedJSONMarshaler` above also reserves `_content_encoding` and rejects metadata that collides with it.

## Disable Streaming [#disable-streaming]

By default the plugin opens a long-lived streaming gRPC handle for publishing (`eventStream.Send`, `eventStoreStream.Send`, `queueUpstream.Send`). Streaming maximizes throughput, but send results come back asynchronously on a background channel. Set `DisableStreaming: true` to use a synchronous unary call per publish instead (`client.SendEvent`, `client.SendEventStore`, `client.SendQueueMessages`), trading throughput for per-message, in-line send-result handling.

```go title="config.go"
// DisableStreaming disables streaming APIs for publishing.
// Default: false (streaming enabled)
DisableStreaming bool
```

Reach for non-streaming when you need per-message error feedback synchronously, when the publisher is short-lived and won't benefit from stream reuse, or when you simply want the simpler error path:

```go title="non_streaming.go"
eventPub, err := kubemq.NewPublisher(kubemq.PublisherConfig{
    Address:          "localhost:50000",
    Pattern:          kubemq.PatternEvents,
    DisableStreaming: true, // uses client.SendEvent() instead of eventStream.Send()
    Logger:           logger,
})
```

The setting applies to all three patterns. For EventsStore and Queues, the non-streaming path also surfaces a per-message result you can inspect immediately — for example, an EventsStore send returns a result whose `Sent` flag the plugin checks before reporting success. The full three-pattern walkthrough is in `examples/advanced/non-streaming-publish`.

## Logger [#logger]

Pass a Watermill `LoggerAdapter` to capture the plugin's internal log lines — publisher/subscriber creation, per-message trace events, and asynchronous stream errors. When `Logger` is nil, `Validate()` substitutes `watermill.NopLogger{}`, so logging is silent by default:

```go title="config.go"
if c.Logger == nil {
    c.Logger = watermillPkg.NopLogger{}
}
```

The quickest option is the built-in standard logger; its two booleans toggle debug and trace output:

```go title="logger.go"
// debug=true, trace=true — verbose output for troubleshooting.
logger := watermill.NewStdLogger(true, true)

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

To route logs into an external library (zap, zerolog, logrus), implement `watermill.LoggerAdapter` yourself and pass your adapter as `Logger`. `examples/observability/watermill-logging` shows both the standard logger and a custom adapter.

## Health Checks [#health-checks]

Both `Publisher` and `Subscriber` expose `HealthCheck(ctx)`, which pings the broker and returns an error if it is unreachable. Internally each calls `client.Ping(ctx)`:

```go title="publisher.go / subscriber.go"
func (p *Publisher) HealthCheck(ctx context.Context) error {
    _, err := p.client.Ping(ctx)
    if err != nil {
        return fmt.Errorf("watermill-kubemq: health check failed: %w", err)
    }
    return nil
}
```

```go title="health_check.go"
if err := pub.HealthCheck(ctx); err != nil {
    log.Printf("Publisher unhealthy: %v", err)
}
if err := sub.HealthCheck(ctx); err != nil {
    log.Printf("Subscriber unhealthy: %v", err)
}
```

A natural place to call these is a readiness endpoint. The following handler returns `200 OK` only when both the publisher and subscriber can reach the broker, and `503 Service Unavailable` otherwise — exactly the shape a Kubernetes readiness probe expects:

```go title="readiness_endpoint.go"
http.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
    checkCtx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
    defer cancel()

    if err := pub.HealthCheck(checkCtx); err != nil {
        w.WriteHeader(http.StatusServiceUnavailable)
        fmt.Fprintf(w, "publisher unhealthy: %v\n", err)
        return
    }
    if err := sub.HealthCheck(checkCtx); err != nil {
        w.WriteHeader(http.StatusServiceUnavailable)
        fmt.Fprintf(w, "subscriber unhealthy: %v\n", err)
        return
    }

    w.WriteHeader(http.StatusOK)
    fmt.Fprintln(w, "ok")
})
```

See `examples/connection/health-check` for the bare API and `examples/observability/health-check-endpoint` for the full HTTP wiring.

## Graceful Shutdown [#graceful-shutdown]

Always `defer Close()` on publishers and subscribers. The two close paths differ:

* **`Publisher.Close()`** closes whichever streaming handle is open and, if the publisher owns the connection, closes the client. It is idempotent — a second call is a no-op.
* **`Subscriber.Close()`** cancels every active subscription, then **waits for all in-flight subscription goroutines to finish** before returning. For the Queues pattern, when a subscription's context is cancelled the bridge goroutine **Nacks any unsettled message** so it returns to the queue for redelivery rather than being lost.

In a router-based service, close components in order: router first (stops accepting new messages), then the subscriber (drains in-flight handlers and Nacks unsettled queue messages), then the publisher:

```go title="graceful_shutdown.go"
sig := <-sigChan // SIGINT or SIGTERM
fmt.Printf("Received signal: %v\n", sig)

// 1. Close the router — stops accepting new messages.
if err := router.Close(); err != nil {
    log.Printf("Router close error: %v", err)
}

// 2. Close the subscriber — cancels subscriptions, waits for goroutines,
//    Nacks unsettled queue messages.
if err := sub.Close(); err != nil {
    log.Printf("Subscriber close error: %v", err)
}

// 3. Close the publisher — closes streaming handles, releases the connection.
if err := pub.Close(); err != nil {
    log.Printf("Publisher close error: %v", err)
}
```

<Callout type="info">
  If you share a connection via `ExistingClient`, the components' `Close()` calls leave it open — close the shared client yourself as the final step. Complete shutdown examples are in `examples/error-handling/graceful-shutdown` and `examples/router/graceful-shutdown`.
</Callout>

## Related [#related]

* [Getting Started](/integrations/watermill/tutorials/getting-started) — install the plugin and run a first Events round-trip