wsbrokerclient

WSBrokerClient

WSBrokerClient is an enterprise-grade Go client library for WSBroker. It provides seamless WebSocket and REST integration, robust mutual TLS (mTLS) certificate management with hot-reloading via certloader, jittered exponential backoff auto-reconnection, dynamic pub/sub channel state management, and stream subscriptions.

Table of Contents

• Features
• Architecture & Diagrams
  • System Architecture
  • Connection & Channel Lifecycle
  • Automatic Reconnection & Resubscription Flow
  • mTLS Handshake & Security Model
• Installation
• Quick Start
• Security & mTLS Configuration
  • mTLS via certloader (MiniVault, File, Env, Dev)
  • mTLS via standard crypto/tls
  • Bearer Token Authentication
• Core Usage Guide
  • Dynamic Channel Subscriptions & Streaming
  • Event Handlers & Callbacks
  • Direct Client-to-Client Messaging
  • Publishing & Broadcasting
  • Heartbeat & Ping/Pong Handling
• REST Management API Client
• Configuration Reference
• License

Features

• High-Performance WebSocket Engine: Powered by coder/websocket with concurrent-safe write synchronization and context-aware timeouts.
• Enterprise mTLS Security: Full integration with go.ub.unibas.ch/cloud/certloader/v2 supporting dynamic hot-reloading of client certificates, MiniVault PKI, local file pairs, environment variables, or standard Go *tls.Config.
• Intelligent Auto-Reconnection: Built-in connection supervisor with jittered exponential backoff and transparent channel resubscription upon network restore.
• Dynamic Pub/Sub & Stream Subscriptions: Join/leave channels dynamically, consume channel events via Go channels (Subscription.Events()), or register target/type-specific handlers.
• Direct Point-to-Point Messaging: Send targeted events directly to specific connected clients.
• Automatic Heartbeat Management: Responds to server pings with standard pong envelopes and sends periodic client-side keepalive pings.
• RESTful Client Included: Full HTTP client implementation for WSBroker REST management endpoints (/api/v1/publish, /api/v1/channels, /api/v1/clients, etc.).

Architecture & Diagrams

System Architecture

graph TB
    subgraph App["Your Go Application"]
        WSClient["WSBroker WebSocket Client\n(pkg/client)"]
        RESTClient["WSBroker REST Client\n(pkg/restclient)"]
        CertLoader["CertLoader PKI Engine\n(go.ub.unibas.ch/cloud/certloader/v2)"]
    end

    subgraph Server["WSBroker Server"]
        WSEndpoint["WebSocket Controller\n(/ws/:name, /ws/channel/:channel)"]
        RESTEndpoint["REST Controller\n(/api/v1/publish, /api/v1/channels)"]
        BrokerEngine["Connection Manager & Worker Pool"]
    end

    CertLoader -->|Supplies mTLS Certificates| WSClient
    CertLoader -->|Supplies mTLS Certificates| RESTClient
    WSClient <-->|mTLS / WSS Encrypted WebSocket| WSEndpoint
    RESTClient -->|mTLS / HTTPS POST & GET| RESTEndpoint
    WSEndpoint <--> BrokerEngine
    RESTEndpoint --> BrokerEngine

Connection & Channel Lifecycle

sequenceDiagram
    autonumber
    actor App as Your Application
    participant Client as WSBrokerClient
    participant Server as WSBroker Server

    Note over App,Server: 1. Connection & Initial Subscriptions
    App->>Client: Connect(ctx)
    Client->>Server: WebSocket Handshake (GET /ws/node-01?channel=alerts)
    Server-->>Client: 101 Switching Protocols (Connected)
    Client-->>App: OnConnect Callback Triggered

    Note over App,Server: 2. Dynamic Channel Attach & Detach
    App->>Client: Attach("metrics", "logs")
    Client->>Server: {"type": "attach", "target": "metrics"}
    Client->>Server: {"type": "attach", "target": "logs"}
    Server-->>Client: Subscribed to metrics & logs

    App->>Client: Detach("logs")
    Client->>Server: {"type": "detach", "target": "logs"}
    Server-->>Client: Unsubscribed from logs

    Note over App,Server: 3. Message Broadcasting & Streaming
    App->>Client: Broadcast("metrics", {"cpu": 45.2})
    Client->>Server: {"type": "message", "target": "metrics", "data": {"cpu": 45.2}}
    Server-->>Client: Inbound Channel Message Dispatch
    Client-->>App: Delivery to Subscription.Events() channel

