hub

Hub

A Go library that embeds a NATS server and provides high-level abstractions for messaging, JetStream, Key-Value Store, and Object Store operations.

Features

• Embedded NATS Server - Run a full NATS server in-process
• Volatile Messaging - Standard publish/subscribe and request-reply patterns
• Persistent Messaging - JetStream with durable/ephemeral subscriptions
• Key-Value Store - Distributed key-value storage with TTL support
• Object Store - Large object storage with metadata
• Clustering - Built-in support for NATS clustering

Installation

go get git.sr.ht/~hyper/hub

Quick Start

package main

import (
    "fmt"
    "log"
    
    "git.sr.ht/~hyper/hub"
)

func main() {
    // Create hub with default options
    opts, err := hub.DefaultNodeOptions()
    if err != nil {
        log.Fatal(err)
    }
    
    h, err := hub.NewHub(opts)
    if err != nil {
        log.Fatal(err)
    }
    defer h.Shutdown()
    
    // Subscribe to messages
    cancel, err := h.SubscribeVolatileViaFanout("greetings", 
        func(subject string, msg []byte) ([]byte, bool) {
            fmt.Printf("Received: %s\n", string(msg))
            return nil, false
        }, 
        func(err error) {
            log.Printf("Error: %v", err)
        })
    if err != nil {
        log.Fatal(err)
    }
    defer cancel()
    
    // Publish a message
    err = h.PublishVolatile("greetings", []byte("Hello, World!"))
    if err != nil {
        log.Fatal(err)
    }
}

Usage

Configuration Options

Hub provides three default configurations:

Basic Node (Full Features)

opts, err := hub.DefaultNodeOptions()

• Full JetStream support
• Clustering capability
• Best for: standalone servers, development

Gateway Node (Network Bridge)

// import "net/url"
opts, err := hub.DefaultGatewayOptions()
remoteGatewayURL, _ := url.Parse("nats://remote:7222")
opts.GatewayRoutes = []struct {
    Name string
    URL  *url.URL
}{
    {Name: "remote", URL: remoteGatewayURL},
}

• Connects different networks
• Best for: multi-datacenter, network isolation

Edge Node (Lightweight)

// import "net/url"
opts, err := hub.DefaultEdgeOptions()
hubURL, _ := url.Parse("nats://central:7422")
opts.LeafNodeRoutes = []*url.URL{hubURL}

• No JetStream (uses central hub)
• Minimal resource usage
• Best for: IoT devices, edge computing

Messaging Patterns

1. Volatile Publish/Subscribe

// Subscribe
cancel, _ := h.SubscribeVolatileViaFanout("events", handler, errorHandler)
defer cancel()

// Publish
h.PublishVolatile("events", []byte("data"))

2. Queue Groups (Load Balancing)

// Multiple workers - only one receives each message
cancel, _ := h.SubscribeVolatileViaQueue("tasks", "workers", handler, errorHandler)

h.PublishVolatile("tasks", []byte("task data"))

3. Request/Reply

// Responder
h.SubscribeVolatileViaFanout("service", func(subject string, msg []byte) ([]byte, bool) {
    return []byte("response"), true // true = send response
}, errorHandler)

// Requester (import "time")
response, _ := h.RequestVolatile("service", []byte("request"), 5*time.Second)

4. Persistent Messaging (JetStream)

// Create stream (import "time")
config := &hub.PersistentConfig{
    Subjects: []string{"orders.>"},
    MaxMsgs:  10000,
    MaxAge:   24 * time.Hour,
}
h.CreateOrUpdatePersistent(config)

// Durable consumer (survives restart)
cancel, _ := h.SubscribePersistentViaDurable("consumer-1", "orders.new", 
    func(subject string, msg []byte) ([]byte, bool, bool) {
        // Process message
        return nil, false, true // last bool = ACK
    }, errorHandler)

// Publish persistent message
h.PublishPersistent("orders.new", []byte("order data"))

5. Key-Value Store

// Create KV store (import "time")
kvConfig := hub.KeyValueStoreConfig{
    Bucket:  "settings",
    TTL:     30 * 24 * time.Hour,
}
h.CreateOrUpdateKeyValueStore(kvConfig)

// Store and retrieve
h.PutToKeyValueStore("settings", "user:123", []byte(`{"theme":"dark"}`))
data, revision, _ := h.GetFromKeyValueStore("settings", "user:123")

6. Object Store

// Create object store
objConfig := hub.ObjectStoreConfig{
    Bucket:   "files",
    MaxBytes: hub.NewSizeFromGigabytes(10),
}
h.CreateObjectStore(objConfig)

// Store object with metadata
metadata := map[string]string{"type": "document"}
h.PutToObjectStore("files", "doc.pdf", fileData, metadata)

// Retrieve object
data, _ := h.GetFromObjectStore("files", "doc.pdf")

API Reference

Hub Management

• NewHub(opts *Options) (*Hub, error) - Create and start hub
• Shutdown() - Gracefully shutdown

Volatile Messaging

• SubscribeVolatileViaFanout(subject, handler, errHandler) - Subscribe (all receive)
• SubscribeVolatileViaQueue(subject, queue, handler, errHandler) - Queue subscribe (one receives)
• PublishVolatile(subject, msg) - Publish message
• RequestVolatile(subject, msg, timeout) - Request with response

Persistent Messaging (JetStream)

• CreateOrUpdatePersistent(config) - Create/update stream
• SubscribePersistentViaDurable(id, subject, handler, errHandler) - Durable consumer
• SubscribePersistentViaEphemeral(subject, handler, errHandler) - Ephemeral consumer
• PublishPersistent(subject, msg) - Publish to stream

Key-Value Store

• CreateOrUpdateKeyValueStore(config) - Create/update KV store
• GetFromKeyValueStore(bucket, key) - Get value
• PutToKeyValueStore(bucket, key, value) - Store value
• UpdateToKeyValueStore(bucket, key, value, revision) - Update with version check
• DeleteFromKeyValueStore(bucket, key) - Delete key
• PurgeKeyValueStore(bucket, key) - Purge all revisions of a key

Object Store

• CreateObjectStore(config) - Create object store
• GetFromObjectStore(bucket, key) - Get object
• PutToObjectStore(bucket, key, data, metadata) - Store object
• DeleteFromObjectStore(bucket, key) - Delete object

Handler Signatures

// Volatile handler
func(subject string, msg []byte) (response []byte, sendReply bool)

// Persistent handler
func(subject string, msg []byte) (response []byte, sendReply bool, ack bool)

// Error handler
func(error)

Examples

Clustering

// import "net/url"
// Node 1
opts1, _ := hub.DefaultNodeOptions()
opts1.Name = "node1"
h1, _ := hub.NewHub(opts1)

// Node 2
opts2, _ := hub.DefaultNodeOptions()
opts2.Name = "node2"
opts2.Routes = []*url.URL{{Host: "node1:6222"}}
h2, _ := hub.NewHub(opts2)

Edge Computing

// import "net/url"
// Central hub
hubOpts, _ := hub.DefaultNodeOptions()
hubOpts.Name = "central"
central, _ := hub.NewHub(hubOpts)

// Edge node
edgeOpts, _ := hub.DefaultEdgeOptions()
edgeOpts.LeafNodeRoutes = []*url.URL{{Host: "central:7422"}}
edge, _ := hub.NewHub(edgeOpts)

Pattern Selection Guide

Requirements

• Go 1.24 or later
• Linux, macOS, or Windows

License

Apache License 2.0 - see LICENSE file for details.