Documentation
¶
Index ¶
- Constants
- Variables
- func SubscribePrefixTyped[M proto.Message](bus *EventBus[*anypb.Any], scope, namespace, topicPrefix string, ...) error
- func UnmarshalAnyPayload(e Event[*anypb.Any], target proto.Message) error
- func ValidateEvent[T any](e Event[T]) error
- type Event
- type EventBus
- func (b *EventBus[T]) Close() error
- func (b *EventBus[T]) Publish(e Event[T]) error
- func (b *EventBus[T]) Subscribe(scope, namespace, topic string, fn Handler[T]) error
- func (bus *EventBus[T]) SubscribeCorrelated(scope, namespace, topic, corrID string, handler Handler[T]) error
- func (bus *EventBus[T]) SubscribeOnce(scope, namespace, topic string, handler Handler[T]) errordeprecated
- func (b *EventBus[T]) SubscribePrefix(scope, namespace, topicPrefix string, fn Handler[T]) error
- func (bus *EventBus[T]) SubscribePrefixWithContext(ctx context.Context, scope, namespace, topicPrefix string, handler Handler[T]) error
- func (bus *EventBus[T]) Unsubscribe(scope, namespace, topic string, target Handler[T])
- func (bus *EventBus[T]) UnsubscribePrefix(scope, namespace, topicPrefix string, target Handler[T])
- func (bus *EventBus[T]) WaitForTopic(ctx context.Context, scope, namespace, topic string, timeout time.Duration, ...) error
- type Handler
- type LocalTransport
- type Options
- type Transport
Constants ¶
const ( // TopicAgentNetworkRunning announces that the network agent reached its // running state; the supervisors' boot phase gate waits on it. Payload: // reserved (no producer publishes a typed payload yet; the gate only // observes arrival). TopicAgentNetworkRunning = "agent.network.running" // TopicPrefixAgentLifecycle is the prefix under which every lifecycle // topic below lives; prefix subscribers use it with SubscribePrefix // (mind the type hazard documented there - sibling topics carry // different payload types). TopicPrefixAgentLifecycle = "agent/lifecycle" // TopicAgentLifecycleAction is the operator/client -> supervisor // command channel. Payload: LifecycleControl. Distinct from // TopicAgentLifecycleControl despite the confusable name: this one // carries requests to act. TopicAgentLifecycleAction = "agent/lifecycle.action" // TopicAgentLifecycleControl carries the lifecycle controller's own // announcements of actions it is applying. Payload: LifecycleControl. TopicAgentLifecycleControl = "agent/lifecycle.control" // TopicAgentLifecycleStatus carries agent state reports. Payload: // LifecycleStatus (run_id is the structural restart-detection key). TopicAgentLifecycleStatus = "agent/lifecycle.status" // TopicAgentLifecycleTransition carries state-machine transition // events. Payload: LifecycleTransition. TopicAgentLifecycleTransition = "agent/lifecycle.transition" // TopicAgentHeartbeat carries an agent's liveness frame. Payload: // Heartbeat - NOT LifecycleStatus. The two answer different // questions: a heartbeat asserts the agent is ALIVE, a status asserts // a TRANSITION. Synthesizing a LifecycleStatus{State:"RUNNING"} for // each heartbeat is the conflation the status proto's own comment // condemns, and it is what the two runners did against a bare string // literal. // // NOTHING SUBSCRIBES YET. Named here rather than left inline because // a topic no consumer can find by name is one no consumer will // subscribe to; the missing consumer belongs in the ledger, not in a // string. TopicAgentHeartbeat = "agent/heartbeat" // TopicSystemShutdown is the operator/client -> supervisor system // shutdown request. Payload: wrapperspb.StringValue holding one of // "poweroff", "reboot", "halt"; decode fails closed on anything // else. In PID-1 mode the supervisor answers with the full // teardown (StopAll -> sync -> umount -> reboot(2)); otherwise it // stops the runtime and exits. TopicSystemShutdown = "system.shutdown" )
Well-known cross-boundary topics. One exported constant per topic; the orchestrator imports these rather than repeating string literals (GAPI-DIV-010; unblocks GOBLIN-DIV-011). Purely in-package topics may stay literals - only cross-boundary topics are promoted here.
Variables ¶
var ErrNoPeer = errors.New("eventbus: no peer connected")
ErrNoPeer reports that a remote publish had nowhere to go: the transport is attached and healthy, and no peer is connected to it.
It is part of the Transport CONTRACT rather than any one transport's internals, which is why it lives beside the interface. It is also why it cannot live in core/transport: that package imports this one.
A single node with no peers is the normal case, not a failure, and the bus treats it as such. The sentinel is returned rather than a bare nil so that a caller which DOES care - a cluster member that expects peers - keeps the distinction a nil would discard.
var ErrWaitTimeout = errors.New("eventbus: wait for topic timed out")
ErrWaitTimeout is returned when WaitForTopic's bound expires before the topic fires. Callers at a boot phase gate treat it as a loud failure: "hang forever" is not an acceptable PID 1 outcome (review R13).
Functions ¶
func SubscribePrefixTyped ¶
func SubscribePrefixTyped[M proto.Message]( bus *EventBus[*anypb.Any], scope, namespace, topicPrefix string, handler func(e Event[*anypb.Any], msg M), ) error
SubscribePrefixTyped subscribes to a topic prefix but invokes handler only for events whose *anypb.Any payload unmarshals into type M. It is the type-safe companion to SubscribePrefix: events carrying a different proto type (a sibling topic under the same prefix), or a nil/undecodable payload, are silently skipped instead of crashing a type-specific handler.
func UnmarshalAnyPayload ¶
UnmarshalAnyPayload extracts and unmarshals a protobuf Any payload from an event. Call it as: eventbus.UnmarshalAnyPayload(e, &myProto)
func ValidateEvent ¶
Types ¶
type Event ¶
type Event[T any] struct { ID string Scope string Namespace string Topic string Payload T Source string Tags []string }
func NewEvent ¶
NewEvent no longer takes a broadcast flag (GAPI-DIV-106). It selected between two Transport methods that did the same thing, because the QUIC transport addressed a single peer: "broadcast" and "publish to the one peer you have" are indistinguishable. Now that the transport holds a peer SET, a remote publish IS to every peer, so there is no second behaviour left for a flag to choose - and the flag was never carried on the wire, so no receiver could ever act on it.
type EventBus ¶
type EventBus[T any] struct { // contains filtered or unexported fields }
func NewInprocBus ¶
func (*EventBus[T]) SubscribeCorrelated ¶
func (bus *EventBus[T]) SubscribeCorrelated(scope, namespace, topic, corrID string, handler Handler[T]) error
SubscribeCorrelated calls handler at most once, for the first event on the exact topic whose ID equals corrID, then unsubscribes. Events with any other ID are ignored (not consumed). This is the request/reply correlation primitive: a caller publishes a request, then waits for the reply whose ID the responder echoed from the request - so concurrent callers sharing one reply topic don't steal each other's responses. The Envelope carries the ID over the wire, so this works across transports.
func (*EventBus[T]) SubscribeOnce
deprecated
SubscribeOnce calls handler at most once for the exact topic, then unsubscribes.
Deprecated: uncorrelated one-shot subscriptions let concurrent callers steal each other's events (review R15). Use SubscribeCorrelated for request/reply, or WaitForTopic for bounded phase gates. Scheduled for removal; see deprecation.jsonl.
func (*EventBus[T]) SubscribePrefix ¶
SubscribePrefix subscribes to every topic that begins with topicPrefix.
TYPE HAZARD: a prefix matches sibling topics that may carry different payload types - e.g. SubscribePrefix("system","",TopicPrefixAgentLifecycle) fires for both TopicAgentLifecycleAction (LifecycleControl) and TopicAgentLifecycleStatus (LifecycleStatus). A handler that assumes one proto type will panic on UnmarshalTo/type-assert when the other arrives. Prefer an exact Subscribe when the handler is payload-typed, or use SubscribePrefixTyped to filter by type.
func (*EventBus[T]) SubscribePrefixWithContext ¶
func (bus *EventBus[T]) SubscribePrefixWithContext(ctx context.Context, scope, namespace, topicPrefix string, handler Handler[T]) error
SubscribePrefixWithContext subscribes with automatic cleanup on context cancellation. This prevents subscription leaks in long-running operations by removing the subscription when the context is cancelled or times out.
func (*EventBus[T]) Unsubscribe ¶
Unsubscribe removes a single handler from an exact topic.
LIMITATION: func values are only comparable by code pointer, which every closure instance from one code site shares - so this can only distinguish handlers defined at different code sites. Do not use it to remove one of several same-callsite subscriptions; the one-shot APIs (SubscribeOnce, SubscribeCorrelated, WaitForTopic, SubscribePrefixWithContext) remove themselves by unique id instead.
func (*EventBus[T]) UnsubscribePrefix ¶
UnsubscribePrefix removes a single handler from a prefix subscription. Shares Unsubscribe's code-pointer limitation.
func (*EventBus[T]) WaitForTopic ¶
func (bus *EventBus[T]) WaitForTopic(ctx context.Context, scope, namespace, topic string, timeout time.Duration, clk clock.Clock) error
WaitForTopic blocks until an event arrives on the exact topic, the context is cancelled, or the timeout expires - whichever comes first. It is the bounded phase-gate primitive: unlike a bare subscription, it can never wait forever. The clock is injectable for deterministic tests; nil uses the real clock.
type LocalTransport ¶
type LocalTransport[T any] struct{}
LocalTransport is a no-op transport for local-only operation.
func (*LocalTransport[T]) Close ¶
func (t *LocalTransport[T]) Close() error
func (*LocalTransport[T]) OnRemoteEvent ¶
func (t *LocalTransport[T]) OnRemoteEvent(func(Event[T]))
func (*LocalTransport[T]) PublishRemote ¶
func (t *LocalTransport[T]) PublishRemote(ctx context.Context, e Event[T]) error
Source Files