Go client

Import path github.com/smartpricing/queen/clients/client-go, package name queen. Requires Go 1.24 or newer. Every network method takes a context.Context as its first argument. Version 1.0.0, aligned with the broker.

import queen "github.com/smartpricing/queen/clients/client-go"

The streaming SDK is a separate package under .../client-go/streams.

Constructor

func New(config interface{}) (*Queen, error)

config accepts a string (single URL), a []string (multiple URLs), a ClientConfig, or a *ClientConfig. Anything else returns invalid config type: expected string, []string, or ClientConfig.

ClientConfig

The exported ClientDefaults, QueueDefaults, ConsumeDefaults, PopDefaults and BufferDefaults variables hold these values, so an application can read one instead of restating it. Unlike the other clients, the load balancer is constructed unconditionally, even for a single URL, so LoadBalancingStrategy and the health settings are always in play.

Retry429Config

Separate from RetryAttempts, which covers 5xx and network faults. A zero field means “use the kind-based default”.

A Retry-After header (seconds) wins over the computed delay, with ±20% jitter either way. A 429 is retried against the same backend and never triggers failover.

*Queen

Ack

messages accepts *Message, Message, []*Message or []Message; anything else returns an error. success maps to AckStatusCompleted ("completed") or AckStatusFailed ("failed"). AckOptions carries ConsumerGroup and Error.

One message uses POST /api/v1/ack, more than one uses POST /api/v1/ack/batch. Both return one AckResponse{Success, Error} per message in request order. ValidateMessages runs first, so a message missing TransactionID or PartitionID fails before any request is sent.

Renew

Accepts a lease-id string, []string, or any of the message forms. Lease ids are deduplicated first: with multi-partition pop every message in a batch shares one LeaseID and one extend call renews every claimed partition, so passing the whole slice issues one HTTP call. RenewResponse carries LeaseID, Success, NewExpiresAt time.Time and Error.

*QueueBuilder

client.Queue(name). Configuration methods return *QueueBuilder; terminal methods are Push, Pop, Consume, ConsumeBatch, Create, Delete, DLQ and FlushBuffer.

Addressing

Queue lifecycle

*OperationBuilder has one method: Execute(ctx) (map[string]interface{}, error). Create validates the queue name first and adds "configured": true to the returned map for parity with the other clients.

QueueConfig holds the fields Create can send, with the value in QueueDefaults:

Producing

*PushBuilder:

PushResponse carries Status ("queued", "duplicate", "failed"), TransactionID and Error. The client mints a UUIDv7 transactionId for items that have none.

responses, err := client.Queue(queueName).Push(payload).Execute(ctx)
if err != nil {
	t.Fatalf("Failed to push message: %v", err)
}

Consuming

Pop

func (qb *QueueBuilder) Pop(ctx context.Context) ([]*Message, error)

Returns queue name, namespace, or task is required when none of the three is set. Route selection: queue plus non-Default partition uses /api/v1/pop/queue/:queue/partition/:partition; queue alone uses /api/v1/pop/queue/:queue; namespace or task alone uses /api/v1/pop.

Consume and ConsumeBatch

func (qb *QueueBuilder) Consume(ctx, handler MessageHandler) *ConsumeBuilder
func (qb *QueueBuilder) ConsumeBatch(ctx, handler BatchMessageHandler) *ConsumeBuilder

MessageHandler is func(ctx context.Context, msg *Message) error; BatchMessageHandler is func(ctx context.Context, msgs []*Message) error. *ConsumeBuilder exposes Execute(ctx) error and Start(ctx) error (Start is an alias). Both block until every worker stops; cancel the context to stop them.

Worker loop behaviour:

• A handler error with AutoAck on nacks the message and the worker continues.
• A 429 that escapes the HTTP layer’s own retry backs off and continues.
• A 403 stops the worker. Use (*HTTPError).IsClusterSuspended() to distinguish the terminal case.

(*Message).Trace

