engineredis

Documentation

Index

• Constants
• func Configure(setter config.Setter) error
• func Plugin(n *node.Node, getter config.Getter) (engine.Engine, error)
• type ChannelID
• type RedisEngine
  • func New(n *node.Node, configs []*ShardConfig) (*RedisEngine, error)
  • func (e *RedisEngine) AddPresence(ch string, uid string, info proto.ClientInfo, expire int) error
  • func (e *RedisEngine) Channels() ([]string, error)
  • func (e *RedisEngine) History(ch string, limit int) ([]proto.Message, error)
  • func (e *RedisEngine) Name() string
  • func (e *RedisEngine) Presence(ch string) (map[string]proto.ClientInfo, error)
  • func (e *RedisEngine) PublishAdmin(message *proto.AdminMessage) <-chan error
  • func (e *RedisEngine) PublishControl(message *proto.ControlMessage) <-chan error
  • func (e *RedisEngine) PublishJoin(message *proto.JoinMessage, opts *channel.Options) <-chan error
  • func (e *RedisEngine) PublishLeave(message *proto.LeaveMessage, opts *channel.Options) <-chan error
  • func (e *RedisEngine) PublishMessage(message *proto.Message, opts *channel.Options) <-chan error
  • func (e *RedisEngine) RemovePresence(ch string, uid string) error
  • func (e *RedisEngine) Run() error
  • func (e *RedisEngine) Shutdown() error
  • func (e *RedisEngine) Subscribe(ch string) error
  • func (e *RedisEngine) Unsubscribe(ch string) error
• type Shard
  • func NewShard(n *node.Node, conf *ShardConfig) (*Shard, error)
  • func (e *Shard) AddPresence(ch string, uid string, info proto.ClientInfo, expire int) error
  • func (e *Shard) Channels() ([]string, error)
  • func (e *Shard) History(ch string, limit int) ([]proto.Message, error)
  • func (e *Shard) Presence(ch string) (map[string]proto.ClientInfo, error)
  • func (e *Shard) PublishAdmin(message *proto.AdminMessage) <-chan error
  • func (e *Shard) PublishControl(message *proto.ControlMessage) <-chan error
  • func (e *Shard) PublishJoin(message *proto.JoinMessage, opts *channel.Options) <-chan error
  • func (e *Shard) PublishLeave(message *proto.LeaveMessage, opts *channel.Options) <-chan error
  • func (e *Shard) PublishMessage(message *proto.Message, opts *channel.Options) <-chan error
  • func (e *Shard) RemovePresence(ch string, uid string) error
  • func (e *Shard) Run() error
  • func (e *Shard) Subscribe(ch string) error
  • func (e *Shard) Unsubscribe(ch string) error
• type ShardConfig

Constants

const (
	// RedisSubscribeChannelSize is the size for the internal buffered channels RedisEngine
	// uses to synchronize subscribe/unsubscribe.
	RedisSubscribeChannelSize = 4096
	// RedisPubSubWorkerChannelSize sets buffer size of channel to which we send all
	// messages received from Redis PUB/SUB connection to process in separate goroutine.
	RedisPubSubWorkerChannelSize = 4096
	// RedisSubscribeBatchLimit is a maximum number of channels to include in a single subscribe
	// call. Redis documentation doesn't specify a maximum allowed but we think it probably makes
	// sense to keep a sane limit given how many subscriptions a single Centrifugo instance might
	// be handling.
	RedisSubscribeBatchLimit = 2048
	// RedisPublishChannelSize is the size for the internal buffered channel RedisEngine uses
	// to collect publish requests.
	RedisPublishChannelSize = 1024
	// RedisPublishBatchLimit is a maximum limit of publish requests one batched publish
	// operation can contain.
	RedisPublishBatchLimit = 2048
	// RedisDataBatchLimit limits amount of data operations combined in one pipeline.
	RedisDataBatchLimit = 8
	// RedisDataChannelSize is a buffer size of channel with data operation requests.
	RedisDataChannelSize = 256
)

const (
	// RedisAPIKeySuffix is a suffix for api queue (LIST) key.
	RedisAPIKeySuffix = ".api"
	// RedisControlChannelSuffix is a suffix for control channel.
	RedisControlChannelSuffix = ".control"
	// RedisPingChannelSuffix is a suffix for ping channel.
	RedisPingChannelSuffix = ".ping"
	// RedisAdminChannelSuffix is a suffix for admin channel.
	RedisAdminChannelSuffix = ".admin"
	// RedisMessageChannelPrefix is a prefix before channel name for client messages.
	RedisMessageChannelPrefix = ".message."
	// RedisJoinChannelPrefix is a prefix before channel name for join messages.
	RedisJoinChannelPrefix = ".join."
	// RedisLeaveChannelPrefix is a prefix before channel name for leave messages.
	RedisLeaveChannelPrefix = ".leave."
)

Functions

func Configure

func Configure(setter config.Setter) error

Configure is a Configurator function for Redis engine.

func Plugin

func Plugin(n *node.Node, getter config.Getter) (engine.Engine, error)

Plugin returns Redis Engine.

Types

type ChannelID

type ChannelID string

ChannelID is unique channel identificator in Redis.

type RedisEngine

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

RedisEngine uses Redis datastructures and PUB/SUB to manage Centrifugo logic. This engine allows to scale Centrifugo - you can run several Centrifugo instances connected to the same Redis and load balance clients between instances.

func New

func New(n *node.Node, configs []*ShardConfig) (*RedisEngine, error)

New initializes Redis Engine.

func (*RedisEngine) AddPresence

func (e *RedisEngine) AddPresence(ch string, uid string, info proto.ClientInfo, expire int) error

AddPresence - see engine interface description.

