Configuration Reference

This page is the authoritative configuration reference for the watermill-kubemq plugin: every config field with its type and default, the enums and validation rules, the marshaling contract, reserved metadata keys, and pinned dependency versions. The methods and constructors themselves are in the API reference. Every entry is drawn from the plugin source; where the repository README and the source disagree, the source wins and the discrepancy is called out.

Package and import

The plugin module is published at github.com/kubemq-io/watermill-kubemq. All public types live in the pkg/kubemq subpackage, conventionally imported under the alias kubemq:

import (
    "github.com/ThreeDotsLabs/watermill"
    "github.com/ThreeDotsLabs/watermill/message"
    kubemq "github.com/kubemq-io/watermill-kubemq/pkg/kubemq"
)

Install it with:

go get github.com/kubemq-io/watermill-kubemq

The plugin is a native gRPC client built on github.com/kubemq-io/kubemq-go/v2. It connects to the KubeMQ broker's gRPC port (50000) directly — there is no HTTP connector and no server-side enable flag to set, unlike the HTTP-based connectors. A broker reachable on 50000 is all that is required:

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 exposes the REST API and health endpoint, and 9090 exposes Prometheus metrics.

PublisherConfig

Configures NewPublisher. QueueMessagePolicy is only consulted when Pattern is PatternQueues.

SubscriberConfig

Configures NewSubscriber. The MaxItems and WaitTimeoutSeconds fields apply only to the Queues pattern; the EventsStore* fields apply only to the EventsStore pattern.

CQConfig

Configures NewCQPublisher.

QueueMessagePolicy

Default delivery policy for queue messages, set on PublisherConfig.QueueMessagePolicy and used only by the Queues pattern.

TLSConfig

TLS settings for the connection, set on the TLS field of any config. File-based credentials (CertFile) take precedence over PEM data (CertData).

Enums

PatternType

Selects the KubeMQ messaging pattern for a Publisher or Subscriber. Each instance serves exactly one pattern.

Any value outside this range fails validation, and String() returns unknown.

EventsStoreStartOption

Controls the starting position for an EventsStore subscription (SubscriberConfig.EventsStoreStartOption).

Validation rules

Each config exposes a Validate() method, invoked automatically by its constructor. Validation both rejects invalid input and applies defaults in place. The table below summarizes the enforced rules.

Marshaling

The plugin converts between Watermill messages and KubeMQ messages through small intermediate types and three interfaces.

DefaultMarshaler maps the Watermill payload to the KubeMQ message Body, and Watermill metadata to KubeMQ Tags. The Watermill message UUID is stored under a reserved tag key exported as the constant WatermillUUIDTag:

const WatermillUUIDTag = "_watermill_uuid"

On marshal, the UUID is written to this tag, and supplying _watermill_uuid as a metadata key yourself is rejected with an error (it is reserved). On unmarshal, the UUID is recovered from the tag; if absent, the KubeMQ message ID is used; if that is also empty, a fresh watermill.NewUUID() is generated. The _watermill_uuid tag is stripped from the reconstructed Watermill metadata.

To customize wire format, implement Marshaler/Unmarshaler (or MarshalerUnmarshaler for CQ) and set it on the corresponding config field — see the Connection & Configuration guide for a worked compressed-marshaler example.

Reserved and auto-populated metadata keys

Several metadata keys carry special meaning. _watermill_uuid is reserved (you cannot set it). The _kubemq_* keys are set by the subscriber on inbound messages from KubeMQ message attributes. The W3C Trace Context keys are injected on publish and extracted on receive by the OpenTelemetry propagator.

Dependency versions

Pinned core dependencies, from go.mod:

Related