event

package
v1.29.0 Latest Latest
Warning

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

 Go to latest Published: Sep 10, 2026 License: MIT Imports: 30 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

This section is empty.

Functions

func EventIDString 

func EventIDString(id *eventpb.EventId) string

func MustGenerateEventID 

func MustGenerateEventID() *eventpb.EventId

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 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 ProtoEventStream 

type ProtoEventStream[E any, P proto.Message] struct {
	sync.Mutex
	// contains filtered or unexported fields
}

func NewProtoEventStream 

func NewProtoEventStream[E any, P proto.Message](
	id string,
	bufferSize int,
	selector func(event E) (P, bool),
) *ProtoEventStream[E, P]

func (*ProtoEventStream[E, P]) Channel 

func (s *ProtoEventStream[E, P]) Channel() <-chan P

func (*ProtoEventStream[E, P]) Close 

func (s *ProtoEventStream[E, P]) Close()

func (*ProtoEventStream[E, P]) ID 

func (s *ProtoEventStream[E, P]) ID() string

func (*ProtoEventStream[E, P]) Notify 

func (s *ProtoEventStream[E, P]) Notify(event E, timeout time.Duration) error

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,
) *Server

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

todo: utilize batching by receiver to optimize internal forwarding RPC calls

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.

func (*Server) StreamEvents 

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

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(event E, timeout time.Duration) error
	Close()
}

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.