broadcast

README

RoadRunner: Event Broadcasting

Features:

• Golang Client and PHP producer
• Multi-topic subscriptions
• In-memory routing and wildcard subscriptions
• Redis and in-memory brokers
• Automatic reconnects

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) Channel() chan *Message
  • func (c *Client) Close() (err error)
  • func (c *Client) Patterns() []string
  • func (c *Client) Publish(msg ...*Message) error
  • func (c *Client) Subscribe(topics ...string) error
  • func (c *Client) SubscribePattern(pattern string) error
  • func (c *Client) Topics() []string
  • func (c *Client) Unsubscribe(topics ...string) error
  • func (c *Client) UnsubscribePattern(pattern string) error
• type Config
  • func (c *Config) Hydrate(cfg service.Config) error
  • func (c *Config) InitDefaults() error
• type Memory
  • func (m *Memory) Publish(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) SubscribePattern(upstream chan *Message, pattern string) error
  • func (m *Memory) Unsubscribe(upstream chan *Message, topics ...string) error
  • func (m *Memory) UnsubscribePattern(upstream chan *Message, pattern string) error
• type Message
• type Redis
  • func (r *Redis) Publish(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) SubscribePattern(upstream chan *Message, pattern string) error
  • func (r *Redis) Unsubscribe(upstream chan *Message, topics ...string) error
  • func (r *Redis) UnsubscribePattern(upstream chan *Message, pattern string) error
• type RedisConfig
• type Router
  • func NewRouter() *Router
  • func (r *Router) Dispatch(msg *Message)
  • func (r *Router) Subscribe(upstream chan *Message, topics ...string) (newTopics []string)
  • func (r *Router) SubscribePattern(upstream chan *Message, pattern string) (newPatterns []string, err error)
  • func (r *Router) Unsubscribe(upstream chan *Message, topics ...string) (dropTopics []string)
  • func (r *Router) UnsubscribePattern(upstream chan *Message, pattern string) (dropPatterns []string)
• type Service
  • func (s *Service) Broker() Broker
  • func (s *Service) Init(cfg *Config, rpc *rpc.Service) (ok bool, err error)
  • func (s *Service) NewClient() *Client
  • func (s *Service) Publish(msg ...*Message) error
  • func (s *Service) Serve() (err error)
  • func (s *Service) Stop()

Constants

const ID = "broadcast"

ID defines public service name.

Types

type Broker

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

	// Stop closes the consumption and disconnects broker.
	Stop()

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

	// SubscribePattern broker to pattern.
	SubscribePattern(upstream chan *Message, pattern string) error

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

	// UnsubscribePattern broker from pattern.
	UnsubscribePattern(upstream chan *Message, pattern string) error

	// Publish one or multiple Channel.
	Publish(messages ...*Message) error
}

Broker defines the ability to operate as message passing broker.

type Client

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

Client subscribes to a given topic and consumes or publish messages to it.

func (*Client) Channel

func (c *Client) Channel() chan *Message

Channel returns incoming messages channel.

func (*Client) Close

func (c *Client) Close() (err error)

Close the client and consumption.

func (*Client) Patterns

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

Patterns return all the patterns client subscribed to.

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

Subscribe client to specific topics.

func (*Client) SubscribePattern

func (c *Client) SubscribePattern(pattern string) error

SubscribePattern subscribe client to the specific topic pattern.

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) error

Unsubscribe client from specific topics

func (*Client) UnsubscribePattern

func (c *Client) UnsubscribePattern(pattern string) error

UnsubscribePattern client from topic pattern.

type Config

type Config struct {
	// 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 Memory

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

Memory manages broadcasting in memory.

func (*Memory) Publish

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

Publish one or multiple Channel.

func (*Memory) Serve

func (m *Memory) Serve() error

Serve serves broker.

func (*Memory) Stop

func (m *Memory) Stop()

Stop closes the consumption and disconnects broker.

func (*Memory) Subscribe

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

Subscribe broker to one or multiple channels.

func (*Memory) SubscribePattern

func (m *Memory) SubscribePattern(upstream chan *Message, pattern string) error

SubscribePattern broker to pattern.

func (*Memory) Unsubscribe

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

Unsubscribe broker from one or multiple channels.

func (*Memory) UnsubscribePattern

func (m *Memory) UnsubscribePattern(upstream chan *Message, pattern string) error

UnsubscribePattern broker from pattern.

type Message

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

	// Payload to be broadcasted. Must be valid json when transferred over RPC.
	Payload json.RawMessage `json:"payload"`
}

Message represent single message.

type Redis

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

Redis based broadcast Router.

func (*Redis) Publish

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

Publish one or multiple Channel.

func (*Redis) Serve

func (r *Redis) Serve() error

Serve serves broker.

func (*Redis) Stop

func (r *Redis) Stop()

Stop closes the consumption and disconnects broker.

func (*Redis) Subscribe

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

Subscribe broker to one or multiple channels.

func (*Redis) SubscribePattern

func (r *Redis) SubscribePattern(upstream chan *Message, pattern string) error

SubscribePattern broker to pattern.

func (*Redis) Unsubscribe

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

Unsubscribe broker from one or multiple channels.

func (*Redis) UnsubscribePattern

func (r *Redis) UnsubscribePattern(upstream chan *Message, pattern string) error

UnsubscribePattern broker from pattern.

type RedisConfig

type RedisConfig struct {
	// Addr of the redis server.
	Addr string

	// Password to redis server.
	Password string

	// DB index.
	DB int
}

RedisConfig configures redis broker.

type Router

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

Router performs internal message routing to multiple subscribers.

func NewRouter

func NewRouter() *Router

NewRouter creates new topic and pattern router.

func (*Router) Dispatch

func (r *Router) Dispatch(msg *Message)

Dispatch to all connected topics.

func (*Router) Subscribe

func (r *Router) Subscribe(upstream chan *Message, topics ...string) (newTopics []string)

Subscribe to topic and return list of newly assigned topics.

func (*Router) SubscribePattern

func (r *Router) SubscribePattern(upstream chan *Message, pattern string) (newPatterns []string, err error)

SubscribePattern subscribes to glob parent and return true and return array of newly added patterns. Error in case if blob is invalid.

func (*Router) Unsubscribe

func (r *Router) Unsubscribe(upstream chan *Message, topics ...string) (dropTopics []string)

Unsubscribe from given list of topics and return list of topics which are no longer claimed.

func (*Router) UnsubscribePattern

func (r *Router) UnsubscribePattern(upstream chan *Message, pattern string) (dropPatterns []string)

UnsubscribePattern unsubscribe from the pattern and returns an array of patterns which are no longer claimed.

type Service

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

Service manages even broadcasting and websocket interface.

func (*Service) Broker

func (s *Service) Broker() Broker

Broker returns associated broker.

func (*Service) Init

func (s *Service) Init(cfg *Config, rpc *rpc.Service) (ok bool, err error)

Init service.

func (*Service) NewClient

func (s *Service) NewClient() *Client

NewClient returns single connected client with ability to consume or produce into associated topic(svc).

func (*Service) Publish

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

Publish one or multiple Channel.

func (*Service) Serve

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

Serve broadcast broker.

func (*Service) Stop

func (s *Service) Stop()

Stop closes broadcast broker.