Automatic Reconnection & Resubscription Flow

sequenceDiagram
    autonumber
    participant App as Application
    participant Client as WSBrokerClient
    participant Server as WSBroker Server

    Note over Client,Server: Normal Operation (Subscribed to "alerts", "telemetry")
    Server--xClient: Network Drop / Connection Lost
    Client->>Client: Update State to Reconnecting
    Client-->>App: OnDisconnect(err) Callback

    loop Exponential Backoff with Jitter
        Client->>Client: Calculate Next Sleep Duration
        Client->>Server: Attempt WebSocket Handshake
        alt Connection Failed
            Server--xClient: Handshake Refused / Network Down
            Client-->>App: OnError(err) Callback
        else Connection Successful
            Server-->>Client: 101 Switching Protocols
            Client->>Client: Update State to Connected
        end
    end

    Note over Client,Server: Channel Resubscription
    Client->>Server: {"type": "attach", "target": "alerts"}
    Client->>Server: {"type": "attach", "target": "telemetry"}
    Client-->>App: OnReconnect() and OnConnect() Callbacks
    Note over App,Server: Seamless Resumption of Stream Flow

mTLS Handshake & Security Model

sequenceDiagram
    autonumber
    participant Client as WSBrokerClient
    participant PKI as CertLoader / MiniVault
    participant Server as WSBroker (mTLS)

    PKI->>Client: Load Client Cert, Key & Trusted Root CA
    Client->>Server: TLS ClientHello (Supported Ciphers, TLS 1.3)
    Server->>Client: ServerHello + Server Certificate
    Client->>Client: Validate Server Cert against Root CA
    Server->>Client: CertificateRequest (Require Client Certificate)
    Client->>Server: Client Certificate + CertificateVerify
    Server->>Server: Validate Client Cert against Server CA Pool
    Server-->>Client: Finished (Secure mTLS Tunnel Established)
    Client->>Server: Upgrade: websocket (Authorization: Bearer <token>)

Installation

go get github.com/je4/wsbrokerclient

Quick Start

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/je4/wsbrokerclient"
)

type SensorTelemetry struct {
	DeviceID    string  `json:"device_id"`
	Temperature float64 `json:"temperature"`
	Humidity    float64 `json:"humidity"`
}

func main() {
	ctx := context.Background()

	// 1. Initialize Client
	cli, err := wsbrokerclient.New(wsbrokerclient.DefaultConfig(),
		wsbrokerclient.WithEndpoint("ws://localhost:8080"),
		wsbrokerclient.WithClientName("sensor-node-01"),
		wsbrokerclient.WithInitialChannels("telemetry", "alerts"),
		wsbrokerclient.WithAutoReconnect(true),
	)
	if err != nil {
		panic(err)
	}
	defer cli.Close()

	// 2. Stream-based subscription
	sub, err := cli.Subscribe(ctx, "telemetry")
	if err != nil {
		panic(err)
	}
	defer sub.Unsubscribe()

	go func() {
		for evt := range sub.Events() {
			var data SensorTelemetry
			if err := evt.UnmarshalData(&data); err == nil {
				fmt.Printf("[%s] Temp: %.1f°C, Humidity: %.1f%%\n", data.DeviceID, data.Temperature, data.Humidity)
			}
		}
	}()

	// 3. Connect to WSBroker
	if err := cli.Connect(ctx); err != nil {
		panic(err)
	}

	// 4. Broadcast sensor data
	_ = cli.Broadcast(ctx, "telemetry", SensorTelemetry{
		DeviceID:    "sensor-node-01",
		Temperature: 22.4,
		Humidity:    48.0,
	})

	time.Sleep(2 * time.Second)
}

Security & mTLS Configuration

mTLS via certloader (MiniVault, File, Env, Dev)

WSBrokerClient directly integrates with go.ub.unibas.ch/cloud/certloader/v2. It handles dynamic client certificates with automatic hot-reloading:

import (
	"github.com/je4/wsbrokerclient"
	"go.ub.unibas.ch/cloud/certloader/v2/pkg/loader"
)

// Example: File-based mTLS
certConfig := &loader.Config{
	Type: "FILE",
	File: &loader.FileConfig{
		Cert: "/etc/ssl/certs/client.cert.pem",
		Key:  "/etc/ssl/private/client.key.pem",
		CA:   []string{"/etc/ssl/certs/ca.crt"},
	},
}

