broadcast

README

RoadRunner: Event Broadcasting

Features:

• Produce and consume events
• Subscribe to topic from Golang or PHP
• Multi-Topic subscription
• WebSocket support
  • Server and topic access verification via PHP
• Redis and In-Memory brokers
• Custom topic command handlers
• Automatic reconnects
• Prometheus metrics

License:

The MIT License (MIT). Please see LICENSE for more information. Maintained by Spiral Scout.

Documentation

Index

• Constants
• type Broker
• type Client
  • func (c *Client) Close()
  • func (c *Client) Publish(msg ...*Message) error
  • func (c *Client) Subscribe(topics ...string) error
  • func (c *Client) Topics() []string
  • func (c *Client) Unsubscribe(topics ...string)
• type Command
  • func (cmd *Command) Unmarshal(v interface{}) error
• type CommandHandler
• type Config
  • func (c *Config) Hydrate(cfg service.Config) error
  • func (c *Config) InitDefaults() error
• type ConnContext
• type ErrorEvent
• type Memory
  • func (m *Memory) Broadcast(messages ...*Message) error
  • func (m *Memory) Serve() error
  • func (m *Memory) Stop()
  • func (m *Memory) Subscribe(upstream chan *Message, topics ...string) error
  • func (m *Memory) Unsubscribe(upstream chan *Message, topics ...string)
• type Message
  • func NewMessage(topic string, payload interface{}) *Message
• type Redis
  • func (r *Redis) Broadcast(messages ...*Message) error
  • func (r *Redis) Serve() error
  • func (r *Redis) Stop()
  • func (r *Redis) Subscribe(upstream chan *Message, topics ...string) error
  • func (r *Redis) Unsubscribe(upstream chan *Message, topics ...string)
• type RedisConfig
• type Service
  • func (s *Service) AddCommand(name string, cmd CommandHandler)
  • func (s *Service) AddListener(l func(event int, ctx interface{}))
  • func (s *Service) Broadcast(msg ...*Message) error
  • func (s *Service) Broker() Broker
  • func (s *Service) Init(cfg *Config, r *rpc.Service, h *rhttp.Service, e env.Environment) (bool, error)
  • func (s *Service) NewClient(upstream chan *Message) *Client
  • func (s *Service) Serve() (err error)
  • func (s *Service) Stop()
  • func (s *Service) Subscribe(upstream chan *Message, topics ...string) error
  • func (s *Service) Unsubscribe(upstream chan *Message, topics ...string)
• type TopicEvent

Constants

const (
	// EventWebsocketConnect fired when new client is connected, the context is *websocket.Conn.
	EventWebsocketConnect = iota + 2500

	// EventWebsocketDisconnect fired when websocket is disconnected, context is empty.
	EventWebsocketDisconnect

	// EventWebsocketJoin caused when topics are being consumed, context if *TopicEvent.
	EventWebsocketJoin

	// EventWebsocketLeave caused when topic consumption are stopped, context if *TopicEvent.
	EventWebsocketLeave

	// EventWebsocketError when any broadcast error occurred, the context is *ErrorEvent.
	EventWebsocketError

	// EventBrokerError the context is error.
	EventBrokerError
)

const ID = "broadcast"

ID defines public service name.

Types

type Broker

type Broker interface {
	// Serve serves broker.
	Serve() error

	// close the consumption and disconnect broker.
	Stop()

	// Subscribe broker to one or multiple topics.
	Subscribe(upstream chan *Message, topics ...string) error

	// Unsubscribe broker from one or multiple topics.
	Unsubscribe(upstream chan *Message, topics ...string)

	// Broadcast one or multiple Messages.
	Broadcast(messages ...*Message) error
}

Broker defines the ability to operate as message passing broker.

type Client

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

NewClient subscribes to a given topic and consumes or publish messages to it. NewClient will be receiving messages it produced.

func (*Client) Close

func (c *Client) Close()

Close the client and consumption.

func (*Client) Publish

func (c *Client) Publish(msg ...*Message) error

Publish message into associated topic or topics.

func (*Client) Subscribe

func (c *Client) Subscribe(topics ...string) error

NewClient client to specific topics.

func (*Client) Topics

func (c *Client) Topics() []string

Topics return all the topics client subscribed to.

func (*Client) Unsubscribe

func (c *Client) Unsubscribe(topics ...string)

Unsubscribe client from specific topics

type Command

type Command struct {
	// Cmd type.
	Cmd string `json:"cmd"`

	// Args contains command specific payload.
	Args json.RawMessage `json:"args"`
}

Command contains information send by user.

func (*Command) Unmarshal

