broker

Documentation

Overview

Package broker handles session registrations and delivery of messages through sessions.

Index

• Variables
• func FeedPending(sess BrokerSession) error
• func GetInfoInt(msg *protocol.ConnectMsg, name string, defaultVal int) (int, error)
• func GetInfoString(msg *protocol.ConnectMsg, name, defaultVal string) (string, error)
• type BaseExchange
• type BroadcastExchange
  • func (sbe *BroadcastExchange) Acked(sess BrokerSession, done bool) error
  • func (sbe *BroadcastExchange) Init()
  • func (sbe *BroadcastExchange) Prepare(sess BrokerSession) (outMessage protocol.SplittableMsg, inMessage interface{}, err error)
• type Broker
• type BrokerConfig
• type BrokerSending
• type BrokerSession
• type ConnMetaExchange
  • func (cbe *ConnMetaExchange) Acked(sess BrokerSession, done bool) error
  • func (cbe *ConnMetaExchange) Prepare(sess BrokerSession) (outMessage protocol.SplittableMsg, inMessage interface{}, err error)
• type ErrAbort
  • func (ea *ErrAbort) Error() string
• type Exchange
• type ExchangesScratchArea
• type LevelsMap
• type SessionTracker
• type UnicastExchange
  • func (sue *UnicastExchange) Acked(sess BrokerSession, done bool) error
  • func (sue *UnicastExchange) Prepare(sess BrokerSession) (outMessage protocol.SplittableMsg, inMessage interface{}, err error)

Variables

var ErrNop = errors.New("nothing to send")

ErrNop returned by Prepare means nothing to do/send.

var ErrUnexpectedValue = &ErrAbort{"unexpected value in message"}

Unexpect value in message

Functions

func FeedPending

func FeedPending(sess BrokerSession) error

FeedPending feeds exchanges covering pending notifications into the session.

func GetInfoInt

func GetInfoInt(msg *protocol.ConnectMsg, name string, defaultVal int) (int, error)

GetInfoInt helps retrieving an integer out of a protocol.ConnectMsg.Info.

func GetInfoString

func GetInfoString(msg *protocol.ConnectMsg, name, defaultVal string) (string, error)

GetInfoString helps retrieveng a string out of a protocol.ConnectMsg.Info.

Types

type BaseExchange

type BaseExchange struct {
	Timestamp time.Time
}

type BroadcastExchange

type BroadcastExchange struct {
	ChanId        store.InternalChannelId
	TopLevel      int64
	Notifications []protocol.Notification
	Decoded       []map[string]interface{}
	BaseExchange
}

BroadcastExchange leads a session through delivering a BROADCAST. For simplicity it is fully public.

func (*BroadcastExchange) Acked

func (sbe *BroadcastExchange) Acked(sess BrokerSession, done bool) error

Acked deals with an ACK for a BROADCAST.

func (*BroadcastExchange) Init

func (sbe *BroadcastExchange) Init()

Init ensures the BroadcastExchange is fully initialized for the sessions.

func (*BroadcastExchange) Prepare

func (sbe *BroadcastExchange) Prepare(sess BrokerSession) (outMessage protocol.SplittableMsg, inMessage interface{}, err error)

Prepare session for a BROADCAST.

type Broker

type Broker interface {
	// Register the session.
	Register(connMsg *protocol.ConnectMsg, track SessionTracker) (BrokerSession, error)
	// Unregister the session.
	Unregister(BrokerSession)
}

Broker is responsible for registring sessions and delivering messages through them.

type BrokerConfig

type BrokerConfig interface {
	// SessionQueueSize gives the session queue size.
	SessionQueueSize() uint
	// BrokerQueueSize gives the internal broker queue size.
	BrokerQueueSize() uint
}

BrokerConfig gives access to the typical broker configuration.

type BrokerSending

type BrokerSending interface {
	// Broadcast channel.
	Broadcast(chanId store.InternalChannelId)
	// Unicast over channels.
	Unicast(chanIds ...store.InternalChannelId)
}

BrokerSending is the notification sending facet of the broker.

type BrokerSession

type BrokerSession interface {
	// SessionChannel returns the session control channel
	// on which the session gets exchanges to perform.
	SessionChannel() <-chan Exchange
	// DeviceIdentifier returns the device id string.
	DeviceIdentifier() string
	// DeviceImageModel returns the device model.
	DeviceImageModel() string
	// DeviceImageChannel returns the device system image channel.
	DeviceImageChannel() string
	// Levels returns the current channel levels for the session
	Levels() LevelsMap
	// ExchangeScratchArea returns the scratch area for exchanges.
	ExchangeScratchArea() *ExchangesScratchArea
	// Get gets the content of the channel with chanId.
	Get(chanId store.InternalChannelId, cachedOk bool) (int64, []protocol.Notification, error)
	// DropByMsgId drops notifications from the channel chanId by message id.
	DropByMsgId(chanId store.InternalChannelId, targets []protocol.Notification) error
	// Feed feeds exchange into the session.
	Feed(Exchange)
	// InternalChannelId() returns the channel id corresponding to the session.
	InternalChannelId() store.InternalChannelId
}

BrokerSession holds broker session state.

type ConnMetaExchange

type ConnMetaExchange struct {
	Msg protocol.OnewayMsg
}

ConnMetaExchange allows to send a CONNBROKEN or CONNWARN message.

func (*ConnMetaExchange) Acked

func (cbe *ConnMetaExchange) Acked(sess BrokerSession, done bool) error

CONNBROKEN/WARN aren't acked.

func (*ConnMetaExchange) Prepare

func (cbe *ConnMetaExchange) Prepare(sess BrokerSession) (outMessage protocol.SplittableMsg, inMessage interface{}, err error)

Prepare session for a CONNBROKEN/WARN.

type ErrAbort

type ErrAbort struct {
	Reason string
}

Session aborted error.

func (*ErrAbort) Error

func (ea *ErrAbort) Error() string

type Exchange

type Exchange interface {
	Prepare(sess BrokerSession) (outMessage protocol.SplittableMsg, inMessage interface{}, err error)
	Acked(sess BrokerSession, done bool) error
}

Exchange leads the session through performing an exchange, typically delivery.

type ExchangesScratchArea

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

Scratch area for exchanges, sessions should hold one of these.

type LevelsMap

type LevelsMap map[store.InternalChannelId]int64

LevelsMap is the type for holding channel levels for session.

type SessionTracker

type SessionTracker interface {
	// SessionId
	SessionId() string
}

type UnicastExchange

type UnicastExchange struct {
	ChanId   store.InternalChannelId
	CachedOk bool
	BaseExchange
}

UnicastExchange leads a session through delivering a NOTIFICATIONS message. For simplicity it is fully public.

func (*UnicastExchange) Acked

func (sue *UnicastExchange) Acked(sess BrokerSession, done bool) error

Acked deals with an ACK for a NOTIFICATIONS.

func (*UnicastExchange) Prepare

func (sue *UnicastExchange) Prepare(sess BrokerSession) (outMessage protocol.SplittableMsg, inMessage interface{}, err error)

Prepare session for a NOTIFICATIONS.