streamq

A lightweight message broker with in-memory or WAL-backed storage, real-time WebSocket and SSE delivery, Prometheus metrics, and a live dashboard. Built in Go.

Built as a learning project to deeply understand pub/sub internals, Go concurrency primitives, streaming protocols, and production observability patterns.

---

What it does

• Topics — create named channels that producers publish to and consumers subscribe from
• Producers — HTTP POST to publish a message; get back an offset and timestamp
• Consumers — subscribe via WebSocket or SSE; choose between broadcast or consumer group delivery
• Consumer groups — multiple consumers in a group split the message load (one message → one consumer); independent groups each get every message
• Replay — reconnecting consumers can resume from any retained offset
• Retention — bounded message retention through memory caps or WAL segment rotation
• Metrics — Prometheus-compatible /metrics endpoint with per-topic publish/delivery/drop counters, active subscriber gauges, consumer group lag, and Go runtime stats
• Dashboard — embedded single-file React dashboard served at /dashboard, with smooth custom canvas charts and no frontend build step

---

Architecture

                    ┌──────────────────────────────────────┐
                    │              Broker                   │
                    │                                       │
  POST /publish ───►│  topic.Publish(msg)                  │
                    │       │                               │
                    │       ├── append to Store            │
                    │       │  (MemoryLog or WALStore)      │
                    │       │                               │
                    │       └── fan-out to subscribers      │
                    │               │                       │
                    │     ┌─────────┴─────────┐            │
                    │     │                   │            │
                    │  ungrouped           grouped          │
                    │  (broadcast)      (round-robin        │
                    │  → ALL subs       → ONE sub           │
                    │                    per group)         │
                    └──────────────────────────────────────┘
                               │              │
                         WebSocket           SSE
                         delivery           delivery
                         goroutine          goroutine
                               │              │
                         metrics.Inc()   metrics.Inc()
                    (MessagesDeliveredTotal, ActiveSubscribers)

Package layout

Package	Responsibility
config	Load .env via godenv, validate at startup
broker	Topic registry, publish, fan-out, consumer group state, metrics instrumentation
store	Append-only memory and WAL-backed logs with offset replay and bounded retention
delivery	WebSocket and SSE long-lived connection handlers with delivery metrics
api	HTTP handlers, route registration, JSON envelope, embedded dashboard
metrics	Prometheus registry and all metric definitions
middleware	Request logger
types	Shared Message type (breaks broker↔store import cycle)

---

Key design decisions

Why channels for fan-out instead of callbacks?

Each subscriber is represented by a buffered channel (chan Message). When a message is published, the broker sends it into each subscriber's channel. The delivery goroutine (one per connection) is blocked in a select reading from that channel.

This means:

• The broker never waits for a slow network write — the channel buffer absorbs bursts
• Closing a channel is the clean signal to a delivery goroutine to stop — no extra signalling mechanism needed
• The broker and delivery layer share zero mutable state

Why SSE and WebSocket instead of just one?

• SSE is unidirectional (server → client), works through HTTP proxies, and has built-in browser reconnect via Last-Event-ID. Best for read-only consumers: dashboards, analytics, log tails.
• WebSocket is full-duplex. When you add explicit acknowledgements or client-side flow control, you need it. Supporting both now means consumers can choose the right protocol for their use case.

Why net/http stdlib instead of Gin?

Go 1.22 added method-based routing and path parameters to ServeMux, which covers everything streamq needs. Using the stdlib means one fewer dependency and a deeper understanding of how middleware chains and handler registration actually work — which every Go web framework builds on anyway.

Why base64-encoded payloads?

Message.Payload is []byte. When marshalled to JSON, Go automatically encodes []byte as base64. This means:

• Producers can send arbitrary binary data (Protobuf, images) through a JSON API
• The broker never inspects or transforms payload bytes

Backpressure: drop vs block

When a subscriber's channel buffer fills (slow consumer), the broker drops the message for that subscriber rather than blocking. This is an explicit tradeoff:

• Blocking would make one slow consumer degrade the entire fan-out — all other subscribers would stop receiving until the slow one drains.
• Dropping keeps the broker responsive. Slow subscribers are a consumer-side problem; the broker signals this via the streamq_messages_dropped_total metric.

Consumer group offsets: commit-only-forward

ConsumerGroupRegistry.Commit() silently ignores commits that would lower the stored offset. A consumer could send a stale offset due to a bug or race; ignoring it prevents accidentally re-delivering messages that were already processed.

Why a custom Prometheus registry?

Using prometheus.NewRegistry() instead of the default global registry means:

• Tests can create isolated registries — no metric state bleeds between test runs
• The /metrics endpoint only exposes what streamq explicitly registers
• No accidental inclusion of third-party library metrics

Consumer group lag computation

