API Reference

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. Every entry is drawn from the plugin source.

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.

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

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.

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

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 is a separate API — it does not implement message.Publisher. It wraps KubeMQ's native Commands and Queries for low-latency request-reply.

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.

Make targets

The repository Makefile provides the following targets:

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

Related