events

package
v0.4.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: May 31, 2026 License: GPL-3.0 Imports: 17 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type MemoryBus 

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

MemoryBus is the default in-memory event bus. It dispatches events to matching subscribers via bounded per-subscriber channels, providing backpressure protection: slow subscribers have events dropped rather than blocking the publisher or consuming unbounded memory.

func NewMemoryBus 

func NewMemoryBus() *MemoryBus

NewMemoryBus creates an in-memory event bus.

func (*MemoryBus) CloseWAL added in v0.4.0 

func (b *MemoryBus) CloseWAL() error

CloseWAL flushes and closes the WAL. Safe to call on a bus without WAL.

func (*MemoryBus) DroppedEvents added in v0.4.0 

func (b *MemoryBus) DroppedEvents() int64

DroppedEvents returns the total number of events dropped across all subscribers due to full channels. Key metric for backpressure monitoring.

func (*MemoryBus) EnableWAL added in v0.4.0 

func (b *MemoryBus) EnableWAL(dir string) error

EnableWAL attaches a WAL writer to the event bus. All published events are persisted to the WAL before being dispatched to subscribers. The WAL must be closed separately via CloseWAL().

func (*MemoryBus) Publish 

func (b *MemoryBus) Publish(ctx context.Context, event contracts.Event) error

Publish dispatches an event to all matching subscribers. Each subscriber has a bounded channel; if the channel is full, the event is dropped for that subscriber and a counter is incremented.

func (*MemoryBus) Request 

func (b *MemoryBus) Request(ctx context.Context, event contracts.Event, timeout time.Duration) (contracts.Event, error)

Request publishes an event and waits for a reply.

func (*MemoryBus) SetAuditLogger added in v0.4.0 

func (b *MemoryBus) SetAuditLogger(a contracts.AuditLogger)

SetAuditLogger configures an audit logger for event bus operations.

func (*MemoryBus) SetNodeID added in v0.4.0 

func (b *MemoryBus) SetNodeID(id string)

SetNodeID sets the node identifier for audit entries.

func (*MemoryBus) SetPublishPolicy added in v0.4.0 

func (b *MemoryBus) SetPublishPolicy(p contracts.PublishPolicyProvider)

SetPublishPolicy configures a publish policy provider consulted before every Publish() call. When nil (default), all publishes are allowed.

func (*MemoryBus) SetSubscriberTimeout added in v0.4.0 

func (b *MemoryBus) SetSubscriberTimeout(moduleID string, timeout time.Duration)

SetSubscriberTimeout sets a per-module handler timeout override. Modules with tight latency requirements can request shorter timeouts. The timeout applies to handlers registered via SubscribeModule.

func (*MemoryBus) Subscribe 

func (b *MemoryBus) Subscribe(ctx context.Context, eventType string, handler contracts.EventHandler) error

Subscribe registers a handler for the given event type. Use "*" to subscribe to all events. The handler is invoked by a dedicated goroutine that drains events from a bounded channel. Slow handlers have events dropped rather than blocking publishers.

func (*MemoryBus) SubscribeFrom added in v0.4.0 

func (b *MemoryBus) SubscribeFrom(ctx context.Context, eventType string, handler contracts.EventHandler, sinceSeq uint64) error

SubscribeFrom subscribes to events of the given type and replays any events with sequence numbers >= sinceSeq before beginning live dispatch. sinceSeq=0 means "replay all available events."

func (*MemoryBus) SubscribeModule added in v0.4.0 

func (b *MemoryBus) SubscribeModule(ctx context.Context, moduleID, eventType string, handler contracts.EventHandler) error

SubscribeModule registers a handler tagged with a module identifier.

func (*MemoryBus) SubscriberCount added in v0.4.0 

func (b *MemoryBus) SubscriberCount() int

SubscriberCount returns the total number of active subscriber channels. A healthy bus always returns >= 0. Used by health probes.

func (*MemoryBus) SubscriberStats added in v0.4.0 

