Go SDK

The native Go SDK (client/client.go) abstracts HTTP calls, handles advanced routing (Priorities, Consumer Groups), and provides a highly resilient WebSocket client with automatic exponential backoff.

---

Installation

go get github.com/x-name15/tinymq/client

Your go.mod will have zero indirect dependencies — TinyMQ's client uses only the Go standard library.

---

HTTP Client: Publishing & Administration

The standard HTTP client uses modern Go idioms, requiring a context.Context for safe cancellation and timeouts. You can use PublishOptions for advanced message routing.

package main

import (
    "context"
    "log"
    "time"

    "github.com/x-name15/tinymq/client"
)

func main() {
    // Initialize client with a default 60s safe timeout
    mq := client.NewClient("http://127.0.0.1:7800", "optional_api_key")
    ctx := context.Background()

    // 1. Topic & Group Administration (Optional)
    // retain=0 creates the topic with no auto-expiry
    mq.CreateTopic(ctx, "orders", "reject", 0)
    // retain=24h auto-expires all messages after 24 hours
    mq.CreateTopic(ctx, "sensor.data", "drop-oldest", 24*time.Hour)
    mq.CreateGroup(ctx, "orders", "billing-service")

    // 2. Standard Publish
    payload := []byte(`{"event": "user_signup", "id": 99}`)
    if err := mq.Publish(ctx, "users.new", payload, nil); err != nil {
        log.Fatalf("publish failed: %v", err)
    }

    // 3. Advanced Publish (Priority, Idempotency, Broadcast & Headers)
    opts := &client.PublishOptions{
        Priority:    "high",
        Broadcast:   true, // Fan-out to all queues
        Idempotency: "txn_987654321",
        Headers: map[string]string{
            "X-Source": "api-gateway",
        },
    }

    if err := mq.Publish(ctx, "users.premium", payload, opts); err != nil {
        log.Fatalf("advanced publish failed: %v", err)
    }
}

PublishOptions Reference

Field	Type	Description
Priority	string	Message priority: "high", "normal", or "low"
Broadcast	bool	Fan-out to all waiting consumers simultaneously (ephemeral)
Idempotency	string	Custom idempotency key to prevent duplicate processing
TTL	time.Duration	Time-To-Live — message expires if not consumed in time
Delay	time.Duration	Delivery delay — message is hidden until duration passes
Headers	map[string]string	Custom X-MQ-* headers stored with the message

CreateTopic — Signature Change (Breaking)

CreateTopic now maps directly to POST /api/topics. The maxQueueSize int parameter has been replaced with retain time.Duration:

// Before (no longer compiles)
mq.CreateTopic(ctx, "orders", "durable", 10000)

// After
mq.CreateTopic(ctx, "orders", "reject", 0)             // no auto-expiry
mq.CreateTopic(ctx, "sensor.data", "drop-oldest", 24*time.Hour) // 24h retention

Passing retain=0 omits the retention field from the request body, leaving the topic with no default TTL.

---

High-Resilience HTTP Polling (Consumer Groups)

For standard HTTP consumers, Subscribe acts as a synchronous long-polling fetcher. It natively supports Consumer Groups for safe, distributed load balancing among multiple workers.

package main

import (
    "context"
    "fmt"
    "log"

    "github.com/x-name15/tinymq/client"
)

func main() {
    mq := client.NewClient("http://127.0.0.1:7800")
    ctx := context.Background()

    // Configure Long-Polling with Consumer Groups
    opts := &client.SubscriptionOptions{
        Timeout: "30s",
        Group:   "billing-service", // Messages are load-balanced across workers
    }

    log.Println("Worker started, polling for orders...")

    // Standard consumer loop
    for {
        msgs, err := mq.Subscribe(ctx, "orders", opts)
        if err != nil {
            log.Printf("Network or broker error: %v (retrying...)", err)
            continue
        }

        for _, msg := range msgs {
            fmt.Printf("Processing order ID: %s | Payload: %s\n", msg.ID, string(msg.Payload))
            // Implement your business logic / DLQ handling here
        }
    }
}

Re-queue & DLQ Integration

When your processing logic fails, call mq.Requeue(ctx, msg). After 3 total failures, the broker automatically isolates the message in {topic}.dlq for later inspection.

---

Real-Time WebSocket Client (Auto-Reconnecting)

For sub-millisecond latency without HTTP overhead, use the native WebSocket client. This is ideal for high-throughput, long-lived connections.

The WSClient features a Thread-Safe Single Read-Loop and an Automated Reconnection Pipeline. If the server drops or the network partitions, the SDK automatically cycles sockets and re-subscribes to all active topics using an exponential backoff strategy (1s up to 32s).

package main

import (
    "fmt"
    "log"

    "github.com/x-name15/tinymq/client"
    "github.com/x-name15/tinymq/internal/message"
)

func main() {
    // NewWSClient automatically dials and boots the auto-reconnect & keepalive loops
    ws, err := client.NewWSClient("127.0.0.1:7800")
    if err != nil {
        log.Fatalf("Initial connection failed: %v", err)
    }
    defer ws.Close() // Safely tears down resources and goroutines

    // 1. Subscribe asynchronously (Thread-Safe)
    err = ws.Subscribe("iot.sensors.*", func(msg message.Message) {
        // Handlers run synchronously per connection by default.
        // For massive concurrency, dispatch to your own worker pool here.
        fmt.Printf("Instant Push -> Topic: %s | Payload: %s\n", msg.Topic, string(msg.Payload))
    })
    if err != nil {
        log.Fatalf("Subscription failed: %v", err)
    }

    // 2. Dynamic Unsubscribe (Optional)
    // ws.Unsubscribe("iot.sensors.*")

    select {} // Block forever while WS handles traffic in the background
}

WSClient Architecture

Component	Description
Thread-Safe Read-Loop	A single goroutine owns all socket reads, eliminating race conditions
Auto-Reconnect Pipeline	On disconnect, cycles the socket and re-subscribes to all active topics
Exponential Backoff	Retry delay starts at 1s and doubles up to 32s
Keepalive	Automatically responds to server Ping frames to keep the connection alive
defer ws.Close()	Safely tears down all goroutines and releases resources

---

API Reference

HTTP Client

• NewClient(baseURL string, apiKey ...string) *Client
• Publish(ctx context.Context, topic string, payload []byte, opts *PublishOptions) error
• Subscribe(ctx context.Context, topic string, opts *SubscriptionOptions) ([]message.Message, error)
• CreateTopic(ctx context.Context, name, policy string, retain time.Duration) error ⚠️ Breaking change from previous versions — maxQueueSize int replaced by retain time.Duration
• CreateGroup(ctx context.Context, topic, group string) error
• Peek(ctx context.Context, topic string, limit int) ([]message.Message, error)

WSClient

• NewWSClient(addr string) (*WSClient, error)
• Subscribe(topic string, handler func(message.Message)) error
• Unsubscribe(topic string)
• Close()