server

Documentation

Overview

packet server provides a MQTT 3.1.1 compliant MQTT server.

Index

• Constants
• Variables
• type Server
  • func New() *Server
  • func (s *Server) AddListener(listener listeners.Listener, config *listeners.Config) error
  • func (s *Server) AddStore(p persistence.Store) error
  • func (s *Server) Close() error
  • func (s *Server) EstablishConnection(lid string, c net.Conn, ac auth.Controller) error
  • func (s *Server) Publish(topic string, payload []byte, retain bool) error
  • func (s *Server) ResendClientInflight(cl *clients.Client, force bool) error
  • func (s *Server) Serve() error

Constants

const (
	Version = "1.0.3" // the server version.
)

Variables

var (
	ErrListenerIDExists     = errors.New("Listener id already exists")
	ErrReadConnectInvalid   = errors.New("Connect packet was not valid")
	ErrConnectNotAuthorized = errors.New("Connect packet was not authorized")
	ErrInvalidTopic         = errors.New("Cannot publish to $ and $SYS topics")

	// SysTopicInterval is the number of milliseconds between $SYS topic publishes.
	SysTopicInterval time.Duration = 30000
)

Types

type Server

type Server struct {
	Listeners *listeners.Listeners // listeners are network interfaces which listen for new connections.
	Clients   *clients.Clients     // clients which are known to the broker.
	Topics    *topics.Index        // an index of topic filter subscriptions and retained messages.
	System    *system.Info         // values about the server commonly found in $SYS topics.
	Store     persistence.Store    // a persistent storage backend if desired.

	Events events.Events // overrideable event hooks.
	// contains filtered or unexported fields
}

Server is an MQTT broker server. It should be created with server.New() in order to ensure all the internal fields are correctly populated.

func New

func New() *Server

New returns a new instance of an MQTT broker.

func (*Server) AddListener

func (s *Server) AddListener(listener listeners.Listener, config *listeners.Config) error

AddListener adds a new network listener to the server.

func (*Server) AddStore

func (s *Server) AddStore(p persistence.Store) error

AddStore assigns a persistent storage backend to the server. This must be called before calling server.Server().

func (*Server) Close

func (s *Server) Close() error

Close attempts to gracefully shutdown the server, all listeners, clients, and stores.

func (*Server) EstablishConnection

func (s *Server) EstablishConnection(lid string, c net.Conn, ac auth.Controller) error

EstablishConnection establishes a new client when a listener accepts a new connection.

func (*Server) Publish

func (s *Server) Publish(topic string, payload []byte, retain bool) error

Publish creates a publish packet from a payload and sends it to the inline.pub channel, where it is written directly to the outgoing byte buffers of any clients subscribed to the given topic. Because the message is written directly within the server, QoS is inherently 2 (exactly once).

func (*Server) ResendClientInflight

func (s *Server) ResendClientInflight(cl *clients.Client, force bool) error

ResendClientInflight attempts to resend all undelivered inflight messages to a client.

func (*Server) Serve

func (s *Server) Serve() error

Serve starts the event loops responsible for establishing client connections on all attached listeners, and publishing the system topics.