core

Documentation

Index

• func UnmarshalMessageIn(data []byte) (*messageIn, error)
• type Cache
  • func NewCache(col collector) *Cache
  • func (c *Cache) ConnectionByUserID(_ context.Context, ch channel.Channel, userID string) common.ConnectionWrapper
  • func (c *Cache) Connections(_ context.Context, ch channel.Channel) []common.ConnectionWrapper
  • func (c *Cache) Remove(_ context.Context, connID string, userID *string)
  • func (c *Cache) Subscribe(_ context.Context, conn common.ConnectionWrapper, channels ...channel.Channel)
  • func (c *Cache) Unsubscribe(_ context.Context, connID string, channels ...channel.Channel)
  • func (c *Cache) UnsubscribeUserID(_ context.Context, connID string, userID string, ch channel.Channel)
• type Dispatch
  • func NewDispatch(store store, logger log.Logger) *Dispatch
  • func (d *Dispatch) SendPrivateMessage(ctx context.Context, ch channel.Channel, userID string, message interface{}) error
  • func (d *Dispatch) SendPublicMessage(ctx context.Context, ch channel.Channel, message interface{}) error
• type MessageOut
• type MessageType
  • func (m MessageType) String() string

Functions

func UnmarshalMessageIn

func UnmarshalMessageIn(data []byte) (*messageIn, error)

UnmarshalMessageIn deserializes the input slice of bytes that has been read from the websocket connection.

Types

type Cache

type Cache struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

Cache is an in-memory storage to store available channels and connections.

func NewCache

func NewCache(col collector) *Cache

NewCache creates a new instance on Cache.

func (*Cache) ConnectionByUserID

func (c *Cache) ConnectionByUserID(_ context.Context, ch channel.Channel, userID string) common.ConnectionWrapper

ConnectionByUserID return the connection if there is a connection for the input userID, and the that connection already subscribed to the input channel. Otherwise, returns nil.

func (*Cache) Connections

func (c *Cache) Connections(_ context.Context, ch channel.Channel) []common.ConnectionWrapper

Connections returns a list of connections that already subscribed to the input channel.

This function is thread-safe and multiple goroutines can get the list of subscribed connection concurrently.

func (*Cache) Remove

func (c *Cache) Remove(_ context.Context, connID string, userID *string)

Remove removes all subscriptions of the input connection id. Removing all subscription means removing connection from the storage.

This function is thread-safe and multiple goroutines can remove a connection from the in-memory storage.

func (*Cache) Subscribe

func (c *Cache) Subscribe(_ context.Context, conn common.ConnectionWrapper, channels ...channel.Channel)

Subscribe stores the subscription for the input connection and list of the channels into the internal maps.

This function is thread-safe and multiple goroutines can subscribe to a list of channel concurrently.

func (*Cache) Unsubscribe

func (c *Cache) Unsubscribe(_ context.Context, connID string, channels ...channel.Channel)

Unsubscribe removes the input channels subscription from the internal maps.

This function is thread-safe and multiple goroutines can unsubscribe a list of channel concurrently.

func (*Cache) UnsubscribeUserID

func (c *Cache) UnsubscribeUserID(_ context.Context, connID string, userID string, ch channel.Channel)

UnsubscribeUserID removes the input channels subscription from the internal maps.

This function is thread-safe and multiple goroutines can unsubscribe a list of channel concurrently.

It removes the userID and connID mapping from the storage. The main usage of this function is when the auth token is expired and there is no need to validate token again. To subscribe again, client should send the channels with token again.

type Dispatch

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

Dispatch is a mechanism to send the public and private messages to the available connection per channel. It uses a storage to get the connections.

func NewDispatch

func NewDispatch(store store, logger log.Logger) *Dispatch

NewDispatch creates a new instance of Dispatch struct.

func (*Dispatch) SendPrivateMessage

func (d *Dispatch) SendPrivateMessage(ctx context.Context, ch channel.Channel, userID string, message interface{}) error

SendPrivateMessage sends the input message to the input channel if the client already authenticated with the input userID. Otherwise, skips and returns.

On each call it will authenticate the connection to ensure token is not expired.

SendPrivateMessage might return token expiration or json marshal errors.

func (*Dispatch) SendPublicMessage

func (d *Dispatch) SendPublicMessage(ctx context.Context, ch channel.Channel, message interface{}) error

SendPublicMessage sends the input message to the available connections of input channel.

This process is thread safe if the store.Connections be thread safe.

SendPublicMessage might return json marshal error.

type MessageOut

type MessageOut struct {
	Channel channel.Channel `json:"channel"`
	Data    interface{}     `json:"data"`
}

MessageOut represents the outbound message. Each the outbound message includes a channel name that the message belongs to it, and the data that is the main content.

type MessageType

type MessageType string

MessageType is an alias type of string that represent client message type. It describes the client action.

const (
	// MessageTypeSubscribe subscribes client to one or more requested channels.
	MessageTypeSubscribe MessageType = "subscribe"

	// MessageTypeUnsubscribe unsubscribes client from the channels that has been
	// already subscribed by the client.
	MessageTypeUnsubscribe MessageType = "unsubscribe"
)

func (MessageType) String

func (m MessageType) String() string