event

package
v1.30.0 Latest Latest
Warning

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

 Go to latest Published: Sep 16, 2026 License: MIT Imports: 34 Imported by: 0

Documentation

Index

Constants

View Source 
const ChatEventsNamespace = "chat-events"

ChatEventsNamespace is the cluster subscription namespace for per-chat event topics, keyed by the raw chat ID bytes. A topic's subscribers are the servers currently hosting at least one open stream belonging to a member of that chat, so a chat publisher resolves the hosting servers in one lookup instead of one per member. Today only group chats register here — DM delivery stays user-keyed — but nothing about the topic shape is group-specific.

View Source 
const UserEventsNamespace = "user-events"

UserEventsNamespace is the cluster subscription namespace for per-user event stream topics, keyed by the raw user ID bytes. A topic's subscribers are the servers currently hosting at least one of that user's open streams.

Variables

View Source 
var ErrStreamLagging = errors.New("stream lagging; closed")

ErrStreamLagging is returned by Notify when the stream's buffer is full. The stream is closed in the same call: its handler returns, the client reconnects and catches up with a delta sync, and no other stream on the server is delayed on its account.

Functions

This section is empty.

Types

type Bus 

type Bus[Key, Event any] struct {
	// contains filtered or unexported fields
}

func NewBus 

func NewBus[Key, Event any]() *Bus[Key, Event]

func (*Bus[Key, Event]) AddHandler 

func (b *Bus[Key, Event]) AddHandler(h Handler[Key, Event])

func (*Bus[Key, Event]) OnEvent 

func (b *Bus[Key, Event]) OnEvent(key Key, e Event)

type EventStream added in v1.30.0 

type EventStream[E any] struct {
	// contains filtered or unexported fields
}

EventStream is the bounded, non-blocking queue between the delivery path and a stream's handler goroutine. Events are enqueued individually and shared by pointer across every stream they are delivered to, so they must be treated as immutable once published; the handler drains whatever has accumulated into one batch per send (see drainReady).

func NewEventStream added in v1.30.0 

func NewEventStream[E any](id string, bufferSize int) *EventStream[E]

func (*EventStream[E]) Channel added in v1.30.0 

func (s *EventStream[E]) Channel() <-chan E

func (*EventStream[E]) Close added in v1.30.0 

func (s *EventStream[E]) Close()

func (*EventStream[E]) ID added in v1.30.0 

func (s *EventStream[E]) ID() string

func (*EventStream[E]) Notify added in v1.30.0 

func (s *EventStream[E]) Notify(event E) error

type Forwarder 

type Forwarder interface {
	ForwardUserEvents(ctx context.Context, events ...*eventpb.UserEvent) error
	ForwardChatEvents(ctx context.Context, events ...*eventpb.ChatEvent) error
}

func NewForwardingClient 

func NewForwardingClient(log *zap.Logger, subscriptions *cluster.Subscriptions, currentRpcApiKey string) Forwarder

type ForwardingClient 

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

ForwardingClient forwards events to the servers hosting their subscribed streams, for processes (or code paths) that host no event streams of their own.

func (ForwardingClient) ForwardChatEvents added in v1.29.0 

func (f ForwardingClient) ForwardChatEvents(ctx context.Context, events ...*eventpb.ChatEvent) error

func (ForwardingClient) ForwardUserEvents 

func (f ForwardingClient) ForwardUserEvents(ctx context.Context, events ...*eventpb.UserEvent) error

type Handler 

type Handler[Key, Event any] interface {
	OnEvent(key Key, e Event)
}

type HandlerFunc 

type HandlerFunc[Key, Event any] func(Key, Event)

HandlerFunc is an adapter to allow the use of ordinary functions as Handlers.

func (HandlerFunc[Key, Event]) OnEvent 

func (f HandlerFunc[Key, Event]) OnEvent(key Key, e Event)

OnEvent calls f(key, e).

type KeyAndEvent 

type KeyAndEvent[Key, Event any] struct {
	Key   Key
	Event Event
}

type Server 

type Server struct {
	eventpb.UnimplementedEventStreamingServer
	// contains filtered or unexported fields
}

func NewServer 

func NewServer(
	log *zap.Logger,
	authz auth.Authorizer,
	accounts account.Store,
	badges badge.Store,
	chats chat.Store,
	subscriptions *cluster.Subscriptions,
	userEventBus *Bus[*commonpb.UserId, *eventpb.Event],
	chatEventBus *Bus[*commonpb.ChatId, *eventpb.ChatEvent],
	staleEventDetectorCtors []StaleEventDetectorCtor[*eventpb.Event],
	currentRpcApiKey string,
	opts ...ServerOption,
) *Server

