goob

A lightweight observability library for Crystal implementing non-blocking channel-based pub/sub with unbounded buffering.

Crystal's Channel requires a fixed buffer size. Choosing the right size is difficult — too small and publishers block; too large and memory is wasted. goob handles buffering dynamically, so slow subscribers never block publishers or other subscribers while preserving stable event order.

Features

• Unbounded buffer — publish as fast as you want without backpressure
• One-to-many — single publisher fans out to any number of subscribers
• Thread-safe — all operations are safe across fibers
• Non-blocking subscribers — a slow subscriber never blocks a fast one
• Stable event order — every subscriber sees events in the order they were published
• Cancellation — each subscriber gets an independent done channel; closing it removes that subscriber. Closing the hub's done channel removes all subscribers.

Example

require "goob"

done = Channel(Nil).new
ob = Goob.new(Int32, done)
events = ob.subscribe(Channel(Nil).new)

ob.publish(1)
ob.publish(2)
ob.publish(3)
done.close

while value = events.receive?
  print value
end
# => 123

API

# Create an observable hub for events of type T.
# Closing `done` cancels all current and future subscriptions.
Goob.new(T, done : Channel(Nil)) : Goob::Observable(T)

# Subscribe to events. Returns a Channel(T) that receives published events.
# The subscription is removed when either `sub_done` or the hub's `done` closes.
observable.subscribe(sub_done : Channel(Nil)) : Channel(T)

# Publish an event to all active subscribers. Never blocks.
observable.publish(event : T)

# Number of active subscribers.
observable.len : Int32
observable.size : Int32  # alias for len

# Create a standalone buffered pipe. Returns {write, events}.
# `write.call(event)` is non-blocking; `events` is the receive side.
Goob.new_pipe(T, done : Channel(Nil)) : {Proc(T, Nil), Channel(T)}

Installation

Add to shard.yml:

dependencies:
  goob:
    github: dsisnero/goob

Then shards install.

Documentation

• Architecture — internal design: Pipe buffering, Observable fan-out, and cancellation
• Development — local setup and quality gates
• Coding Guidelines — project conventions
• Testing — test strategy and spec coverage
• PR Workflow — pull request process