func (m *Message) Trace(ctx context.Context, config TraceConfig) (*TraceResponse, error)

TraceConfig carries TraceName string, TraceNames []string, EventType string and Data map[string]interface{}. Posts to /api/v1/traces.

Buffering

FlushBuffer(ctx) error flushes this builder’s queue/partition buffer. *BufferManager exposes Add, Flush(ctx, key), FlushAll(ctx), GetStats() BufferStats, GetBuffer(key) and Clear(). BufferStats carries ActiveBuffers, TotalBufferedMessages, OldestBufferAge and FlushesPerformed.

Dead-letter queue

DLQ(consumerGroup string) *DLQBuilder.

Read-only. Admin.RetryMessage exists but returns an error without making a request, so replay from Go means reading the row and pushing the payload again, or calling POST /api/v1/messages/{partitionId}/{transactionId}/retry directly.

*TransactionBuilder

client.Transaction(). Pushes and acks in one PostgreSQL transaction, all-or-nothing.

resp, err := client.Transaction().
    Ack(msg, queen.AckStatusCompleted, queen.AckOptions{ConsumerGroup: "workers"}).
    Queue("orders.enriched").Push(payload).
    Commit(ctx)

*Admin

client.Admin(). Every method takes a context and returns map[string]interface{}, the broker’s JSON body verbatim, except where noted.

Resources

ListQueuesParams carries Namespace, Task, Limit, Offset. GetPartitions requires a queue name and returns the queue-detail payload: the broker bundles partition data there rather than exposing a partitions collection.

Messages, DLQ and traces

ListMessagesParams carries Queue, Partition, Status, ConsumerGroup, Limit, Offset. ListDLQParams carries Queue, ConsumerGroup, Partition, From, To, Limit, Offset.

Status and analytics

GetStatusParams carries Queue, Namespace, Task.

Consumer groups

SeekConsumerGroupOptions carries Timestamp string (RFC3339) and ToEnd bool, which are mutually exclusive. The wire body is {"toEnd": true} or {"timestamp": "…"}.

System

Errors

Non-2xx responses come back as *HTTPError:

var he *queen.HTTPError
if errors.As(err, &he) && he.StatusCode == 429 { /* ... */ }

IsClusterSuspended() reports the terminal 403 that no amount of retrying resolves. Error() renders as HTTP <status> [<code>]: <body>.

A 204 response yields a nil map, because the broker sends no body at all on 204.

*HttpClient

Reachable via GetHttpClient() when you need a route the SDK does not wrap.

All four return (map[string]interface{}, error); a top-level JSON array is wrapped as {"data": [...]}. WithLongPollRetry() is the only RequestOption: it marks the request as a long-poll pop so a 429 backs off indefinitely instead of using the bounded budget.

Logging

Set by the QUEEN_CLIENT_LOG environment variable, read once at package init:

SetLogLevel(level LogLevel) and GetLogLevel() LogLevel change it at runtime.

Helpers

Message

Streaming SDK

import "github.com/smartpricing/queen/clients/client-go/streams"

s := streams.From(queen.Queue("orders").AsStreamSource())

AsStreamSource() on a *QueueBuilder adapts it to runtime.Source. Every combinator returns a new *Stream.

Window options: WithGracePeriod, WithIdleFlushMs, WithEventTime, WithAllowedLateness, WithOnLate. streams/helpers provides TokenBucketGate(TokenBucketGateOptions) and SlidingWindowGate(SlidingWindowGateOptions), both returning a GateFn.

RunOptions

*Runner exposes Metrics() Metrics with the counters CyclesTotal, FlushCyclesTotal, MessagesTotal, PushItemsTotal, StateOpsTotal, LateEventsTotal, ErrorsTotal, GateAllowsTotal, GateDenialsTotal and LastError.

The chain shape is fingerprinted into a config hash; re-deploying a different chain under the same QueryID is rejected at registration unless Reset is set. Only operator kinds and their structural config are hashed, not the bodies of your functions.

Same surface, other languages