server

Documentation

Overview

packet server provides a MQTT 3.1.1 & 5.0 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) CleanSession(cl *clients.Client)
  • 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) Serve() error

Constants

const (
	// Version indicates the current server version.
	Version = "1.2.0"

	// Single The running environment is single node or cluster mode
	Single uint = iota
	Cluster
)

const (
	StorageDel = iota
	StorageAdd
)

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
	// authentication parameters.
	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     = errors.New("Client disconnected")
	ErrClientReconnect      = errors.New("Client attemped to reconnect")
	ErrServerShutdown       = errors.New("Server is shutting down")
	ErrSessionReestablished = errors.New("Session reestablished")
	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 {
	// RunMode program running mode，1 single or 2 cluster
	RunMode uint

	// 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

	// ReceiveMaximum is the maximum number of QOS1 & 2 messages allowed to be "inflight"
	ReceiveMaximum int

	// InflightHandling is the handling mode of inflight message when the receive-maximum is exceeded, 0 closes the connection or 1 overwrites the old inflight message
	InflightHandling int
}

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) CleanSession

func (s *Server) CleanSession(cl *clients.Client)

CleanSession

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) 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.