jay/go-honeybee
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Updated documentation.

Browse Source
This commit is contained in:
Jay
2026-05-20 10:44:53 -04:00
parent 59f7b86a2e
commit 5b31db304a
3 changed files with 220 additions and 306 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CONFIG.md
+43 -102
View File
@@ -2,13 +2,38 @@
All configuration is done through option functions applied at construction time. Invalid values return errors at application time; configs produced through `NewXConfig` constructors are validated at construction and cannot be saved in an invalid state. All configuration is done through option functions applied at construction time. Invalid values return errors at application time; configs produced through `NewXConfig` constructors are validated at construction and cannot be saved in an invalid state.
There are three config scopes. `ConnectionConfig` governs a single WebSocket connection: its write behavior, ping interval, buffers, and retry policy. `WorkerConfig` governs a single worker's behavior, with separate types for inbound and outbound. `PoolConfig` bundles a connection config, a worker config, and an optional worker factory — it is a thin container. There are three config scopes. `ConnectionConfig` governs a single WebSocket connection: its write behavior, ping interval, buffers, and retry policy. `WorkerConfig` governs a single worker's behavior. `PoolConfig` bundles a connection config, a worker config, and an optional worker factory — it is a thin container.
Logging can be controlled independently at the pool, worker, and connection levels. Each scope has a `LoggingEnabled` flag and a `LogLevel` override. When `LoggingEnabled` is false, no logger is constructed for that scope regardless of what handler is passed to the pool. When `LogLevel` is set, it overrides the handler's own level filter for that scope only, using a wrapping handler that enforces the minimum level before delegating. When `LogLevel` is nil, the handler's own level filtering applies unchanged. Logging can be controlled independently at the pool, worker, and connection levels. Each scope has a `LoggingEnabled` flag and a `LogLevel` override. When `LoggingEnabled` is false, no logger is constructed for that scope regardless of what handler is passed to the pool. When `LogLevel` is set, it overrides the handler's own level filter for that scope only, using a wrapping handler that enforces the minimum level before delegating. When `LogLevel` is nil, the handler's own level filtering applies unchanged.
## Defaults
| Scope | Setting | Default | Disabled by | Notes |
|---|---|---|---|---|
| Connection | `WriteTimeout` | 30s | `0` | Per-message write deadline |
| Connection | `RequestHeader` | User-Agent | — | honeybee/0.1.0 |
| Connection | `PingInterval` | 20s | `0` | ±10% jitter applied per interval |
| Connection | `IncomingBufferSize` | 100 | — | Must be positive |
| Connection | `ErrorsBufferSize` | 10 | — | Must be positive |
| Connection | `LoggingEnabled` | true | `false` | |
| Connection | `LogLevel` | nil | — | nil defers to handler's own filter |
| Retry | enabled | yes | `WithoutRetry()` | Governs `Connect()` only |
| Retry | `MaxRetries` | 0 | — | 0 means infinite |
| Retry | `InitialDelay` | 1s | — | Must be positive |
| Retry | `MaxDelay` | 60s | — | Must be ≥ InitialDelay |
| Retry | `JitterFactor` | 0.2 | `0.0` | Range [0.0, 1.0] |
| Pool | `InboxBufferSize` | 256 | — | Must be positive |
| Pool | `EventsBufferSize` | 10 | — | Must be positive |
| Pool | `LoggingEnabled` | true | `false` | |
| Pool | `LogLevel` | nil | — | |
| Worker | `KeepaliveTimeout` | 60s | `0` | 0 disables keepalive |
| Worker | `ReconnectDelay` | 2s | `0` | 0 means reconnect immediately |
| Worker | `LoggingEnabled` | true | `false` | |
| Worker | `LogLevel` | nil | — | |
## Connection Options ## Connection Options
`ConnectionConfig` is defined in the `transport` package. Import `git.wisehodl.dev/jay/go-honeybee/transport` to construct one, then pass it to a pool via `inbound.WithConnectionConfig` or `outbound.WithConnectionConfig`. `ConnectionConfig` is defined in the `transport` package. Import `git.wisehodl.dev/jay/go-honeybee/transport` to construct one, then pass it to a pool via `honeybee.WithConnectionConfig`.
These options are passed to `transport.NewConnectionConfig`. These options are passed to `transport.NewConnectionConfig`.
@@ -38,7 +63,7 @@ Sets the capacity of the channel that carries connection-level errors to the con
### Retry ### Retry
The retry policy governs the `Connect()` call only. It does not affect outbound worker reconnection, which is controlled by `ReconnectDelay` on the worker config. The retry policy governs the `Connect()` call only. It does not affect worker reconnection, which is controlled by `ReconnectDelay` on the worker config.
**`WithoutRetry()`** **`WithoutRetry()`**
Disables retry entirely. `Connect()` returns on the first dial failure. Disables retry entirely. `Connect()` returns on the first dial failure.
@@ -63,133 +88,49 @@ Enables or disables logging for the connection. When false, no logger is constru
**`WithConnectionLogLevel(slog.Level)`** **`WithConnectionLogLevel(slog.Level)`**
Overrides the minimum log level for connection-scoped records. Does not affect pool or worker logging. Overrides the minimum log level for connection-scoped records. Does not affect pool or worker logging.
## Inbound Pool Options ## Pool Options
Import `git.wisehodl.dev/jay/go-honeybee/inbound`. Import `git.wisehodl.dev/jay/go-honeybee`.
### Pool ### Pool
These are passed to `inbound.NewPoolConfig`. These are passed to `honeybee.NewPoolConfig`.
**`inbound.WithInboxBufferSize(int)`** **`honeybee.WithInboxBufferSize(int)`**
Sets the capacity of the pool's shared inbox channel. Must be at least 1. Sets the capacity of the pool's shared inbox channel. Must be at least 1.
**`inbound.WithEventsBufferSize(int)`** **`honeybee.WithEventsBufferSize(int)`**
Sets the capacity of the pool's events channel. Must be at least 1. Sets the capacity of the pool's events channel. Must be at least 1.
**`inbound.WithPoolLoggingEnabled(bool)`** **`honeybee.WithPoolLoggingEnabled(bool)`**
Enables or disables pool-level logging. Enables or disables pool-level logging.
**`inbound.WithPoolLogLevel(slog.Level)`** **`honeybee.WithPoolLogLevel(slog.Level)`**
Overrides the minimum log level for pool-scoped records only. Overrides the minimum log level for pool-scoped records only.
### Worker ### Worker
These are passed to `inbound.NewWorkerConfig` or embedded in the pool config. These are passed to `honeybee.NewWorkerConfig` or embedded in the pool config.
**`inbound.WithInactivityTimeout(duration)`** **`honeybee.WithKeepaliveTimeout(duration)`**
Enables the inactivity watchdog. When no data messages or pong replies arrive within this duration, the peer is evicted and `inbound.EventEvictedPolicy` is emitted. When set to zero, the watchdog is disabled and connections persist until removed or remotely terminated. Must not be negative.
**`inbound.WithMaxQueueSize(int)`**
Bounds the worker's internal message queue. When the queue is full, the oldest undelivered message is dropped. When set to zero, the queue is unbounded. Must not be negative.
**`inbound.WithWorkerLoggingEnabled(bool)`**
Enables or disables worker-level logging.
**`inbound.WithWorkerLogLevel(slog.Level)`**
Overrides the minimum log level for worker-scoped records only.
### Wiring
**`inbound.WithConnectionConfig(*transport.ConnectionConfig)`**
Supplies a connection config that is applied to every socket added to the pool.
**`inbound.WithWorkerConfig(*inbound.WorkerConfig)`**
Supplies a worker config that is applied to every worker the pool creates.
**`inbound.WithWorkerFactory(inbound.WorkerFactory)`**
Replaces the default worker constructor. See [EXTEND.md](EXTEND.md) for the factory contract.
## Outbound Pool Options
Import `git.wisehodl.dev/jay/go-honeybee/outbound`.
### Pool
These are passed to `outbound.NewPoolConfig`.
**`outbound.WithInboxBufferSize(int)`**
Sets the capacity of the pool's shared inbox channel. Must be at least 1.
**`outbound.WithEventsBufferSize(int)`**
Sets the capacity of the pool's events channel. Must be at least 1.
**`outbound.WithPoolLoggingEnabled(bool)`**
Enables or disables pool-level logging.
**`outbound.WithPoolLogLevel(slog.Level)`**
Overrides the minimum log level for pool-scoped records only.
### Worker
These are passed to `outbound.NewWorkerConfig` or embedded in the pool config.
**`outbound.WithKeepaliveTimeout(duration)`**
Enables the keepalive mechanism. When no heartbeat (inbound data, outbound send, or pong reply) is observed within this duration, the current connection is closed and a new one is dialed. When set to zero, keepalive is disabled. Must not be negative. Enables the keepalive mechanism. When no heartbeat (inbound data, outbound send, or pong reply) is observed within this duration, the current connection is closed and a new one is dialed. When set to zero, keepalive is disabled. Must not be negative.
**`outbound.WithReconnectDelay(duration)`** **`honeybee.WithReconnectDelay(duration)`**
Sets the delay between a disconnect and the next dial attempt. Applies after every session end, including those triggered by keepalive. The default of 2 seconds prevents tight reconnect loops against unavailable peers. Set to zero in tests or when immediate reconnection is required. Must not be negative. Sets the delay between a disconnect and the next dial attempt. Applies after every session end, including those triggered by keepalive. The default of 2 seconds prevents tight reconnect loops against unavailable peers. Set to zero in tests or when immediate reconnection is required. Must not be negative.
**`outbound.WithMaxQueueSize(int)`** **`honeybee.WithWorkerLoggingEnabled(bool)`**
Bounds the worker's internal message queue. When the queue is full, the oldest undelivered message is dropped. When set to zero, the queue is unbounded. Must not be negative.
**`outbound.WithWorkerLoggingEnabled(bool)`**
Enables or disables worker-level logging. Enables or disables worker-level logging.
**`outbound.WithWorkerLogLevel(slog.Level)`** **`honeybee.WithWorkerLogLevel(slog.Level)`**
Overrides the minimum log level for worker-scoped records only. Overrides the minimum log level for worker-scoped records only.
### Wiring ### Wiring
**`outbound.WithConnectionConfig(*transport.ConnectionConfig)`** **`honeybee.WithConnectionConfig(*transport.ConnectionConfig)`**
Supplies a connection config used when dialing each peer. Supplies a connection config used when dialing each peer.
**`outbound.WithWorkerConfig(*outbound.WorkerConfig)`** **`honeybee.WithWorkerConfig(*honeybee.WorkerConfig)`**
Supplies a worker config applied to every worker the pool creates. Supplies a worker config applied to every worker the pool creates.
**`outbound.WithWorkerFactory(outbound.WorkerFactory)`** **`honeybee.WithWorkerFactory(honeybee.WorkerFactory)`**
Replaces the default worker constructor. See [EXTEND.md](EXTEND.md) for the factory contract. Replaces the default worker constructor. See [EXTEND.md](EXTEND.md) for the factory contract.
## Defaults
| Scope | Setting | Default | Disabled by | Notes |
|---|---|---|---|---|
| Connection | `WriteTimeout` | 30s | `0` | Per-message write deadline |
| Connection | `RequestHeader` | User-Agent | — | honeybee/0.1.0 |
| Connection | `PingInterval` | 20s | `0` | ±10% jitter applied per interval |
| Connection | `IncomingBufferSize` | 100 | — | Must be positive |
| Connection | `ErrorsBufferSize` | 10 | — | Must be positive |
| Connection | `LoggingEnabled` | true | `false` | |
| Connection | `LogLevel` | nil | — | nil defers to handler's own filter |
| Retry | enabled | yes | `WithoutRetry()` | Governs `Connect()` only |
| Retry | `MaxRetries` | 0 | — | 0 means infinite |
| Retry | `InitialDelay` | 1s | — | Must be positive |
| Retry | `MaxDelay` | 60s | — | Must be ≥ InitialDelay |
| Retry | `JitterFactor` | 0.2 | `0.0` | Range [0.0, 1.0] |
| Inbound pool | `InboxBufferSize` | 256 | — | Must be positive |
| Inbound pool | `EventsBufferSize` | 10 | — | Must be positive |
| Inbound pool | `LoggingEnabled` | true | `false` | |
| Inbound pool | `LogLevel` | nil | — | |
| Inbound worker | `InactivityTimeout` | 0 | `0` | 0 disables watchdog |
| Inbound worker | `MaxQueueSize` | 0 | `0` | 0 means unbounded |
| Inbound worker | `LoggingEnabled` | true | `false` | |
| Inbound worker | `LogLevel` | nil | — | |
| Outbound pool | `InboxBufferSize` | 256 | — | Must be positive |
| Outbound pool | `EventsBufferSize` | 10 | — | Must be positive |
| Outbound pool | `LoggingEnabled` | true | `false` | |
| Outbound pool | `LogLevel` | nil | — | |
| Outbound worker | `KeepaliveTimeout` | 60s | `0` | 0 disables keepalive |
| Outbound worker | `ReconnectDelay` | 2s | `0` | 0 means reconnect immediately |
| Outbound worker | `MaxQueueSize` | 0 | `0` | 0 means unbounded |
| Outbound worker | `LoggingEnabled` | true | `false` | |
| Outbound worker | `LogLevel` | nil | — | |
EXTEND.md
+18 -72
View File
@@ -1,10 +1,10 @@
# Extending Pools # Extending the Pool
The pool owns peer registration, event plumbing, and lifecycle management. The worker owns everything that happens on the wire between registration and the terminal event. Everything between `pool.Add` or `pool.Connect` and the final disconnect event is the worker's responsibility, and it is fully replaceable. The pool owns peer registration, event plumbing, and lifecycle management. The worker owns everything that happens on the wire between registration and the terminal event. Everything between `pool.Connect` and the final disconnect event is the worker's responsibility, and it is fully replaceable.
## The Worker Interface ## The Worker Interface
Both pools accept any type that satisfies: The pool accepts any type that satisfies:
```go ```go
type Worker interface { type Worker interface {
@@ -13,7 +13,7 @@ type Worker interface {
Send(data []byte) error Send(data []byte) error
Stats() WorkerStats Stats() WorkerStats
} }
```` ```
The behavioral contract for each method: The behavioral contract for each method:
@@ -30,22 +30,12 @@ The behavioral contract for each method:
The pool constructs a `PoolPlugin` and passes it to `Start`. It gives the worker access to pool-level channels and the logging handler. The pool constructs a `PoolPlugin` and passes it to `Start`. It gives the worker access to pool-level channels and the logging handler.
```go ```go
// inbound.PoolPlugin
type PoolPlugin struct {
Inbox chan<- inbound.InboxMessage
Events chan<- inbound.PoolEvent
InboxCounter *atomic.Uint64
OnExit inbound.OnExitFunction // inbound only
Handler slog.Handler
}
// outbound.PoolPlugin
type PoolPlugin struct { type PoolPlugin struct {
ID string ID string
Inbox chan<- outbound.InboxMessage Inbox chan<- honeybee.InboxMessage
Events chan<- outbound.PoolEvent Events chan<- honeybee.PoolEvent
InboxCounter *atomic.Uint64 InboxCounter *atomic.Uint64
Dialer outbound.Dialer Dialer honeybee.Dialer
ConnectionConfig *transport.ConnectionConfig ConnectionConfig *transport.ConnectionConfig
Handler slog.Handler Handler slog.Handler
} }
@@ -53,69 +43,27 @@ type PoolPlugin struct {
**`Inbox`** The shared channel that delivers received messages to the pool's consumer. All peers in the pool deliver to the same inbox channel. Workers must include their peer ID in each `InboxMessage`. **`Inbox`** The shared channel that delivers received messages to the pool's consumer. All peers in the pool deliver to the same inbox channel. Workers must include their peer ID in each `InboxMessage`.
**`Events`** The shared channel for lifecycle events. Outbound workers emit `outbound.EventConnected` and `outbound.EventDisconnected` directly. Inbound workers do not touch this channel directly; instead they call `OnExit`, and the pool translates the exit kind into the appropriate event. All events include a timestamp in the `At` field. **`Events`** The shared channel for lifecycle events. Workers emit `honeybee.EventConnected` and `honeybee.EventDisconnected` directly. All events include a timestamp in the `At` field.
**`InboxCounter`** An atomic counter owned by the pool. Workers must increment this once for each message forwarded to `Inbox`. The pool reads it in `Stats()`. **`InboxCounter`** An atomic counter owned by the pool. Workers must increment this once for each message forwarded to `Inbox`. The pool reads it in `Stats()`.
**`OnExit` (inbound only)** Must be called exactly once when an inbound worker exits on its own initiative — that is, when the peer closed the connection or the watchdog fired, not when `Stop` was called. The pool wraps the underlying function in a `sync.Once` as a safety net, but a well-behaved worker calls it once. Calling `OnExit` removes the peer from the pool's registry and emits the appropriate event.
**`Handler`** The `slog.Handler` passed to the pool constructor. The pool constructs and injects scoped loggers before calling the factory, so most workers will use the logger they receive from the factory rather than constructing one from `Handler` directly. `Handler` is available for workers that manage sub-components that need their own loggers. **`Handler`** The `slog.Handler` passed to the pool constructor. The pool constructs and injects scoped loggers before calling the factory, so most workers will use the logger they receive from the factory rather than constructing one from `Handler` directly. `Handler` is available for workers that manage sub-components that need their own loggers.
## Extending the Inbound Pool ## Extending the Pool
### Factory Signature ### Factory Signature
```go ```go
type inbound.WorkerFactory func( type honeybee.WorkerFactory func(
ctx context.Context,
id string,
conn *transport.Connection,
config *inbound.WorkerConfig,
logger *slog.Logger,
) (inbound.Worker, error)
```
The pool calls the factory under its write lock when `Add` or `Replace` is called. The factory must return without blocking. The pool constructs `logger` from the worker logging config before calling the factory; pass it to your worker or ignore it.
The factory is set via `inbound.WithWorkerFactory` on the pool config.
### Building Blocks
The default inbound worker is assembled from these exported functions. Each can be reused, replaced, or wrapped independently.
**`RunReader(ctx, onExit, conn, messages, heartbeat, logger)`** Reads from `conn.Incoming()` until the channel closes or `ctx` is cancelled. Forwards each message to `messages` and sends a signal on `heartbeat` for each message received. When the channel closes, classifies the exit — clean disconnect, unexpected close, or read error — and calls `onExit` with the appropriate `WorkerExitKind`. Does not call `onExit` on context cancellation.
**`RunHeartbeatForwarder(ctx, conn, heartbeat, logger)`** Reads from `conn.Heartbeat()` and forwards each signal to `heartbeat`. This propagates pong replies from the connection layer into the worker's heartbeat channel, so pongs reset the inactivity watchdog alongside data messages.
**`RunQueue(id, ctx, in, out, maxQueueSize, droppedCount, bufferDepth)`** Buffers messages between the reader and the forwarder. When `maxQueueSize` is positive and the queue is full, the oldest message is dropped and `droppedCount` is incremented. When `maxQueueSize` is zero, the queue grows without bound. `bufferDepth` is an `*atomic.Int64` maintained as the current number of items in the queue.
**`RunForwarder(id, ctx, messages, inbox, workerProcessedCount, poolInboxCount)`** Reads from `messages` and writes to `inbox`. Increments both counters on each successful delivery.
**`RunWatchdog(ctx, onInactive, heartbeat, timeout, logger)`** Monitors `heartbeat`. Resets a timer on each signal. When the timer fires, calls `onInactive(ExitPolicy)`. When `timeout` is zero, the watchdog is disabled: it drains `heartbeat` without acting and exits when `ctx` is cancelled.
### Replacement Patterns
**Swap one block.** Embed `*inbound.DefaultWorker`, override `Start`, reuse the goroutines you want unchanged, and substitute your own implementation for the one you are replacing. For example, to tag every forwarded message with a sequence number, reuse `RunReader`, `RunQueue`, and `RunWatchdog` verbatim, and write a custom forwarder.
**Wrap the plugin.** Intercept `Inbox` by substituting a channel you own, process messages, then forward to the original. The worker signature is unchanged; you compose behavior by controlling what the building blocks see.
**Implement from scratch.** Satisfy the `Worker` interface directly. The only obligations are the behavioral contracts on `Start`, `Stop`, `Send`, `Stats`, and `OnExit` described above.
## Extending the Outbound Pool
### Factory Signature
```go
type outbound.WorkerFactory func(
ctx context.Context, ctx context.Context,
id string, id string,
logger *slog.Logger, logger *slog.Logger,
) (outbound.Worker, error) ) (honeybee.Worker, error)
``` ```
The pool calls the factory under its write lock when `Connect` is called. The factory must return without blocking. Note that the outbound factory does not receive a `*transport.Connection`; the worker is responsible for dialing and managing its own connections. The pool constructs `logger` from the worker logging config before calling the factory. The pool calls the factory under its write lock when `Connect` is called. The factory must return without blocking. The worker is responsible for dialing and managing its own connections. The pool constructs `logger` from the worker logging config before calling the factory.
The factory is set via `outbound.WithWorkerFactory` on the pool config. The factory is set via `honeybee.WithWorkerFactory` on the pool config.
### Building Blocks ### Building Blocks
@@ -123,14 +71,12 @@ The factory is set via `outbound.WithWorkerFactory` on the pool config.
**`RunKeepalive(ctx, heartbeat, keepalive, timeout, logger)`** Monitors `heartbeat`. Resets a timer on each signal. When the timer fires, sends a signal on `keepalive` to notify the session that the connection should be replaced. When `timeout` is zero, keepalive is disabled: it drains `heartbeat` without acting and exits when `ctx` is cancelled. **`RunKeepalive(ctx, heartbeat, keepalive, timeout, logger)`** Monitors `heartbeat`. Resets a timer on each signal. When the timer fires, sends a signal on `keepalive` to notify the session that the connection should be replaced. When `timeout` is zero, keepalive is disabled: it drains `heartbeat` without acting and exits when `ctx` is cancelled.
**`RunReader(ctx, onStop, conn, messages, heartbeat, logger)`** Reads from `conn.Incoming()` until the channel closes or `ctx` is cancelled. Forwards each message to `messages` and sends a signal on `heartbeat`. On exit, calls `conn.Close()` and then `onStop`. **`RunReader(id, ctx, onStop, conn, inbox, heartbeat, logger)`** Reads from `conn.Incoming()` until the channel closes or `ctx` is cancelled. Builds an `InboxMessage` inline with the peer ID and writes it directly to `inbox`. Sends a signal on `heartbeat` for each message. On exit, calls `conn.Close()` and then `onStop`.
**`RunHeartbeatForwarder(ctx, conn, heartbeat, logger)`** Reads from `conn.Heartbeat()` and forwards each signal to `heartbeat`. Propagates pong replies into the worker's heartbeat channel so pongs reset the keepalive timer alongside data messages and sends. **`RunHeartbeatForwarder(ctx, conn, heartbeat, logger)`** Reads from `conn.Heartbeat()` and forwards each signal to `heartbeat`. Propagates pong replies into the worker's heartbeat channel so pongs reset the keepalive timer alongside data messages and sends.
**`RunStopMonitor(ctx, onStop, conn, keepalive, logger)`** Waits for either `ctx.Done` or a signal on `keepalive`. On either, calls `conn.Close()` and then `onStop`. This is how a keepalive expiry propagates into a session tear-down. **`RunStopMonitor(ctx, onStop, conn, keepalive, logger)`** Waits for either `ctx.Done` or a signal on `keepalive`. On either, calls `conn.Close()` and then `onStop`. This is how a keepalive expiry propagates into a session tear-down.
**`RunForwarder(id, ctx, messages, inbox, workerProcessedCount, poolInboxCount)`** Reads from `messages` and writes to `inbox`. Increments both counters on each successful delivery. Identical in behavior to the inbound variant.
**`Session`** The coordination struct that ties the above blocks together for one connection lifecycle. `Session.Start` runs a loop: request a dial, wait for a connection, run `RunReader`, `RunHeartbeatForwarder`, and `RunStopMonitor` concurrently, wait for them to finish, emit `EventDisconnected`, sleep for `ReconnectDelay`, then repeat. `Session` is exported so it can be embedded or used directly in a custom worker. **`Session`** The coordination struct that ties the above blocks together for one connection lifecycle. `Session.Start` runs a loop: request a dial, wait for a connection, run `RunReader`, `RunHeartbeatForwarder`, and `RunStopMonitor` concurrently, wait for them to finish, emit `EventDisconnected`, sleep for `ReconnectDelay`, then repeat. `Session` is exported so it can be embedded or used directly in a custom worker.
### Replacement Patterns ### Replacement Patterns
@@ -139,14 +85,14 @@ The factory is set via `outbound.WithWorkerFactory` on the pool config.
**Replace the session loop.** Construct your own loop using `RunDialer`, `RunKeepalive`, and the session-level blocks. This gives you control over reconnection logic, back-off behavior, or multi-connection strategies while keeping the lower-level I/O blocks intact. **Replace the session loop.** Construct your own loop using `RunDialer`, `RunKeepalive`, and the session-level blocks. This gives you control over reconnection logic, back-off behavior, or multi-connection strategies while keeping the lower-level I/O blocks intact.
**Implement from scratch.** Satisfy the `Worker` interface directly. You are responsible for dialing, managing connection state, forwarding messages to `pool.Inbox`, emitting `EventConnected` and `EventDisconnected` to `pool.Events`, and incrementing `pool.InboxCounter`. **Implement from scratch.** Satisfy the `Worker` interface directly. You are responsible for dialing, managing connection state, forwarding messages to `pool.Inbox`, emitting `honeybee.EventConnected` and `honeybee.EventDisconnected` to `pool.Events`, and incrementing `pool.InboxCounter`.
## Factory Constraints ## Factory Constraints
Factories for both pool types are called while the pool holds its write lock. Two constraints follow from this directly. The factory is called while the pool holds its write lock. Two constraints follow from this directly.
**Factories must not block.** Any operation that could wait — dialing a connection, acquiring another lock, reading from a channel — will deadlock or stall the pool. All blocking work belongs inside `Start`, not inside the factory. **Factories must not block.** Any operation that could wait — dialing a connection, acquiring another lock, reading from a channel — will deadlock or stall the pool. All blocking work belongs inside `Start`, not inside the factory.
**Factories must not call pool methods.** `pool.Add`, `pool.Send`, `pool.Remove`, and similar methods all acquire the same lock the factory is called under. Calling them from the factory will deadlock. **Factories must not call pool methods.** `pool.Connect`, `pool.Send`, `pool.Remove`, and similar methods all acquire the same lock the factory is called under. Calling them from the factory will deadlock.
The factory's only job is to construct and return a worker. If construction itself can fail — for example, because a config value is invalid — return the error and the pool will propagate it to the caller of `Add` or `Connect`. The factory's only job is to construct and return a worker. If construction itself can fail — for example, because a config value is invalid — return the error and the pool will propagate it to the caller of `Connect`.
README.md
+159 -132
View File
@@ -5,38 +5,31 @@ WebSocket connection and pool primitives in Go. Built for Nostr.
## Library Map ## Library Map
```txt ```txt
transport/ single-connection primitives honeybee.go Pool, Worker, public types
connection.go *Connection, state machine, reader goroutine, pinger
transport/ single-connection primitives and helpers
connection.go Connection, state machine, reader goroutine, pinger
config.go ConnectionConfig, RetryConfig, options config.go ConnectionConfig, RetryConfig, options
retry.go exponential backoff with jitter retry.go exponential backoff with jitter
socket.go Dialer interface, AcquireSocket socket.go Dialer interface, AcquireSocket
url.go parsing and normalization url.go parsing and normalization
watchdog.go IdleWatchdog helper
inbound/ pool for peer-initiated connections
pool.go Pool, Peer, event plumbing
worker.go Worker interface, DefaultWorker, Run* functions
config.go WorkerConfig, PoolConfig, options
outbound/ pool for self-initiated connections
pool.go Pool, Peer, event plumbing
worker.go Worker interface, DefaultWorker, Session, Run* functions
config.go WorkerConfig, PoolConfig, options
logging/ structured log construction logging/ structured log construction
logging.go logger constructors, ForcedLevelHandler logging.go logger constructors, ForcedLevelHandler
types/ internal interfaces types/ shared interfaces (Dialer, Socket, ReceivedMessage)
honeybeetest/ test helpers and mocks for consumers honeybeetest/ test helpers and mocks for consumers
```` ```
## What This Library Does ## What This Library Does
Honeybee is a reliable and simple library for managing websocket connections and pools. Honeybee is a minimal, general-purpose WebSocket transport library.
- Handles websocket connections and pools cleanly and safely. - Client-Side: Manage a pool of outbound peer connections that reconnect automatically and surface lifecycle events.
- Provides two pools: one to manage outbound peers and another to manage inbound peers. - Server-Side: Wrap already-upgraded sockets in the connection primitive, which provides a ping-based heartbeat, automated read-loop, concurrent-safe writes, and classified disconnect errors.
- Exposes statistics at the connection, worker, and pool levels. - The same connection primitive may also be used directly on the client side when pool semantics are not needed, providing automated dialing retry with exponential backoff and jitter.
- Exposes a means to replace the internal pool worker to inject custom extensions. - Exposes a means to completely replace the internal pool worker to inject custom behavior.
## What This Library Does Not Do ## What This Library Does Not Do
@@ -45,7 +38,7 @@ Honeybee is a pure transport layer, but it is also a deliberately simple one. Ho
Honeybee does not provide: Honeybee does not provide:
- interpretation of message content. All messages are treated equally. - interpretation of message content. All messages are treated equally.
- message queuing, prioritization, batching, or coalescing. - message queuing, buffering, prioritization, batching, or coalescing.
- rate limiting, circuit breakers, token buckets, or adaptive throttling. - rate limiting, circuit breakers, token buckets, or adaptive throttling.
- broadcast, fanout, or any many-to-many message routing. - broadcast, fanout, or any many-to-many message routing.
- compression strategies, prepared message caching, or encoding optimization. - compression strategies, prepared message caching, or encoding optimization.
@@ -65,11 +58,146 @@ If the primary repository is unavailable, use the `replace` directive in your go
replace git.wisehodl.dev/jay/go-honeybee => github.com/wisehodl/go-honeybee latest replace git.wisehodl.dev/jay/go-honeybee => github.com/wisehodl/go-honeybee latest
``` ```
## Usage ## Outbound Pool
### Bare Connection The pool connects to peers by their URLs and keeps them connected. It reconnects automatically when a connection drops and proactively refreshes inactive connections.
Use `transport` directly when you need one socket and do not want pool semantics. ```go
import "git.wisehodl.dev/jay/go-honeybee"
pool, err := honeybee.NewPool(ctx, nil, handler)
if err != nil { /* handle error or panic */ }
defer pool.Close()
if err := pool.Connect("wss://peer.example.com"); err != nil { /* handle error */ }
go func() {
for msg := range pool.Inbox() {
// msg.ID is the normalized URL
// msg.Data is the payload
// msg.ReceivedAt is the timestamp
}
}()
go func() {
for ev := range pool.Events() {
// ev.At is the event timestamp
switch ev.Kind {
case honeybee.EventConnected:
case honeybee.EventDisconnected:
}
}
}()
pool.Send("wss://peer.example.com", []byte("hello"))
```
**Usage Notes:**
URLs are normalized by the pool. `wss://peer.example.com`, `wss://peer.example.com/`, and `WSS://Peer.Example.Com:443` all identify the same peer. `honeybee.NormalizeURL` is also available directly if you need to use the same URLs as keys elsewhere.
Every time a connection is established, `honeybee.EventConnected` is emitted. Every time a connection drops for any reason, `honeybee.EventDisconnected` is emitted. A peer that reconnects three times produces three Connected/Disconnected pairs.
Keepalive is configured via `honeybee.WithKeepaliveTimeout`. The worker records a heartbeat on every inbound message, every successful send, and every received pong. If no heartbeats arrive before the keepalive timer fires, the connection is proactively disconnected and reconnected. When set to zero, keepalive is disabled.
After a disconnect, the worker waits for `ReconnectDelay` before attempting the next connection. The default is 2 seconds. Set to zero in tests or when you need immediate reconnection.
`Send` returns `ErrConnectionUnavailable` during the gap between a disconnect and the next successful reconnect. Callers should wait for `EventConnected` before retrying and maintain their own write buffers if needed.
Dial failures are handled internally by the worker's retry logic and documented in structured logs. These do not stop the pool; it continues retrying according to the connection's retry config.
## Server-Side Usage
### Connection
Use `transport.NewConnectionFromSocket` when your HTTP upgrade handler gives you an open socket. The connection starts in `StateConnected`; do not call `Connect`.
```go
import "git.wisehodl.dev/jay/go-honeybee/transport"
// wsConn is a *websocket.Conn from your upgrade handler
conn, err := transport.NewConnectionFromSocket(wsConn, nil, logger)
if err != nil { /* handle error */ }
defer conn.Close()
go func() {
for data := range conn.Incoming() { // data: []byte
// process incoming messages
}
}()
go func() {
for err := range conn.Errors() {
// log or handle disconnects / read errors
}
}()
// Send can be called concurrently
conn.Send([]byte("hello"))
```
`Send` is safe for concurrent callers. `Close` is idempotent and safe to call from any goroutine.
When the reader exits, exactly one classified error reaches `Errors()` before the channel closes.
- `ErrPeerClosedClean` for normal closure
- `ErrPeerClosedUnexpected` for abnormal close codes
- `ErrReadError` for anything else
Pass an `*slog.Logger` as the third argument to get structured logs. Pass nil to disable logging entirely.
### IdleWatchdog
The watchdog helper detects clients that have gone silent. Wire activity signals from your `Incoming()` consumer, on each Send() call, and from the connection's `Heartbeat()` channel into it, and provide a callback to invoke when the timeout fires. Feeding all three sources means a client that neither sends or receives data but still responds to pings will not be considered idle. Since there is no reconnect loop on the server side, the typical callback is `conn.Close()`.
```go
import "time"
activity := make(chan struct{}, 1)
signal := func() {
select {
case activity <- struct{}{}:
default:
}
}
// Feed data messages:
go func() {
for data := range conn.Incoming() {
signal()
// process data
}
}()
// Feed sends:
func SendWithHeartbeat(conn *Connection, data []byte) error {
err := conn.Send(data)
if err != nil {
return err
}
signal()
return nil
}
// Feed pong heartbeats:
go func() {
for range conn.Heartbeat() {
signal()
}
}()
// Start the watchdog:
go transport.IdleWatchdog(ctx, activity, 30*time.Second, func() {
conn.Close()
})
```
When no activity signal arrives within the timeout, `onTimeout` is called once and the watchdog exits. When `ctx` is cancelled, the watchdog exits without calling `onTimeout`. When the timeout is zero or negative, the watchdog drains activity signals and waits for `ctx` to be cancelled without ever firing.
## Bare Connection
Use `transport.NewConnection` when you need a single outbound connection without pool semantics — for example, a one-shot query or a custom use-case. It adds several conveniences over a raw socket: a retry loop with exponential backoff, concurrent-safe writes, automatic write deadline enforcement, classified disconnect errors, and observable connection state.
```go ```go
import "git.wisehodl.dev/jay/go-honeybee/transport" import "git.wisehodl.dev/jay/go-honeybee/transport"
@@ -88,14 +216,15 @@ go func() {
go func() { go func() {
for err := range conn.Errors() { for err := range conn.Errors() {
// log or handle // log or handle disconnects / read errors
} }
}() }()
// Send can be called concurrently
conn.Send([]byte("hello")) conn.Send([]byte("hello"))
``` ```
The connection goes through four states: `StateDisconnected`, `StateConnecting`, `StateConnected`, `StateClosed`. Transitions are atomic and observable via `conn.State()`. Once closed, the connection should not be reused. Instead, construct a new one with the same URL and reconnect. The connection goes through four states: `StateDisconnected`, `StateConnecting`, `StateConnected`, `StateClosed`. Transitions are atomic and observable via `conn.State()`. Once closed, the connection should not be reused; construct a new one with the same URL and reconnect.
`Send` is safe for concurrent callers. `Close` is idempotent and safe to call from any goroutine. `Send` is safe for concurrent callers. `Close` is idempotent and safe to call from any goroutine.
@@ -105,118 +234,17 @@ When the reader exits, exactly one classified error reaches `Errors()` before th
- `ErrPeerClosedUnexpected` for abnormal close codes - `ErrPeerClosedUnexpected` for abnormal close codes
- `ErrReadError` for anything else - `ErrReadError` for anything else
Consumers use this to decide whether the disconnect was expected. No other errors are sent by the connection.
Pass an `*slog.Logger` as the third argument to get structured logs. Pass nil to disable logging entirely.
### Inbound Pool
The inbound pool manages connections initiated by peers. The consumer accepts a WebSocket somewhere else and hands the resulting socket to the pool along with an ID.
Each pool requires a non-empty string ID. This ID is attached to all structured log records emitted by the pool, its workers, and their connections.
```go
import "git.wisehodl.dev/jay/go-honeybee/inbound"
pool, err := inbound.NewPool(ctx, "my-pool", nil, handler)
if err != nil { /* handle error */ }
defer pool.Close()
// When a peer connects at the HTTP layer:
if err := pool.Add(peerID, socket); err != nil { /* handle error */ }
// Consume inbound data from all peers on one channel:
go func() {
for msg := range pool.Inbox() {
// msg.ID identifies the peer
// msg.Data is the payload
// msg.ReceivedAt is the timestamp
}
}()
// React to peer lifecycle:
go func() {
for ev := range pool.Events() {
// ev.At is the event timestamp
switch ev.Kind {
case inbound.EventDisconnected: // clean close
case inbound.EventDroppedClose: // peer closed with abnormal code
case inbound.EventDroppedError: // read error
case inbound.EventEvictedPolicy: // inactivity timeout
}
}
}()
pool.Send(peerID, []byte("response"))
```
`Add`, `Replace`, and `Remove` do not emit events. Events are emitted only when a worker exits on its own, either when the peer closed the socket or it was determined to be inactive.
Use `Replace` if you need to swap the socket for a peer while keeping its ID. No events are emitted during this process.
The watchdog is configured via `inbound.WithInactivityTimeout`. When set to zero, it is disabled. When set, the watchdog observes message traffic and disconnects the peer if no messages arrive within the configured duration. The watchdog is disabled by default, meaning connections persist until manually removed or remotely terminated.
### Outbound Pool
The outbound pool connects to peers by their URLs and keeps them connected. It reconnects automatically when a connection drops and proactively refreshes inactive connections.
Each pool requires a non-empty string ID. This ID is attached to all structured log records emitted by the pool, its workers, and their connections.
```go
import "git.wisehodl.dev/jay/go-honeybee/outbound"
pool, err := outbound.NewPool(ctx, "my-pool", nil, handler)
if err != nil { /* handle error */ }
defer pool.Close()
if err := pool.Connect("wss://peer.example.com"); err != nil { /* handle error */ }
go func() {
for msg := range pool.Inbox() {
// msg.ID is the normalized URL
// msg.Data is the payload
// msg.ReceivedAt is the timestamp
}
}()
go func() {
for ev := range pool.Events() {
// ev.At is the event timestamp
switch ev.Kind {
case outbound.EventConnected:
case outbound.EventDisconnected:
}
}
}()
pool.Send("wss://peer.example.com", []byte("hello"))
```
URLs are normalized by the pool. `wss://peer.example.com`, `wss://peer.example.com/`, and `WSS://Peer.Example.Com:443` all identify the same peer. `outbound.NormalizeURL` is also available directly if you need to use the same URLs as keys elsewhere.
Every time a connection is established, `outbound.EventConnected` is emitted. Every time a connection drops for any reason, `outbound.EventDisconnected` is emitted. A peer that reconnects three times produces three Connected/Disconnected pairs.
Keepalive is configured via `outbound.WithKeepaliveTimeout`. The worker records a heartbeat on every inbound message, every successful send, and every received pong. If no heartbeats arrive before the keepalive timer fires, the connection is proactively disconnected and reconnected. When set to zero, keepalive is disabled.
After a disconnect, the worker waits for `ReconnectDelay` before attempting the next connection. The default is 2 seconds. Set to zero in tests or when you need immediate reconnection.
`Send` returns `ErrConnectionUnavailable` during the gap between a disconnect and the next successful reconnect. Callers should wait for `OutboundEventConnected` before retrying and maintain their own write buffers if needed.
Dial failures are handled internally by the worker's retry logic and documented in structured logs. These do not stop the pool; it continues retrying according to the connection's retry config.
## Ping-Pong Heartbeats ## Ping-Pong Heartbeats
Connections send periodic WebSocket ping frames and listen for the corresponding pong replies. A received pong registers as a heartbeat signal within the worker. Connections send periodic WebSocket ping frames and listen for the corresponding pong replies. A received pong registers as a heartbeat signal within the worker.
For inbound workers, pong-derived heartbeats reset the inactivity watchdog timer alongside data messages. A peer that sends no data but responds to pings will not be evicted. Pong-derived heartbeats reset the keepalive timer alongside data messages and sends. A peer that sends no data but responds to pings will not be disconnected and reconnected by the keepalive mechanism.
For outbound workers, pong-derived heartbeats reset the keepalive timer. A peer that sends no data but responds to pings will not be disconnected and reconnected by the keepalive mechanism. The ping interval is configured via `transport.WithPingInterval` on the `transport.ConnectionConfig`. Import `git.wisehodl.dev/jay/go-honeybee/transport` to construct a `ConnectionConfig`, then pass it to the pool via `honeybee.WithConnectionConfig`, or supply it directly to `NewConnection` and `NewConnectionFromSocket`. The default is 20 seconds. Set to zero to disable pings entirely, in which case only data messages and outbound sends generate heartbeats.
The ping interval is configured via `transport.WithPingInterval` on the `transport.ConnectionConfig`. Import `git.wisehodl.dev/jay/go-honeybee/transport` to construct a `ConnectionConfig`, then pass it to the pool via `inbound.WithConnectionConfig` or `outbound.WithConnectionConfig`. The default is 20 seconds. Set to zero to disable pings entirely, in which case only data messages and outbound sends generate heartbeats.
## Statistics ## Statistics
Pools, workers, and connections expose counters and channel depths that can be sampled at any time. All values are snapshots; counters are monotonically increasing and are not reset between reconnects on an outbound worker. Pools, workers, and connections expose counters and channel depths that can be sampled at any time. All values are snapshots; counters are monotonically increasing and are not reset between reconnects.
```go ```go
// Pool-level snapshot // Pool-level snapshot
@@ -229,8 +257,7 @@ stats := pool.Stats()
// Single peer // Single peer
peerStats, err := pool.PeerStats(peerID) peerStats, err := pool.PeerStats(peerID)
// peerStats.Worker — queue depths, buffer depth, processed/dropped/sent counts // peerStats.Worker — channel depths, processed/sent counts
// peerStats.Connection — channel depths, receive/send/heartbeat counts (inbound)
// Bare connection (transport package) // Bare connection (transport package)
connStats := conn.Stats() // conn is a *transport.Connection connStats := conn.Stats() // conn is a *transport.Connection
@@ -241,7 +268,7 @@ connStats := conn.Stats() // conn is a *transport.Connection
The pool owns peer registration, event plumbing, and lifecycle. The worker owns what happens on the wire. The default worker can be replaced entirely or composed from the exported `Run*` building blocks that Honeybee provides. The pool owns peer registration, event plumbing, and lifecycle. The worker owns what happens on the wire. The default worker can be replaced entirely or composed from the exported `Run*` building blocks that Honeybee provides.
See EXTEND.md for the worker interface contract, the `PoolPlugin` fields, and the available building blocks for both inbound and outbound pools. See EXTEND.md for the worker interface contract, the `PoolPlugin` fields, and the available building blocks for the pool worker.
## Configuration ## Configuration