pubsub

Documentation

Index

• Constants
• type Client
  • func New(conn ConnectionParam) *Client
  • func (c *Client) Connect()
  • func (c *Client) GetState() uint8
  • func (c *Client) Listener(kind ListenerType, event string) *Listener
  • func (e *Client) On(ev EventType, h EventHandler)
  • func (e *Client) OnChannel(ev EventType) chan Event
  • func (e *Client) Once(ev EventType, h EventHandler)
  • func (c *Client) Subscribe(listener *Listener)
  • func (c *Client) TearDown()
  • func (c *Client) Unsubscribe(listener *Listener)
  • func (e *Client) WaitFor(ev EventType)
• type ConnectionParam
• type Event
• type EventHandler
• type EventType
• type Listener
  • func (l *Listener) Resubscribe()
  • func (l *Listener) Unsubscribe()
• type ListenerType
• type LogReconnectPolicy
  • func (l *LogReconnectPolicy) Next() time.Duration
  • func (l *LogReconnectPolicy) Reset()
• type ReconnectPolicy
• type StaticReconnectPolicy
  • func (s *StaticReconnectPolicy) Next() time.Duration
  • func (r *StaticReconnectPolicy) Reset()

Constants

const (
	// Not currently connected to a server.
	DisconnectedState uint8 = iota
	// Connected to a server, but not yet linked as a pubsub client.
	ConnectedState
	// We're in the process of closing the client.
	ClosingState
	// We were connected, but closed gracefully.
	ClosedState
)

Types

type Client

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

The Client is responsible for maintaining a subscribed redis client, reconnecting and resubscribing if it drops.

func New

func New(conn ConnectionParam) *Client

Creates and returns a new pubsub client client and subscribes to it.

func (*Client) Connect

func (c *Client) Connect()

Tries to reconnect to pubsub, looping until we're able to do so successfully. This must be called to activate the client.

func (*Client) GetState

func (c *Client) GetState() uint8

Gets the current client state.

func (*Client) Listener

func (c *Client) Listener(kind ListenerType, event string) *Listener

Convenience function to create a new listener for an event.

func (*Client) On

func (e *Client) On(ev EventType, h EventHandler)

Adds a handler that's executed when an event happens.

func (*Client) OnChannel

func (e *Client) OnChannel(ev EventType) chan Event

Creates a channel that gets written to when a new event comes in.

func (*Client) Once

func (e *Client) Once(ev EventType, h EventHandler)

Adds a handler that's executed once when an event is emitted.

func (*Client) Subscribe

func (c *Client) Subscribe(listener *Listener)

Subscribes to a Redis event. Strings are sent back down the listener channel when they come in, and

func (*Client) TearDown

func (c *Client) TearDown()

Tears down the client - closes the connection and stops listening for connections.

func (*Client) Unsubscribe

func (c *Client) Unsubscribe(listener *Listener)

Removes the listener from the list of subscribers. If it's the last one listening to that Redis event, we unsubscribe entirely.

func (*Client) WaitFor

func (e *Client) WaitFor(ev EventType)

Blocks until an event is received. Mainly for backwards-compatibility.

type ConnectionParam

type ConnectionParam struct {
	// Host:port
	Address string
	// Optional password. Defaults to no authentication.
	Password string
	// Policy to use for reconnections (defaults to
	// LogReconnectPolicy with a base of 10 and factor of 1 ms)
	Policy ReconnectPolicy
	// Dial timeout for redis (defaults to no timeout)
	Timeout time.Duration
}

Used to denote the parameters of the redis connection.

type Event

type Event struct {
	Type   EventType
	Packet interface{}
}

The event is sent down the client's Events channel when something happens!

type EventHandler

type EventHandler func(Event)

A function which handles an incoming event.

type EventType

type EventType uint8

Events that are sent down the "Events" channel.

const (
	ConnectedEvent EventType = iota
	DisconnectedEvent
	ClosingEvent
	ClosedEvent
	MessageEvent
	SubscribeEvent
	UnsubscribeEvent
	ErrorEvent
	AnyEvent
)

type Listener

type Listener struct {
	// The event slug we're listening for.
	Event string
	// and its type
	Type ListenerType
	// The channel we send events down for plain subscriptions.
	Messages chan redis.Message
	// The channel we send events down for pattern subscriptions.
	PMessages chan redis.PMessage
	// The client it's attached to.
	Client *Client
	// Whether its active. True by default, false after unsubscribed.
	Active bool
}

The listener is used to keep track of events that the client is listening to.

func (*Listener) Resubscribe

func (l *Listener) Resubscribe()

Resubscribes the listener.

func (*Listener) Unsubscribe

func (l *Listener) Unsubscribe()

Unsubscribes the listener.

type ListenerType

type ListenerType uint8

Used to denote the type of listener - channel or pattern.

const (
	Channel ListenerType = iota
	Pattern
)

type LogReconnectPolicy

type LogReconnectPolicy struct {
	// Base for the logarithim
	Base float64
	// Duration multiplier for the calculated value.
	Factor time.Duration
	// contains filtered or unexported fields
}

Reconnect policy which increases the reconnect day in a logarithmic fashion. The equation used is delay = log(tries) / log(base) * Factor

func (*LogReconnectPolicy) Next

func (l *LogReconnectPolicy) Next() time.Duration

func (*LogReconnectPolicy) Reset

func (l *LogReconnectPolicy) Reset()

type ReconnectPolicy

type ReconnectPolicy interface {
	// Gets the next reconnect time, usually incrementing some
	// counter so the next attempt is longer.
	Next() time.Duration
	// Resets the number of attempts.
	Reset()
}

The ReconnectPolicy defines the pattern of delay times used after a connection is lost before waiting to reconnection.

type StaticReconnectPolicy

type StaticReconnectPolicy struct {
	// Delay each time.
	Delay time.Duration
}

Reconnect policy which waits a set period of time on each connect.

func (*StaticReconnectPolicy) Next

func (s *StaticReconnectPolicy) Next() time.Duration

func (*StaticReconnectPolicy) Reset

func (r *StaticReconnectPolicy) Reset()