cli, err := wsbrokerclient.New(wsbrokerclient.DefaultConfig(),
	wsbrokerclient.WithEndpoint("wss://broker.internal.org:8080"),
	wsbrokerclient.WithClientName("service-a"),
	wsbrokerclient.WithTLSCertLoader(certConfig),
)

mTLS via standard crypto/tls

import (
	"crypto/tls"
	"crypto/x509"
	"os"

	"github.com/je4/wsbrokerclient"
)

cert, _ := tls.LoadX509KeyPair("client.crt", "client.key")
caCert, _ := os.ReadFile("ca.crt")
caPool := x509.NewCertPool()
caPool.AppendCertsFromPEM(caCert)

tlsConfig := &tls.Config{
	Certificates: []tls.Certificate{cert},
	RootCAs:      caPool,
	MinVersion:   tls.VersionTLS13,
}

cli, err := wsbrokerclient.New(wsbrokerclient.DefaultConfig(),
	wsbrokerclient.WithEndpoint("wss://localhost:8080"),
	wsbrokerclient.WithTLSConfig(tlsConfig),
)

Bearer Token Authentication

cli, err := wsbrokerclient.New(wsbrokerclient.DefaultConfig(),
	wsbrokerclient.WithEndpoint("wss://localhost:8080"),
	wsbrokerclient.WithBearerToken("your-secure-bearer-token"),
)

Core Usage Guide

Dynamic Channel Subscriptions & Streaming

// Stream subscription using Go channel
sub, err := cli.Subscribe(ctx, "orders")
if err != nil {
	log.Fatal(err)
}
defer sub.Unsubscribe()

for evt := range sub.Events() {
	fmt.Printf("Received order event: %s\n", evt.String())
}

// Or manually attach/detach channels
_ = cli.Attach(ctx, "notifications", "billing")
_ = cli.Detach(ctx, "billing")

Event Handlers & Callbacks

// Listen to all messages targeted to "alerts" channel
cli.OnChannel("alerts", func(evt *wsbrokerclient.Event) {
	fmt.Println("Alert received:", evt)
})

// Listen to direct messages sent to this client
cli.OnDirect(func(evt *wsbrokerclient.Event) {
	fmt.Printf("Direct message from %s: %s\n", evt.Source, evt.Data)
})

// Connection lifecycle callbacks
cli.OnConnect(func() {
	fmt.Println("Connected to WSBroker!")
})

cli.OnDisconnect(func(err error) {
	fmt.Printf("Disconnected from WSBroker: %v\n", err)
})

cli.OnReconnect(func() {
	fmt.Println("Successfully reconnected and re-subscribed!")
})

Direct Client-to-Client Messaging

// Send a direct message to a specific client identifier ("worker-42")
err := cli.SendDirect(ctx, "worker-42", map[string]string{
	"task": "process_chunk",
	"chunk_id": "9812",
})

Publishing & Broadcasting

// Broadcast structured JSON to all subscribers of "events"
err := cli.Broadcast(ctx, "events", map[string]any{
	"event": "user_signup",
	"user_id": 1042,
})

REST Management API Client

WSBrokerClient includes a dedicated REST API client for interacting with WSBroker's administrative endpoints:

import (
	"context"
	"fmt"

	"github.com/je4/wsbrokerclient"
)

func main() {
	ctx := context.Background()

	restCli, err := wsbrokerclient.NewRESTClient(wsbrokerclient.RESTConfig{
		BaseURL: "https://localhost:8080",
		Bearer:  "secret-token",
		// TLSCertLoader or TLSConfig can also be configured here for mTLS REST access
	})
	if err != nil {
		panic(err)
	}
	defer restCli.Close()

	// 1. Health-check
	pong, _ := restCli.Ping(ctx)
	fmt.Println("Ping status:", pong)

	// 2. Publish to channel
	resp, _ := restCli.PublishToChannel(ctx, "alerts", "rest-admin", "Maintenance scheduled")
	fmt.Printf("Published to %d subscribers\n", resp.Subscribers)

	// 3. List active channels
	channels, _ := restCli.GetChannels(ctx)
	for _, ch := range channels {
		fmt.Printf("Channel %s: %d active clients\n", ch.Name, ch.Subscribers)
	}

	// 4. Force disconnect a client
	_ = restCli.DisconnectClient(ctx, "bad-client-node")
}

Configuration Reference

Client Configuration (client.Config)

License

Copyright © 2024-2026 Jürgen Enge / Universitätsbibliothek Basel.

Licensed under the Apache License, Version 2.0.