subscriber

Documentation

Index

• type Config
• type ConnectionProvider
• type OfflinePublish
• type OnlinePublish
• type Subscriptions
• type Type
  • func New(c *Config) *Type
  • func (s *Type) Acquire()
  • func (s *Type) HasSubscriptions() bool
  • func (s *Type) Hash() uintptr
  • func (s *Type) ID() string
  • func (s *Type) Offline(shutdown bool)
  • func (s *Type) Online(c OnlinePublish)
  • func (s *Type) OnlineRedirect(c OnlinePublish)
  • func (s *Type) Publish(p *packet.Publish, grantedQoS packet.QosType, ops packet.SubscriptionOptions, ...) error
  • func (s *Type) Release()
  • func (s *Type) Subscribe(topic string, params *topicsTypes.SubscriptionParams) (packet.QosType, []*packet.Publish, error)
  • func (s *Type) Subscriptions() Subscriptions
  • func (s *Type) UnSubscribe(topic string) error
  • func (s *Type) Version() packet.ProtocolVersion

Types

type Config

type Config struct {
	ID               string
	Topics           topicsTypes.SubscriberInterface
	OnOfflinePublish OfflinePublish
	OfflineQoS0      bool
	Version          packet.ProtocolVersion
}

Config subscriber config options

type ConnectionProvider

type ConnectionProvider interface {
	ID() string
	Subscriptions() Subscriptions
	Subscribe(string, *topicsTypes.SubscriptionParams) (packet.QosType, []*packet.Publish, error)
	UnSubscribe(string) error
	HasSubscriptions() bool
	Online(c OnlinePublish)
	OnlineRedirect(c OnlinePublish)
	Offline(bool)
	Hash() uintptr
	Version() packet.ProtocolVersion
}

ConnectionProvider passed to present network connection

type OfflinePublish

type OfflinePublish func(string, *packet.Publish)

OfflinePublish invoked when subscriber respective to sessions receive message

type OnlinePublish

type OnlinePublish func(*packet.Publish)

OnlinePublish invoked when subscriber respective to sessions receive message

type Subscriptions

type Subscriptions map[string]*topicsTypes.SubscriptionParams

Subscriptions contains active subscriptions with respective subscription parameters

type Type

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

Type subscriber object

func New

func New(c *Config) *Type

New allocate new subscriber

func (*Type) Acquire

func (s *Type) Acquire()

Acquire prevent subscriber being deleted before active writes finished

func (*Type) HasSubscriptions

func (s *Type) HasSubscriptions() bool

HasSubscriptions either has active subscriptions or not

func (*Type) Hash

func (s *Type) Hash() uintptr

Hash returns address of the provider struct. Used by topics provider as a key to subscriber object

func (*Type) ID

func (s *Type) ID() string

ID get subscriber id

func (*Type) Offline

func (s *Type) Offline(shutdown bool)

Offline put session offline if shutdown is true it does unsubscribe from all active subscriptions

func (*Type) Online

func (s *Type) Online(c OnlinePublish)

Online moves subscriber to online state since this moment all of publishes are forwarded to provided callback

func (*Type) OnlineRedirect

func (s *Type) OnlineRedirect(c OnlinePublish)

OnlineRedirect set new online publish callback

func (*Type) Publish

func (s *Type) Publish(p *packet.Publish, grantedQoS packet.QosType, ops packet.SubscriptionOptions, ids []uint32) error

Publish message accordingly to subscriber state online: forward message to session offline: persist message

func (*Type) Release

func (s *Type) Release()

Release subscriber once topics provider finished write

func (*Type) Subscribe

func (s *Type) Subscribe(topic string, params *topicsTypes.SubscriptionParams) (packet.QosType, []*packet.Publish, error)

Subscribe to given topic

func (*Type) Subscriptions

func (s *Type) Subscriptions() Subscriptions

Subscriptions list active subscriptions

func (*Type) UnSubscribe

func (s *Type) UnSubscribe(topic string) error

UnSubscribe from given topic

func (*Type) Version

func (s *Type) Version() packet.ProtocolVersion

Version return MQTT protocol version