Client SDK

Overview

The SDK lives in github.com/Ali-Hasan-Khan/dsend/client. It exposes two clients that mirror the broker's two roles:

• Producer — publishes messages and manages queues.
• Consumer — subscribes to a queue, receives deliveries, and acknowledges them.

Each client holds a single persistent TCP connection to the broker and speaks the JSON wire protocol directly. No HTTP, no auto-reconnect, no magic — just synchronous calls and explicit context.Context handling.

Quickstart

1. Start the broker

Build and run the server. It binds 127.0.0.1:8080 and appends to ./data/wal.log.

$ go build -o dsend ./cmd/dsend
$ ./dsend server

2. Publish

package main

import (
	"context"
	"fmt"

	"github.com/Ali-Hasan-Khan/dsend/client"
)

func main() {
	p, err := client.NewProducer("localhost:8080")
	if err != nil {
		panic(err)
	}
	defer p.Close()

	if err := p.Publish(context.Background(), "orders", "order-1042 created"); err != nil {
		panic(err)
	}
	fmt.Println("published")
}

3. Consume and acknowledge

package main

import (
	"context"
	"fmt"

	"github.com/Ali-Hasan-Khan/dsend/client"
)

func main() {
	c, err := client.NewConsumer("localhost:8080")
	if err != nil {
		panic(err)
	}
	defer c.Close()

	if err := c.Subscribe("orders"); err != nil {
		panic(err)
	}

	for {
		msg, err := c.Receive(context.Background())
		if err != nil {
			panic(err)
		}

		// do real work here...
		fmt.Println("received:", msg.Payload)

		if err := c.Ack(msg.AckToken); err != nil {
			panic(err)
		}
	}
}

Producer

A producer dials the broker once and keeps the connection open. All methods take a context.Context so you can time out or cancel cleanly.

Construct

Opens a persistent TCP connection to the broker at addr.

Methods

Publishing

Payloads are strings. Keep the payload small and treat it as opaque — the broker never inspects the contents. A message is written to the WAL before it is queued, so a crash between the two steps doesn't lose it.

ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()

// Named queue. The queue is created implicitly on first publish.
err := p.Publish(ctx, "emails", `{"to":"ada@example.com","subject":"hi"}`)

// Default queue — no name required.
err = p.Publish(ctx, "", "hello, world")

Queue administration

if err := p.CreateQueue(ctx, "orders"); err != nil {
	panic(err)
}

names, err := p.ListQueues(ctx)
if err != nil {
	panic(err)
}
fmt.Println(names) // ["default", "orders"]

if err := p.DeleteQueue(ctx, "orders"); err != nil {
	panic(err)
}

Consumer

A consumer is push-based: subscribe once, then loop on Receive. The broker schedules deliveries round-robin across all consumers subscribed to a queue.

Construct & subscribe

Subscribe registers this connection as a consumer session for the queue. One consumer connection can subscribe to a single queue at a time.

Receiving

Blocks until the broker pushes the next delivery. A ReceivedMessage carries:

Acknowledging

Tell the broker the message was processed successfully. The broker removes the ack token from its in-flight registry and never redelivers that message.

for {
	msg, err := c.Receive(ctx)
	if err != nil {
		// transient read deadline or network error; try again
		if ctx.Err() != nil {
			return // caller cancelled
		}
		continue
	}

	if err := handle(msg.Payload); err != nil {
		// skip ack: the broker will redeliver later
		continue
	}

	if err := c.Ack(msg.AckToken); err != nil {
		return
	}
}

Unsubscribing

Removes this session from the queue's round-robin rotation and stops deliveries. Closing the connection also unregisters the session.

Managing queues

A queue is an isolated, fixed-capacity in-memory ring buffer with its own consumers, in-flight messages, DLQ, and metrics. The default queue exists implicitly; publishing to any other name creates the queue on first use.

• Implicit creation — publishing to orders creates it if it doesn't exist.
• Explicit creation — CreateQueue if you want to manage it up front.
• Capacity — 20 messages per queue by default (see configuration).
• Deletion — DeleteQueue removes the queue from the broker.

$ ./dsend queue create orders
Queue created successfully

$ ./dsend queue list
Queues: [default orders]

$ ./dsend queue delete orders
Queue deleted successfully

Metrics

Every counter maps to a concrete delivery event. Query broker-wide metrics or scope them to one queue. The CLI exposes the same data: ./dsend metrics or ./dsend metrics --queue orders.

Types

type Metric struct {
	ProducedCount        int
	AckedCount           int
	DlqCount             int
	InflightCount        int
	RedeliveredCount     int
	ConsumerSessionCount int
	QueueDepth           int
}

type QueueMetric struct {
	Name string
	Metric
}

type BrokerMetrics struct {
	Queues []QueueMetric
	Total  Metric
}

Fields

m, err := p.Metrics(ctx)
if err != nil {
	panic(err)
}

for _, q := range m.Queues {
	fmt.Printf("%s: depth=%d in-flight=%d acked=%d\n",
		q.Name, q.QueueDepth, q.InflightCount, q.AckedCount)
}

q, err := p.QueueMetrics(ctx, "orders")
if err != nil {
	panic(err)
}
fmt.Println(q.DlqCount)

CLI reference

The dsend binary wraps the SDK for scripting and quick tests.

Wire protocol

Clients and broker exchange length-delimited JSON over TCP. Each message is a request with a type field; the broker replies with a response carrying success and, on failure, error.

Request types

Example

{"type":"publish","queue":"orders","message":{"payload":"order-1042 created"}}

{"success":true,"error":""}

{
  "success": true,
  "message": {
    "id": "a1b2c3d4-...",
    "payload": "order-1042 created",
    "timestamp": "2026-07-31T12:00:00Z",
    "retry": 0
  },
  "ack_token": "9f8e7d6c-..."
}

Delivery semantics

Push with round-robin

The broker pushes deliveries to subscribed consumers in round-robin order. Add more consumers to a queue and load is spread across them automatically. Each consumer receives one message at a time.

At-least-once

When a message is delivered, the broker moves it from the ring buffer into an in-flight registry keyed by ack token. It stays there until one of three things happens:

1. Ack — the consumer processed it; it's removed permanently.
2. Ack timeout — if the consumer doesn't ack within AckTimeout, the message is requeued and redelivered.
3. Retry limit — after MaxRetries redeliveries, the message is moved to the queue's DLQ.

Default configuration

Configuration is set in engine.DefaultConfig() at broker startup (cmd/dsend/server.go).

Dead-letter queue

Each queue has its own DLQ. Messages that exhaust their retries land there instead of being dropped. The DlqCount metric tracks how many have.

Persistence & recovery

The broker appends every state change — publish, ack, requeue, dead-letter — to a write-ahead log. On restart it replays the WAL and rebuilds queues, in-flight state, and session ownership.