server

Documentation

Overview

package server provides a MQTT 3.1.1 compliant MQTT server.

Index

• Constants
• Variables
• type Options
• type Server
  • func New() *Server
  • func NewServer(opts *Options) *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 indicates the current server version.
	Version = "1.1.1"
)

Variables

var (
	// ErrListenerIDExists indicates that a listener with the same id already exists.
	ErrListenerIDExists = errors.New("listener id already exists")

	// ErrReadConnectInvalid indicates that the connection packet was invalid.
	ErrReadConnectInvalid = errors.New("connect packet was not valid")

	// ErrConnectNotAuthorized indicates that the connection packet had incorrect auth values.
	ErrConnectNotAuthorized = errors.New("connect packet was not authorized")

	// ErrInvalidTopic indicates that the specified topic was not valid.
	ErrInvalidTopic = errors.New("cannot publish to $ and $SYS topics")

	// ErrRejectPacket indicates that a packet should be dropped instead of processed.
	ErrRejectPacket = errors.New("packet rejected")

	// ErrClientDisconnect indicates that a client disconnected from the server.
	ErrClientDisconnect = errors.New("client disconnected")

	// ErrClientReconnect indicates that a client attempted to reconnect while still connected.
	ErrClientReconnect = errors.New("client sent connect while connected")

	// ErrServerShutdown is propagated when the server shuts down.
	ErrServerShutdown = errors.New("server is shutting down")

	// ErrSessionReestablished indicates that an existing client was replaced by a newly connected
	// client. The existing client is disconnected.
	ErrSessionReestablished = errors.New("client session re-established")

	// ErrConnectionFailed indicates that a client connection attempt failed for other reasons.
	ErrConnectionFailed = errors.New("connection attempt failed")

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

Types

type Options

type Options struct {
	// BufferSize overrides the default buffer size (circ.DefaultBufferSize) for the client buffers.
	BufferSize int

	// BufferBlockSize overrides the default buffer block size (DefaultBlockSize) for the client buffers.
	BufferBlockSize int

	// InflightTTL specifies the duration that a queued inflight message should exist before being purged.
	InflightTTL int64
}

Options contains configurable options for the server.

type Server

type Server struct {
	Events    events.Events        // overrideable event hooks.
	Store     persistence.Store    // a persistent storage backend if desired.
	Options   *Options             // configurable server options.
	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.
	// 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 MQTT server with no options. This method has been deprecated and will be removed in a future release. Please use NewServer instead.

func NewServer

func NewServer(opts *Options) *Server

NewServer returns a new instance of an MQTT broker with optional values where applicable.

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.