NewServer constructs the event streaming server and starts its membership reconcile sweep (see reconcileMembership) in the background. The sweep runs until Shutdown, which every caller must eventually invoke — a Server that is constructed and dropped without it leaks the sweep's goroutine.

func (*Server) ForwardChatEvents added in v1.29.0 

func (s *Server) ForwardChatEvents(ctx context.Context, events ...*eventpb.ChatEvent) error

func (*Server) ForwardEvents 

func (s *Server) ForwardEvents(ctx context.Context, req *eventpb.ForwardEventsRequest) (*eventpb.ForwardEventsResponse, error)

ForwardEvents is the internal RPC receiving events forwarded by the server that observed them. Delivery here is local-only: the sender already resolved this server as a subscriber, and re-resolving would at best repeat its work and at worst bounce an event between servers holding mutually stale caches. An event arriving for a user with no local streams (the row outlived the last stream by a cache window) is dropped; the client's delta sync is the backstop.

func (*Server) ForwardUserEvents 

func (s *Server) ForwardUserEvents(ctx context.Context, events ...*eventpb.UserEvent) error

func (*Server) OnChatEvent added in v1.29.0 

func (s *Server) OnChatEvent(_ *commonpb.ChatId, e *eventpb.ChatEvent)

OnChatEvent is the chat bus handler; the payload carries the chat ID and exclusions itself, so the bus key rides along only for the bus's shape.

func (*Server) OnEvent 

func (s *Server) OnEvent(userID *commonpb.UserId, e *eventpb.Event)

func (*Server) Shutdown added in v1.29.0 

func (s *Server) Shutdown()

Shutdown closes every open client stream — each StreamEvents handler returns and cleans up its subscription — and refuses new ones. Call it on shutdown after Subscriptions.Drain and before the gRPC server's GracefulStop: the streams are held open indefinitely by connected clients, so a GracefulStop without this never returns. Closed clients reconnect to a healthy server and delta sync. Idempotent. It also stops the membership reconcile's sweep; a reconcile in flight finishes against sessions that refuse it (see streamSession.close).

func (*Server) StreamEvents 

func (s *Server) StreamEvents(stream grpc.BidiStreamingServer[eventpb.StreamEventsRequest, eventpb.StreamEventsResponse]) error

type ServerOption added in v1.30.0 

type ServerOption func(*Server)

ServerOption tunes a Server at construction.

func WithMembershipReconcile added in v1.30.0 

func WithMembershipReconcile(interval, tick time.Duration) ServerOption

WithMembershipReconcile overrides the membership reconcile's schedule (see reconcileMembership): how often each user's memberships are re-read, and the sweep's tick. Tests use it to run the reconcile fast; production keeps the defaults.

type StaleEventDetector 

type StaleEventDetector[Event any] interface {
	ShouldDrop(event Event) bool
}

type StaleEventDetectorCtor 

type StaleEventDetectorCtor[Event any] func() StaleEventDetector[Event]

type Stream 

type Stream[E any] interface {
	ID() string

	// Notify enqueues an event for the stream's handler without blocking: a
	// full buffer closes the stream and returns ErrStreamLagging rather than
	// waiting on it. Delivery to a topic with many streams therefore costs one
	// channel operation per stream, however slow any one client is.
	Notify(event E) error

	Close()
}

Stream is one open client stream as the delivery path sees it.

type TestEventObserver 

type TestEventObserver[Key, Event any] struct {
	// contains filtered or unexported fields
}

func NewTestEventObserver 

func NewTestEventObserver[Key, Event any]() *TestEventObserver[Key, Event]

func (*TestEventObserver[Key, Event]) GetEvents 

func (h *TestEventObserver[Key, Event]) GetEvents(filter func(Key) bool) []*KeyAndEvent[Key, Event]

func (*TestEventObserver[Key, Event]) OnEvent 

func (h *TestEventObserver[Key, Event]) OnEvent(key Key, event Event)

func (*TestEventObserver[Key, Event]) Reset 

func (h *TestEventObserver[Key, Event]) Reset()

func (*TestEventObserver[Key, Event]) WaitFor 

func (h *TestEventObserver[Key, Event]) WaitFor(t *testing.T, condition func([]*KeyAndEvent[Key, Event]) bool)

func (*TestEventObserver[Key, Event]) WaitForWithTimeout 

func (h *TestEventObserver[Key, Event]) WaitForWithTimeout(t *testing.T, timeout time.Duration, condition func([]*KeyAndEvent[Key, Event]) bool)

Source Files

View all Source files

Directories

Path Synopsis

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.