func (b *MemoryBus) SubscriberStats() []SubscriberStat

SubscriberStats returns per-subscriber metrics for monitoring.

func (*MemoryBus) SubscriptionStats added in v0.4.0 

func (b *MemoryBus) SubscriptionStats() []SubscriptionStat

SubscriptionStats returns a summary of active subscriptions grouped by event type. Used by the admin API for runtime introspection.

func (*MemoryBus) Unsubscribe 

func (b *MemoryBus) Unsubscribe(ctx context.Context, eventType string, handler contracts.EventHandler) error

Unsubscribe removes a handler for the given event type.

func (*MemoryBus) UnsubscribeAll added in v0.4.0 

func (b *MemoryBus) UnsubscribeAll(ctx context.Context, moduleID string) error

UnsubscribeAll removes all subscriptions tagged with the given module ID.

func (*MemoryBus) UpdateMinSubscriberSeq added in v0.4.0 

func (b *MemoryBus) UpdateMinSubscriberSeq(seq uint64)

UpdateMinSubscriberSeq updates the WAL's minimum subscriber sequence for pruning purposes. Call this periodically or when subscribers advance.

type SubscriberStat added in v0.4.0 

type SubscriberStat struct {
	EventType    string `json:"event_type"`
	ModuleID     string `json:"module_id,omitempty"`
	Processed    int64  `json:"processed"`
	Dropped      int64  `json:"dropped"`
	AvgLatencyUs int64  `json:"avg_latency_us"`
}

SubscriberStat provides per-subscriber backpressure and latency metrics.

type SubscriptionStat added in v0.4.0 

type SubscriptionStat struct {
	EventType       string `json:"event_type"`
	SubscriberCount int    `json:"subscriber_count"`
}

SubscriptionStat describes subscriptions for a given event type.

type WALSegment added in v0.4.0 

type WALSegment struct {
	Path      string
	FirstSeq  uint64
	LastSeq   uint64
	Size      int64
	CreatedAt time.Time
}

WALSegment is a single WAL journal file on disk.

type WALWriter added in v0.4.0 

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

WALWriter provides a write-ahead log for the event bus. Events are appended as JSONL with a monotonically increasing sequence number. Segments are rotated when they exceed walSegmentSize. Old segments are pruned when all active subscribers have passed their sequence numbers.

func NewWALWriter added in v0.4.0 

func NewWALWriter(dir string) (*WALWriter, error)

NewWALWriter opens or creates a WAL in the given directory. If the directory doesn't exist, it is created.

func (*WALWriter) Close added in v0.4.0 

func (w *WALWriter) Close() error

Close flushes and closes the WAL.

func (*WALWriter) Flush added in v0.4.0 

func (w *WALWriter) Flush() error

Flush ensures all buffered writes are persisted to disk.

func (*WALWriter) LastSeq added in v0.4.0 

func (w *WALWriter) LastSeq() uint64

LastSeq returns the highest sequence number written.

func (*WALWriter) ReplayFrom added in v0.4.0 

func (w *WALWriter) ReplayFrom(ctx context.Context, sinceSeq uint64, fn func(contracts.Event) error) error

ReplayFrom reads events from the WAL starting at the given sequence number and invokes the callback for each event. If sinceSeq is 0, replays from the beginning. The callback is called synchronously; it should not block.

func (*WALWriter) Segments added in v0.4.0 

func (w *WALWriter) Segments() []WALSegment

Segments returns metadata for all WAL segments.

func (*WALWriter) SetMinSubscriberSeq added in v0.4.0 

func (w *WALWriter) SetMinSubscriberSeq(seq uint64)

SetMinSubscriberSeq informs the WAL of the lowest subscriber sequence number. Segments with LastSeq < minSeq can be pruned.

func (*WALWriter) Write added in v0.4.0 

func (w *WALWriter) Write(event contracts.Event) (uint64, error)

Write appends an event to the WAL. Thread-safe. Returns the assigned sequence number.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.