KubeMQ Go SDK

The KubeMQ SDK for Go enables Go developers to communicate with KubeMQ message broker servers, supporting Events, Events Store, Queues, Commands, and Queries messaging patterns.

v1 users: v1.x receives security patches for 12 months after v2.0.0 GA. See COMPATIBILITY.md for details and MIGRATION.md for upgrade instructions.

Table of Contents

• Installation
• Quick Start
• Messaging Patterns
• Configuration
• Error Handling
• Troubleshooting
• Examples
• Migration from v1
• Contributing
• License

Installation

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

Requires Go 1.23 or later. Compatible with KubeMQ server v2.2 or later (see COMPATIBILITY.md).

Quick Start

Quick Start: Events

Prerequisites

• Go 1.23+
• KubeMQ server running on localhost:50000 (docker run -d -p 50000:50000 kubemq/kubemq)
• Install: go get github.com/kubemq-io/kubemq-go/v2

Send an Event

package main

import (
    "context"
    "log"

    "github.com/kubemq-io/kubemq-go/v2"
)

func main() {
    ctx := context.Background()
    client, err := kubemq.NewClient(ctx,
        kubemq.WithAddress("localhost", 50000),
    )
    if err != nil {
        log.Fatal(err)
    }
    defer client.Close()

    err = client.SendEvent(ctx, kubemq.NewEvent().
        SetChannel("notifications").
        SetBody([]byte("hello kubemq")),
    )
    if err != nil {
        log.Fatal(err)
    }
    log.Println("Event sent successfully")
}

Receive Events

package main

import (
    "context"
    "fmt"
    "log"

    "github.com/kubemq-io/kubemq-go/v2"
)

func main() {
    ctx := context.Background()
    client, err := kubemq.NewClient(ctx,
        kubemq.WithAddress("localhost", 50000),
    )
    if err != nil {
        log.Fatal(err)
    }
    defer client.Close()

    sub, err := client.SubscribeToEvents(ctx, "notifications", "",
        kubemq.WithOnEvent(func(event *kubemq.Event) {
            fmt.Printf("Received event: %s\n", event.Body)
        }),
        kubemq.WithOnError(func(err error) {
            log.Println("Subscription error:", err)
        }),
    )
    if err != nil {
        log.Fatal(err)
    }
    defer sub.Unsubscribe()

    <-ctx.Done()
}

Quick Start: Queues

// Send
msg := kubemq.NewQueueMessage().SetChannel("tasks").SetBody([]byte("hello"))
result, err := client.SendQueueMessage(ctx, msg)

// Receive
resp, err := client.ReceiveQueueMessages(ctx, &kubemq.ReceiveQueueMessagesRequest{
    Channel:             "tasks",
    MaxNumberOfMessages: 1,
    WaitTimeSeconds:     10,
})
// Process resp.Messages, then AckAllQueueMessages

Quick Start: Commands (RPC)

// Send command
cmdResp, err := client.SendCommand(ctx, kubemq.NewCommand().
    SetChannel("orders").
    SetBody([]byte("create order")).
    SetTimeout(10*time.Second))

// Handle commands (subscribe)
sub, err := client.SubscribeToCommands(ctx, "orders", "",
    kubemq.WithOnCommandReceive(func(cmd *kubemq.CommandReceive) {
        resp := kubemq.NewResponse().
            SetRequestId(cmd.Id).
            SetResponseTo(cmd.ResponseTo).
            SetBody([]byte("done")).
            SetExecutedAt(time.Now())
        _ = client.SendResponse(ctx, resp)
    }),
)

Messaging Patterns

Pattern	Delivery Guarantee	Use When	Example Use Case
Events	At-most-once	Fire-and-forget broadcasting to multiple subscribers	Real-time notifications, log streaming
Events Store	At-least-once (persistent)	Subscribers must not miss messages, even if offline	Audit trails, event sourcing, replay
Queues	At-least-once (with ack)	Work must be processed by exactly one consumer with acknowledgment	Job processing, task distribution
Commands	At-most-once (request/reply)	You need a response confirming an action was executed	Device control, configuration changes
Queries	At-most-once (request/reply)	You need to retrieve data from a responder	Data lookups, service-to-service reads

Configuration

