bus

Documentation

Overview

Package bus provides an in-memory event bus for DevTools consumers.

Index

• func RedactPayload(raw json.RawMessage) json.RawMessage
• type Bus
  • func New(capacity int) *Bus
  • func (b *Bus) AppendCause(parent, child uint64) bool
  • func (b *Bus) Publish(ev Event) uint64
  • func (b *Bus) Snapshot() []Event
  • func (b *Bus) Stats() Stats
  • func (b *Bus) Subscribe(ctx context.Context, fromSeq uint64) <-chan Event
  • func (b *Bus) SubscribeWithCloser(ctx context.Context, fromSeq uint64) (<-chan Event, func())
• type Event
  • func RedactEvent(ev Event) Event
• type Stats

Functions

func RedactPayload

func RedactPayload(raw json.RawMessage) json.RawMessage

RedactPayload redacts known secret fields and truncates large strings.

Types

type Bus

type Bus struct {
	// contains filtered or unexported fields
}

Bus is an in-process fan-out bus backed by a bounded ring buffer.

func New

func New(capacity int) *Bus

New constructs a Bus with the requested ring capacity.

func (*Bus) AppendCause

func (b *Bus) AppendCause(parent, child uint64) bool

AppendCause links parent as a cause of child when child is still in the ring.

func (*Bus) Publish

func (b *Bus) Publish(ev Event) uint64

Publish appends ev to the ring and fans out to active subscribers.

The fan-out sends happen after the subscriber map snapshot is taken and the bus mutex is released, so a slow or full subscriber channel cannot block other Publish calls or Subscribe/Unsubscribe operations.

func (*Bus) Snapshot

func (b *Bus) Snapshot() []Event

Snapshot returns all buffered events ordered from oldest to newest.

func (*Bus) Stats

func (b *Bus) Stats() Stats

Stats returns a point-in-time bus summary.

func (*Bus) Subscribe

func (b *Bus) Subscribe(ctx context.Context, fromSeq uint64) <-chan Event

Subscribe returns a stream of events newer than fromSeq, plus live updates.

The subscription is cleaned up when ctx is cancelled. Callers that cannot supply a cancellable context (e.g. background bookkeeping) should use SubscribeWithCloser to avoid leaking the cleanup goroutine and channel.

func (*Bus) SubscribeWithCloser

func (b *Bus) SubscribeWithCloser(ctx context.Context, fromSeq uint64) (<-chan Event, func())

SubscribeWithCloser is like Subscribe but also returns a closer function. Calling the closer (idempotent) unregisters the subscriber and stops the cleanup goroutine, even if ctx is never cancelled. Callers should call it when the subscription is no longer needed (e.g. when an SSE request handler returns).

type Event

type Event struct {
	Seq        uint64          `json:"seq"`
	At         int64           `json:"at"`
	Kind       string          `json:"kind"`
	Source     string          `json:"source,omitempty"`
	EntityKind string          `json:"entity_kind,omitempty"`
	EntityID   string          `json:"entity_id,omitempty"`
	ParentSeq  uint64          `json:"parent_seq,omitempty"`
	CausesSeq  []uint64        `json:"causes_seq,omitempty"`
	Payload    json.RawMessage `json:"payload,omitempty"`
	Err        string          `json:"err,omitempty"`
}

Event is the canonical DevTools event envelope.

func RedactEvent

func RedactEvent(ev Event) Event

RedactEvent returns a copy of ev with sensitive payload fields redacted.

type Stats

type Stats struct {
	Capacity    int    `json:"capacity"`
	Size        int    `json:"size"`
	OldestSeq   uint64 `json:"oldest_seq"`
	NewestSeq   uint64 `json:"newest_seq"`
	Published   uint64 `json:"published"`
	Evicted     uint64 `json:"evicted"`
	Dropped     uint64 `json:"dropped"`
	Subscribers int    `json:"subscribers"`
}

Stats reports current bus capacity and runtime counters.