func (cmd *Command) Unmarshal(v interface{}) error

Unmarshal command data.

type CommandHandler

type CommandHandler func(ctx *ConnContext, cmd []byte)

CommandHandler handles custom commands.

type Config

type Config struct {
	// Path defines on this URL the middleware must be activated. Same path must be handled by underlying
	// application kernel to authorize the consumption. Optional.
	Path string

	// RedisConfig configures redis broker.
	Redis *RedisConfig
}

Config configures the broadcast extension.

func (*Config) Hydrate

func (c *Config) Hydrate(cfg service.Config) error

Hydrate reads the configuration values from the source configuration.

func (*Config) InitDefaults

func (c *Config) InitDefaults() error

InitDefaults enables in memory broadcast configuration.

type ConnContext

type ConnContext struct {
	// Upstream to push Messages into.
	Upstream chan *Message

	// Conn to the client.
	Conn *websocket.Conn

	// Topics contain list of currently subscribed topics.
	Topics []string
}

ConnContext represents the connection and it's state.

type ErrorEvent

type ErrorEvent struct {
	// Conn specific to the error.
	Conn *websocket.Conn

	// Caused contains job specific error.
	Caused error
}

ErrorEvent represents singular broadcast error event.

type Memory

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

Memory manages broadcasting in memory.

func (*Memory) Broadcast

func (m *Memory) Broadcast(messages ...*Message) error

Broadcast one or multiple Messages.

func (*Memory) Serve

func (m *Memory) Serve() error

Serve serves broker.

func (*Memory) Stop

func (m *Memory) Stop()

close the consumption and disconnect broker.

func (*Memory) Subscribe

func (m *Memory) Subscribe(upstream chan *Message, topics ...string) error

Subscribe broker to one or multiple channels.

func (*Memory) Unsubscribe

func (m *Memory) Unsubscribe(upstream chan *Message, topics ...string)

Unsubscribe broker from one or multiple channels.

type Message

type Message struct {
	// Topic message been pushed into.
	Topic string `json:"topic"`

	// Payload to be broadcasted. Must be valid JSON.
	Payload json.RawMessage `json:"payload"`
}

Message represent single message.

func NewMessage

func NewMessage(topic string, payload interface{}) *Message

NewMessage creates new message with JSON payload.

type Redis

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

Redis based broadcast router.

func (*Redis) Broadcast

func (r *Redis) Broadcast(messages ...*Message) error

Broadcast one or multiple Messages.

func (*Redis) Serve

func (r *Redis) Serve() error

Serve serves broker.

func (*Redis) Stop

func (r *Redis) Stop()

close the consumption and disconnect broker.

func (*Redis) Subscribe

func (r *Redis) Subscribe(upstream chan *Message, topics ...string) error

Subscribe broker to one or multiple channels.

func (*Redis) Unsubscribe

func (r *Redis) Unsubscribe(upstream chan *Message, topics ...string)

Unsubscribe broker from one or multiple channels.

type RedisConfig

type RedisConfig struct {
	Addr     string
	Password string
	DB       int
}

RedisConfig configures redis broker.

type Service

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

Service manages even broadcasting and websocket interface.

func (*Service) AddCommand

func (s *Service) AddCommand(name string, cmd CommandHandler)

AddCommand attached custom client command handler, for websocket only.

func (*Service) AddListener

func (s *Service) AddListener(l func(event int, ctx interface{}))

AddListener attaches server event controller.

func (*Service) Broadcast

func (s *Service) Broadcast(msg ...*Message) error

Broadcast one or multiple Messages.

func (*Service) Broker

func (s *Service) Broker() Broker

Broker returns associated broker.

func (*Service) Init

func (s *Service) Init(cfg *Config, r *rpc.Service, h *rhttp.Service, e env.Environment) (bool, error)

Init service.

func (*Service) NewClient

func (s *Service) NewClient(upstream chan *Message) *Client

NewClient returns single connected client with ability to consume or produce into topic.

func (*Service) Serve

func (s *Service) Serve() (err error)

Serve broadcast broker.

func (*Service) Stop

func (s *Service) Stop()

close broadcast broker.

func (*Service) Subscribe

func (s *Service) Subscribe(upstream chan *Message, topics ...string) error

Subscribe broker to one or multiple topics.

func (*Service) Unsubscribe

func (s *Service) Unsubscribe(upstream chan *Message, topics ...string)

Unsubscribe broker from one or multiple topics.

type TopicEvent

type TopicEvent struct {
	// Conn associated with topics.
	Conn *websocket.Conn

	// Topics specific to event.
	Topics []string
}

TopicEvent caused when topic is joined or left.