eventbroker

Documentation

Index

• Constants
• Variables
• type ReplayEntry
• type Subscriber
  • func (s *Subscriber) MarkClosed()
  • func (s *Subscriber) Send(evt apitypes.WorkspaceSSEEvent)
• type UserEventBroker
  • func NewUserEventBroker() *UserEventBroker
  • func (b *UserEventBroker) CleanupUser(userID string)
  • func (b *UserEventBroker) CleanupWorkspace(workspaceID string)
  • func (b *UserEventBroker) PublishToUser(userID string, evt apitypes.WorkspaceSSEEvent)
  • func (b *UserEventBroker) PublishToWorkspace(workspaceID string, evt apitypes.WorkspaceSSEEvent)
  • func (b *UserEventBroker) RecordWorkspaceOwner(workspaceID, userID string)
  • func (b *UserEventBroker) Replay(userID string, lastID uint64) ([]ReplayEntry, bool)
  • func (b *UserEventBroker) SubscribeUser(userID string) (*Subscriber, error)
  • func (b *UserEventBroker) SubscribeWorkspace(workspaceID string) (*Subscriber, error)
  • func (b *UserEventBroker) UnsubscribeUser(userID string, s *Subscriber)
  • func (b *UserEventBroker) UnsubscribeWorkspace(workspaceID string, s *Subscriber)
  • func (b *UserEventBroker) UserSubscriberCount(userID string) int
  • func (b *UserEventBroker) WorkspaceOwner(workspaceID string) string
  • func (b *UserEventBroker) WorkspaceSubscriberCount(workspaceID string) int

Constants

const (
	BrokerChannelBuffer = 16

	MaxSubscribersPerUser = 20
	HeartbeatSentinelType = "_heartbeat"
)

Variables

var ErrTooManySubscribers = &apierrors.APIError{
	Type:    apierrors.ErrorTypeRateLimit,
	Code:    "too_many_subscribers",
	Message: "too many active SSE subscribers for user",
}

ErrTooManySubscribers is returned when SubscribeUser hits MaxSubscribersPerUser. It is a *apierrors.APIError (RateLimited/429) so the centralized HTTP error handler maps it automatically. Callers can still use errors.Is (backwards compat) and errors.As (typed path).

Types

type ReplayEntry

type ReplayEntry struct {
	ID    uint64
	Event apitypes.WorkspaceSSEEvent
}

type Subscriber

type Subscriber struct {
	Ch chan apitypes.WorkspaceSSEEvent
	// contains filtered or unexported fields
}

func (*Subscriber) MarkClosed

func (s *Subscriber) MarkClosed()

func (*Subscriber) Send

func (s *Subscriber) Send(evt apitypes.WorkspaceSSEEvent)

type UserEventBroker

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

func NewUserEventBroker

func NewUserEventBroker() *UserEventBroker

func (*UserEventBroker) CleanupUser

func (b *UserEventBroker) CleanupUser(userID string)

func (*UserEventBroker) CleanupWorkspace

func (b *UserEventBroker) CleanupWorkspace(workspaceID string)

func (*UserEventBroker) PublishToUser

func (b *UserEventBroker) PublishToUser(userID string, evt apitypes.WorkspaceSSEEvent)

func (*UserEventBroker) PublishToWorkspace

func (b *UserEventBroker) PublishToWorkspace(workspaceID string, evt apitypes.WorkspaceSSEEvent)

func (*UserEventBroker) RecordWorkspaceOwner

func (b *UserEventBroker) RecordWorkspaceOwner(workspaceID, userID string)

func (*UserEventBroker) Replay

func (b *UserEventBroker) Replay(userID string, lastID uint64) ([]ReplayEntry, bool)

func (*UserEventBroker) SubscribeUser

func (b *UserEventBroker) SubscribeUser(userID string) (*Subscriber, error)

func (*UserEventBroker) SubscribeWorkspace

func (b *UserEventBroker) SubscribeWorkspace(workspaceID string) (*Subscriber, error)

func (*UserEventBroker) UnsubscribeUser

func (b *UserEventBroker) UnsubscribeUser(userID string, s *Subscriber)

func (*UserEventBroker) UnsubscribeWorkspace

func (b *UserEventBroker) UnsubscribeWorkspace(workspaceID string, s *Subscriber)

func (*UserEventBroker) UserSubscriberCount

func (b *UserEventBroker) UserSubscriberCount(userID string) int

func (*UserEventBroker) WorkspaceOwner

func (b *UserEventBroker) WorkspaceOwner(workspaceID string) string

func (*UserEventBroker) WorkspaceSubscriberCount

func (b *UserEventBroker) WorkspaceSubscriberCount(workspaceID string) int