Option	Default	Description
WithAddress(host, port)	localhost:50000	KubeMQ server address
WithClientId(id)	Auto-generated UUID	Unique client identifier
WithCredentialProvider(p)	None (insecure)	Authentication credential provider
WithTLS(caCertFile)	None (plaintext)	TLS with CA cert file; use WithTLSConfig(cfg) for full config
WithReconnectPolicy(p)	Infinite retries, 1s–30s backoff	Reconnection behavior on disconnect
WithConnectionTimeout(d)	10s	Maximum time to establish initial connection
WithRetryPolicy(p)	3 retries, 100ms–10s backoff	Retry policy for transient operation failures
WithLogger(l)	None (silent)	Structured logger for SDK diagnostics
WithTracerProvider(tp)	None (no tracing)	OpenTelemetry tracer provider
WithMeterProvider(mp)	None (no metrics)	OpenTelemetry meter provider

Error Handling

All SDK operations return errors as *kubemq.KubeMQError, which provides structured error information:

err := client.SendEvent(ctx, event)
if err != nil {
    var ke *kubemq.KubeMQError
    if errors.As(err, &ke) {
        fmt.Printf("Code: %s, Retryable: %v\n", ke.Code, ke.IsRetryable)

        switch ke.Code {
        case kubemq.ErrCodeTimeout:
            // Retry with longer timeout
        case kubemq.ErrCodeAuthentication:
            // Check credentials
        case kubemq.ErrCodeTransient:
            // Automatic retry was exhausted; consider manual retry
        }
    }
}

The SDK automatically retries transient errors using exponential backoff with jitter. Configure retry behavior with WithRetryPolicy:

client, _ := kubemq.NewClient(ctx,
    kubemq.WithAddress("localhost", 50000),
    kubemq.WithRetryPolicy(kubemq.RetryPolicy{
        MaxRetries:     5,
        InitialBackoff: 200 * time.Millisecond,
        MaxBackoff:     15 * time.Second,
        Multiplier:     2.0,
        JitterMode:     kubemq.JitterFull,
    }),
)

For the full error code reference, see the API documentation.

Troubleshooting

Problem	Quick Fix
Connection refused	Verify KubeMQ server is running: kubectl get pods -l app=kubemq
Authentication failed	Check token validity; see Troubleshooting Guide
Timeout / deadline exceeded	Increase timeout via context.WithTimeout or WithRetryPolicy
No messages received	Verify channel name matches publisher; check subscriber group
TLS handshake failure	Verify certificate paths and CA trust chain

See TROUBLESHOOTING.md for the complete guide with error messages and step-by-step solutions.

Examples

Runnable examples are in the examples directory:

• examples/pubsub — Events pub/sub
• examples/queues — Queue send/receive with ack
• examples/cq — Commands and Queries (RPC)

Run an example:

go run ./examples/pubsub/

Migration from v1

For users upgrading from v1, see the Migration Guide.

For a complete list of changes by version, see the Changelog.

Contributing

See CONTRIBUTING.md for development setup, coding standards, and pull request process.

License

Apache License 2.0 — see LICENSE for details.

---

Deprecation Policy

• Deprecated APIs are annotated with // Deprecated: Use X instead. (recognized by go vet).
• Deprecated APIs remain functional for at least 2 minor versions or 6 months, whichever is longer.
• Deprecation notices name the replacement API.
• Removed APIs are listed in CHANGELOG.md and MIGRATION.md.

Performance

Characteristics

Parameter	Value	Notes
Transport	gRPC (HTTP/2)	Single multiplexed connection
Max message size	100 MB (send and receive)	Configurable via WithMaxSendMessageSize / WithMaxReceiveMessageSize
Max concurrent streams	Unlimited (gRPC default)	Limited by server configuration
Connection setup	~50-200ms typical	Includes TLS handshake if secured
Keepalive interval	10s (configurable)	Dead connection detected within 15s

Performance Tips

1. Reuse the client instance. Create one kubemq.Client and share it across all goroutines.
2. Use batching for high-throughput queue sends. Aim for 10–100 messages per batch with SendQueueMessages.
3. Do not block subscription callbacks. Offload slow work to worker pools.
4. Close streams when done. Call Unsubscribe() when no longer needed.

See BENCHMARKS.md for reproducible benchmark results.