Lag is recomputed on every CommitOffset call: lag = latest_offset − committed_offset. The broker is the only place that has both values simultaneously — it owns the topic (latest offset) and the group registry (committed offset). This keeps the gauge accurate without a polling goroutine.

Embedded dashboard via go:embed

dashboard.html is baked into the binary at compile time using //go:embed. The final Docker image has no external static files — the dashboard ships with the exact version of the broker that serves it. No web server config, no volume mounts.

The dashboard keeps the runtime lightweight: React handles view state and polling, while the overview and consumer-lag charts are drawn directly on <canvas> with requestAnimationFrame animations. There is no bundled frontend app and no charting dependency to build or ship.

Dashboard tabs are hash-addressable for direct links and screenshot capture.

Dashboard tab screenshots

Overview

Topics

Lag

Live Feed

---

Getting started

Run locally (Go installed)

git clone https://github.com/GordenArcher/streamq
cd streamq
go mod tidy
go run main.go

Run with Docker

docker compose up --build

Run with demo producer (publishes a message every 2s)

docker compose --profile demo up --build

---

Endpoints

Topics

Method	Route	Description
POST	/topics	Create a topic
GET	/topics	List all topics with stats
GET	/topics/:name	Get a single topic's stats
DELETE	/topics/:name	Delete a topic, disconnect all subscribers

Publishing

Method	Route	Description
POST	/topics/:name/publish	Publish a message

Payload must be base64-encoded. Returns the assigned offset.

PAYLOAD=$(echo -n '{"amount":100,"currency":"GHS"}' | base64)

curl -X POST localhost:8080/topics/payments/publish \
  -H 'Content-Type: application/json' \
  -d "{\"payload\": \"$PAYLOAD\"}"

Subscribing

Method	Route	Description
GET	/topics/:name/subscribe/sse	SSE subscription
GET	/topics/:name/subscribe/ws	WebSocket subscription

Query parameters

Param	Default	Description
group	(none)	Consumer group ID. Omit for broadcast.
from_offset	(tail)	Replay from this absolute offset. Pass 0 for full history.

Observability

Route	Description
GET /metrics	Prometheus text exposition — scrape with Prometheus or Grafana Agent
GET /dashboard	Embedded React dashboard
GET /health	Health check for Docker/load balancer probes

---

Metrics reference

Metric	Type	Labels	Description
streamq_messages_published_total	Counter	topic	Messages successfully published
streamq_messages_delivered_total	Counter	topic, protocol	Messages written to subscriber connections
streamq_messages_dropped_total	Counter	topic	Messages dropped due to slow consumer (full buffer)
streamq_active_subscribers	Gauge	topic, protocol	Current live connections
streamq_consumer_group_lag	Gauge	topic, group	Messages a group is behind the latest offset
streamq_topic_message_count	Gauge	topic	Retained messages per topic

Plus standard Go runtime metrics: go_goroutines, go_memstats_heap_inuse_bytes, GC stats, etc.

---

Running tests

# All tests
go test ./...

# With race detector (recommended — tests concurrency logic)
go test ./... -race

# Verbose output
go test ./... -race -v

What's tested

Package	Coverage
store	Sequential offsets, memory eviction, WAL persistence, segment rotation, Since() edge cases, concurrent reads/writes
broker/topic	Subscribe replay, tail start, Unsubscribe channel close, broadcast fan-out, consumer group delivery, group + ungrouped coexistence, concurrent publish/subscribe
broker/consumer_group	Unknown group sentinel, commit/read, monotonic advance, topic isolation, PurgeTopic, concurrent access

---

Configuration

All config is loaded from .env via godenv.

Variable	Default	Description
SERVER_PORT	8080	TCP port to listen on
MAX_MESSAGE_RETENTION	1000	Max messages per topic before eviction in memory mode
STORAGE_MODE	memory	Storage backend: memory or wal
WAL_DIR	./data	Root directory for WAL segment files
WAL_SEGMENT_SIZE	67108864	Bytes per WAL segment before rotation
WAL_MAX_SEGMENTS	10	Max retained WAL segments per topic
LOG_LEVEL	info	Log verbosity: debug, info, warn, error

---

Roadmap

• In-memory message log with offset replay
• WebSocket + SSE delivery
• Consumer groups with round-robin delivery
• Prometheus metrics + /metrics endpoint
• Embedded React dashboard (/dashboard) with custom canvas charts
• Disk-backed message log (WAL)
• Explicit consumer acknowledgements over WebSocket
• Per-topic retention policies (TTL-based)
• Partitions + parallel consumer groups

---

Built with

• Go 1.25 — stdlib net/http for routing
• gorilla/websocket — WebSocket upgrade and framing
• prometheus/client_golang — Prometheus metrics
• godenv — zero-dependency env loader