Connection & 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:

docker run -d \  --name kubemq \  -p 50000:50000 \  -p 8080:8080 \  -p 9090:9090 \  -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY \  europe-docker.pkg.dev/kubemq/images/kubemq:next

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

• github.com/kubemq-io/watermill-kubemq installed (see Getting Started with Watermill)
• The broker above running and reachable

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:

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.

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.

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:

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:

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,
})

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:

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

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,
})

// 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,
})

// 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,
})

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.

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

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

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:

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:

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.

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

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

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

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.

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

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

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:

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:

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

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

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
}

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:

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

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:

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

Related

• Getting Started — install the plugin and run a first Events round-trip