func (*RedisEngine) Channels

func (e *RedisEngine) Channels() ([]string, error)

Channels - see engine interface description.

func (*RedisEngine) History

func (e *RedisEngine) History(ch string, limit int) ([]proto.Message, error)

History - see engine interface description.

func (*RedisEngine) Name

func (e *RedisEngine) Name() string

Name returns name of engine.

func (*RedisEngine) Presence

func (e *RedisEngine) Presence(ch string) (map[string]proto.ClientInfo, error)

Presence - see engine interface description.

func (*RedisEngine) PublishAdmin

func (e *RedisEngine) PublishAdmin(message *proto.AdminMessage) <-chan error

PublishAdmin- see engine interface description.

func (*RedisEngine) PublishControl

func (e *RedisEngine) PublishControl(message *proto.ControlMessage) <-chan error

PublishControl - see engine interface description.

func (*RedisEngine) PublishJoin

func (e *RedisEngine) PublishJoin(message *proto.JoinMessage, opts *channel.Options) <-chan error

PublishJoin - see engine interface description.

func (*RedisEngine) PublishLeave

func (e *RedisEngine) PublishLeave(message *proto.LeaveMessage, opts *channel.Options) <-chan error

PublishLeave - see engine interface description.

func (*RedisEngine) PublishMessage

func (e *RedisEngine) PublishMessage(message *proto.Message, opts *channel.Options) <-chan error

PublishMessage - see engine interface description.

func (*RedisEngine) RemovePresence

func (e *RedisEngine) RemovePresence(ch string, uid string) error

RemovePresence - see engine interface description.

func (*RedisEngine) Run

func (e *RedisEngine) Run() error

Run runs engine after node initialized.

func (*RedisEngine) Shutdown

func (e *RedisEngine) Shutdown() error

Shutdown shuts down Redis engine.

func (*RedisEngine) Subscribe

func (e *RedisEngine) Subscribe(ch string) error

Subscribe - see engine interface description.

func (*RedisEngine) Unsubscribe

func (e *RedisEngine) Unsubscribe(ch string) error

Unsubscribe - see engine interface description.

type Shard

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

Shard has everything to connect to Redis instance.

func NewShard

func NewShard(n *node.Node, conf *ShardConfig) (*Shard, error)

NewShard initializes new Redis shard.

func (*Shard) AddPresence

func (e *Shard) AddPresence(ch string, uid string, info proto.ClientInfo, expire int) error

AddPresence - see engine interface description.

func (*Shard) Channels

func (e *Shard) Channels() ([]string, error)

Channels - see engine interface description. Requires Redis >= 2.8.0 (http://redis.io/commands/pubsub)

func (*Shard) History

func (e *Shard) History(ch string, limit int) ([]proto.Message, error)

History - see engine interface description.

func (*Shard) Presence

func (e *Shard) Presence(ch string) (map[string]proto.ClientInfo, error)

Presence - see engine interface description.

func (*Shard) PublishAdmin

func (e *Shard) PublishAdmin(message *proto.AdminMessage) <-chan error

PublishAdmin - see engine interface description.

func (*Shard) PublishControl

func (e *Shard) PublishControl(message *proto.ControlMessage) <-chan error

PublishControl - see engine interface description.

func (*Shard) PublishJoin

func (e *Shard) PublishJoin(message *proto.JoinMessage, opts *channel.Options) <-chan error

PublishJoin - see engine interface description.

func (*Shard) PublishLeave

func (e *Shard) PublishLeave(message *proto.LeaveMessage, opts *channel.Options) <-chan error

PublishLeave - see engine interface description.

func (*Shard) PublishMessage

func (e *Shard) PublishMessage(message *proto.Message, opts *channel.Options) <-chan error

PublishMessage - see engine interface description.

func (*Shard) RemovePresence

func (e *Shard) RemovePresence(ch string, uid string) error

RemovePresence - see engine interface description.

func (*Shard) Run

func (e *Shard) Run() error

Run runs Redis shard.

func (*Shard) Subscribe

func (e *Shard) Subscribe(ch string) error

Subscribe - see engine interface description.

func (*Shard) Unsubscribe

func (e *Shard) Unsubscribe(ch string) error

Unsubscribe - see engine interface description.

type ShardConfig

type ShardConfig struct {
	// Host is Redis server host.
	Host string
	// Port is Redis server port.
	Port string
	// Password is password to use when connecting to Redis database. If empty then password not used.
	Password string
	// DB is Redis database number as string. If empty then database 0 used.
	DB string
	// MasterName is a name of Redis instance master Sentinel monitors.
	MasterName string
	// SentinelAddrs is a slice of Sentinel addresses.
	SentinelAddrs []string
	// PoolSize is a size of Redis connection pool.
	PoolSize int
	// API enables listening for API queues to publish API commands into Centrifugo via pushing
	// commands into Redis queue.
	API bool
	// NumAPIShards is a number of sharded API queues in Redis to increase volume of commands
	// (most probably publish) that Centrifugo instance can process.
	NumAPIShards int
	// Prefix to use before every channel name and key in Redis.
	Prefix string
	// PubSubNumWorkers sets how many PUB/SUB message processing workers will be started.
	// By default we start runtime.NumCPU() workers.
	PubSubNumWorkers int
	// ReadTimeout is a timeout on read operations. Note that at moment it should be greater
	// than node ping publish interval in order to prevent timing out Pubsub connection's
	// Receive call.
	ReadTimeout time.Duration
	// WriteTimeout is a timeout on write operations
	WriteTimeout time.Duration
	// ConnectTimeout is a timeout on connect operation
	ConnectTimeout time.Duration
}

ShardConfig is struct with Redis Engine options.