hub

Documentation

Index

• type Gate
  • func NewGate(ctx context.Context, l *slog.Logger) *Gate
  • func (g *Gate) Broadcast(streamMsg *common.StreamMessage)
  • func (g *Gate) Size() int
  • func (g *Gate) Subscribe(session HubSession, stream string, identifier string)
  • func (g *Gate) Unsubscribe(session HubSession, stream string, identifier string)
• type Hub
  • func NewHub(poolSize int, l *slog.Logger) *Hub
  • func (h *Hub) AddSession(session HubSession)
  • func (h *Hub) Broadcast(stream string, data string)
  • func (h *Hub) BroadcastMessage(msg *common.StreamMessage)
  • func (h *Hub) FindByIdentifier(id string) HubSession
  • func (h *Hub) RemoteDisconnect(msg *common.RemoteDisconnectMessage)
  • func (h *Hub) RemoveSession(session HubSession)
  • func (h *Hub) RemoveSessionLater(s HubSession)
  • func (h *Hub) Run()
  • func (h *Hub) Sessions() []HubSession
  • func (h *Hub) Shutdown()
  • func (h *Hub) Size() int
  • func (h *Hub) StreamsSize() int
  • func (h *Hub) SubscribeSession(session HubSession, stream string, identifier string)
  • func (h *Hub) UniqSize() int
  • func (h *Hub) UnsubscribeSession(session HubSession, stream string, identifier string)
  • func (h *Hub) UnsubscribeSessionFromChannel(session HubSession, targetIdentifier string)
• type HubRegistration
• type HubSession
• type HubSessionInfo
  • func NewHubSessionInfo(session HubSession) *HubSessionInfo
  • func (hs *HubSessionInfo) AddStream(stream string, identifier string)
  • func (hs *HubSessionInfo) RemoveStream(stream string, identifier string)

Types

type Gate

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

Gate plays the role of a shard for the hub. It keeps subscriptions for some streams (a particular shard) and is used to broadcast messages to all subscribers of these streams.

func NewGate

func NewGate(ctx context.Context, l *slog.Logger) *Gate

NewGate creates a new gate.

func (*Gate) Broadcast

func (g *Gate) Broadcast(streamMsg *common.StreamMessage)

Broadcast sends a message to all subscribers of the stream.

func (*Gate) Size

func (g *Gate) Size() int

Size returns a number of uniq streams

func (*Gate) Subscribe

func (g *Gate) Subscribe(session HubSession, stream string, identifier string)

Subscribe adds a session to the stream.

func (*Gate) Unsubscribe

func (g *Gate) Unsubscribe(session HubSession, stream string, identifier string)

Unsubscribe removes a session from the stream.

type Hub

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

Hub stores all the sessions and the corresponding subscriptions info

func NewHub

func NewHub(poolSize int, l *slog.Logger) *Hub

NewHub builds new hub instance

func (*Hub) AddSession

func (h *Hub) AddSession(session HubSession)

func (*Hub) Broadcast

func (h *Hub) Broadcast(stream string, data string)

Broadcast enqueues data broadcasting to a stream

func (*Hub) BroadcastMessage

func (h *Hub) BroadcastMessage(msg *common.StreamMessage)

BroadcastMessage enqueues broadcasting a pre-built StreamMessage

func (*Hub) FindByIdentifier

func (h *Hub) FindByIdentifier(id string) HubSession

func (*Hub) RemoteDisconnect

func (h *Hub) RemoteDisconnect(msg *common.RemoteDisconnectMessage)

RemoteDisconnect enqueues remote disconnect command

func (*Hub) RemoveSession

func (h *Hub) RemoveSession(session HubSession)

func (*Hub) RemoveSessionLater

func (h *Hub) RemoveSessionLater(s HubSession)

RemoveSession enqueues session un-registration

func (*Hub) Run

func (h *Hub) Run()

Run makes hub active

func (*Hub) Sessions

func (h *Hub) Sessions() []HubSession

func (*Hub) Shutdown

func (h *Hub) Shutdown()

Shutdown sends shutdown command to hub

func (*Hub) Size

func (h *Hub) Size() int

Size returns a number of active sessions

func (*Hub) StreamsSize

func (h *Hub) StreamsSize() int

StreamsSize returns a number of uniq streams

func (*Hub) SubscribeSession

func (h *Hub) SubscribeSession(session HubSession, stream string, identifier string)

func (*Hub) UniqSize

func (h *Hub) UniqSize() int

UniqSize returns a number of uniq identifiers

func (*Hub) UnsubscribeSession

func (h *Hub) UnsubscribeSession(session HubSession, stream string, identifier string)

func (*Hub) UnsubscribeSessionFromChannel

func (h *Hub) UnsubscribeSessionFromChannel(session HubSession, targetIdentifier string)

type HubRegistration

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

HubRegistration represents registration event ("add" or "remove")

type HubSession

type HubSession interface {
	GetID() string
	GetIdentifiers() string
	Send(msg encoders.EncodedMessage)
	DisconnectWithMessage(msg encoders.EncodedMessage, code string)
}

type HubSessionInfo

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

HubSessionInfo is used to track registered sessions

func NewHubSessionInfo

func NewHubSessionInfo(session HubSession) *HubSessionInfo

func (*HubSessionInfo) AddStream

func (hs *HubSessionInfo) AddStream(stream string, identifier string)

func (*HubSessionInfo) RemoveStream

func (hs *HubSessionInfo) RemoveStream(stream string, identifier string)