mqtt

package module
v2.0.0-...-9ffe12e Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Feb 1, 2023 License: MIT Imports: 22 Imported by: 0

README ¶

build status Coverage Status Go Report Card Go Reference contributions welcome

Mochi MQTT Broker

The fully compliant, embeddable high-performance Go MQTT v5 (and v3.1.1) broker server

Mochi MQTT is an embeddable fully compliant MQTT v5 broker server written in Go, designed for the development of telemetry and internet-of-things projects. The server can be used either as a standalone binary or embedded as a library in your own applications, and has been designed to be as lightweight and fast as possible, with great care taken to ensure the quality and maintainability of the project.

What is MQTT?

MQTT stands for MQ Telemetry Transport. It is a publish/subscribe, extremely simple and lightweight messaging protocol, designed for constrained devices and low-bandwidth, high-latency or unreliable networks (Learn more). Mochi MQTT fully implements version 5.0.0 of the MQTT protocol.

What's new in Version 2.0.0?

Version 2.0.0 takes all the great things we loved about Mochi MQTT v1.0.0, learns from the mistakes, and improves on the things we wished we'd had. It's a total from-scratch rewrite, designed to fully implement MQTT v5 as a first-class feature.

Don't forget to use the new v2 import paths:

import "github.com/mignatovich/mqtt/v2"
  • Full MQTTv5 Feature Compliance, compatibility for MQTT v3.1.1 and v3.0.0:
    • User and MQTTv5 Packet Properties
    • Topic Aliases
    • Shared Subscriptions
    • Subscription Options and Subscription Identifiers
    • Message Expiry
    • Client Session Expiry
    • Send and Receive QoS Flow Control Quotas
    • Server-side Disconnect and Auth Packets
    • Will Delay Intervals
    • Plus all the original MQTT features of Mochi MQTT v1, such as Full QoS(0,1,2), $SYS topics, retained messages, etc.
  • Developer-centric:
    • Most core broker code is now exported and accessible, for total developer control.
    • Full featured and flexible Hook-based interfacing system to provide easy 'plugin' development.
    • Direct Packet Injection using special inline client, or masquerade as existing clients.
  • Performant and Stable:
    • Our classic trie-based Topic-Subscription model.
    • A new fixed 'FanPool' worker queues to ensure consistent resource allocation and throughput reliability.
    • Passes all Paho Interoperability Tests for MQTT v5 and MQTT v3.
    • Over a thousand carefully considered unit test scenarios.
  • TCP, Websocket, (including SSL/TLS) and $SYS Dashboard listeners.
  • Built-in Redis, Badger, and Bolt Persistence using Hooks (but you can also make your own).
  • Built-in Rule-based Authentication and ACL Ledger using Hooks (also make your own).

There is no upgrade path from v1.0.0. Please review the documentation and this readme to get a sense of the changes required (e.g. the v1 events system, auth, and persistence have all been replaced with the new hooks system).

Compatibility Notes

Because of the overlap between the v5 specification and previous versions of mqtt, the server can accept both v5 and v3 clients, but note that in cases where both v5 an v3 clients are connected, properties and features provided for v5 clients will be downgraded for v3 clients (such as user properties).

Support for MQTT v3.0.0 and v3.1.1 is considered hybrid-compatibility. Where not specifically restricted in the v3 specification, more modern and safety-first v5 behaviours are used instead - such as expiry for inflight and retained messages, and clients - and quality-of-service flow control limits.

Roadmap

  • Please open an issue to request new features or event hooks!
  • Cluster support.
  • Enhanced Metrics support.
  • File-based server configuration (supporting docker).

Quick Start

Running the Broker with Go

Mochi MQTT can be used as a standalone broker. Simply checkout this repository and run the cmd/main.go entrypoint in the cmd folder which will expose tcp (:1883), websocket (:1882), and dashboard (:8080) listeners.

cd cmd
go build -o mqtt && ./mqtt
Using Docker

A simple Dockerfile is provided for running the cmd/main.go Websocket, TCP, and Stats server:

docker build -t mochi:latest .
docker run -p 1883:1883 -p 1882:1882 -p 8080:8080 mochi:latest

Developing with Mochi MQTT

Importing as a package

Importing Mochi MQTT as a package requires just a few lines of code to get started.

import (
  "log"

  "github.com/mignatovich/mqtt/v2"
  "github.com/mignatovich/mqtt/v2/hooks/auth"
  "github.com/mignatovich/mqtt/v2/listeners"
)

func main() {
  // Create the new MQTT Server.
  server := mqtt.New(nil)
  
  // Allow all connections.
  _ = server.AddHook(new(auth.AllowHook), nil)
  
  // Create a TCP listener on a standard port.
  tcp := listeners.NewTCP("t1", ":1883", nil)
  err := server.AddListener(tcp)
  if err != nil {
    log.Fatal(err)
  }
  
  err = server.Serve()
  if err != nil {
    log.Fatal(err)
  }
}

Examples of running the broker with various configurations can be found in the examples folder.

Network Listeners

The server comes with a variety of pre-packaged network listeners which allow the broker to accept connections on different protocols. The current listeners are:

Listener Usage
listeners.NewTCP A TCP listener
listeners.NewUnixSock A Unix Socket listener
listeners.NewWebsocket A Websocket listener
listeners.NewHTTPStats An HTTP $SYS info dashboard

Use the listeners.Listener interface to develop new listeners. If you do, please let us know!

A *listeners.Config may be passed to configure TLS.

Examples of usage can be found in the examples folder or cmd/main.go.

Server Options and Capabilities

A number of configurable options are available which can be used to alter the behaviour or restrict access to certain features in the server.

server := mqtt.New(&mqtt.Options{
  Capabilities: mqtt.Capabilities{
    MaximumSessionExpiryInterval: 3600,
    Compatibilities: mqtt.Compatibilities{
      ObscureNotAuthorized: true,
    },
  },
  SysTopicResendInterval: 10,
})

Review the mqtt.Options, mqtt.Capabilities, and mqtt.Compatibilities structs for a comprehensive list of options.

Event Hooks

A universal event hooks system allows developers to hook into various parts of the server and client life cycle to add and modify functionality of the broker. These universal hooks are used to provide everything from authentication, persistent storage, to debugging tools.

Hooks are stackable - you can add multiple hooks to a server, and they will be run in the order they were added. Some hooks modify values, and these modified values will be passed to the subsequent hooks before being returned to the runtime code.

Type Import Info
Access Control mochi-co/mqtt/hooks/auth . AllowHook Allow access to all connecting clients and read/write to all topics.
Access Control mochi-co/mqtt/hooks/auth . Auth Rule-based access control ledger.
Persistence mochi-co/mqtt/hooks/storage/bolt Persistent storage using BoltDB (deprecated).
Persistence mochi-co/mqtt/hooks/storage/badger Persistent storage using BadgerDB.
Persistence mochi-co/mqtt/hooks/storage/redis Persistent storage using Redis.
Debugging mochi-co/mqtt/hooks/debug Additional debugging output to visualise packet flow.

Many of the internal server functions are now exposed to developers, so you can make your own Hooks by using the above as examples. If you do, please Open an issue and let everyone know!

Access Control
Allow Hook

By default, Mochi MQTT uses a DENY-ALL access control rule. To allow connections, this must overwritten using an Access Control hook. The simplest of these hooks is the auth.AllowAll hook, which provides ALLOW-ALL rules to all connections, subscriptions, and publishing. It's also the simplest hook to use:

server := mqtt.New(nil)
_ = server.AddHook(new(auth.AllowHook), nil)

Don't do this if you are exposing your server to the internet or untrusted networks - it should really be used for development, testing, and debugging only.

Auth Ledger

The Auth Ledger hook provides a sophisticated mechanism for defining access rules in a struct format. Auth ledger rules come in two forms: Auth rules (connection), and ACL rules (publish subscribe).

Auth rules have 4 optional criteria and an assertion flag:

Criteria Usage
Client client id of the connecting client
Username username of the connecting client
Password password of the connecting client
Remote the remote address or ip of the client
Allow true (allow this user) or false (deny this user)

ACL rules have 3 optional criteria and an filter match:

Criteria Usage
Client client id of the connecting client
Username username of the connecting client
Remote the remote address or ip of the client
Filters an array of filters to match

Rules are processed in index order (0,1,2,3), returning on the first matching rule. See hooks/auth/ledger.go to review the structs.

server := mqtt.New(nil)
err := server.AddHook(new(auth.Hook), &auth.Options{
    Ledger: &auth.Ledger{
    Auth: auth.AuthRules{ // Auth disallows all by default
      {Username: "peach", Password: "password1", Allow: true},
      {Username: "melon", Password: "password2", Allow: true},
      {Remote: "127.0.0.1:*", Allow: true},
      {Remote: "localhost:*", Allow: true},
    },
    ACL: auth.ACLRules{ // ACL allows all by default
      {Remote: "127.0.0.1:*"}, // local superuser allow all
      {
        // user melon can read and write to their own topic
        Username: "melon", Filters: auth.Filters{
          "melon/#":   auth.ReadWrite,
          "updates/#": auth.WriteOnly, // can write to updates, but can't read updates from others
        },
      },
      {
        // Otherwise, no clients have publishing permissions
        Filters: auth.Filters{
          "#":         auth.ReadOnly,
          "updates/#": auth.Deny,
        },
      },
    },
  }
})

The ledger can also be stored as JSON or YAML and loaded using the Data field:

err = server.AddHook(new(auth.Hook), &auth.Options{
    Data: data, // build ledger from byte slice: yaml or json
})

See examples/auth/encoded/main.go for more information.

Persistent Storage
Redis

A basic Redis storage hook is available which provides persistence for the broker. It can be added to the server in the same fashion as any other hook, with several options. It uses github.com/go-redis/redis/v8 under the hook, and is completely configurable through the Options value.

err := server.AddHook(new(redis.Hook), &redis.Options{
  Options: &rv8.Options{
    Addr:     "localhost:6379", // default redis address
    Password: "",               // your password
    DB:       0,                // your redis db
  },
})
if err != nil {
  log.Fatal(err)
}

For more information on how the redis hook works, or how to use it, see the examples/persistence/redis/main.go or hooks/storage/redis code.

Badger DB

There's also a BadgerDB storage hook if you prefer file based storage. It can be added and configured in much the same way as the other hooks (with somewhat less options).

err := server.AddHook(new(badger.Hook), &badger.Options{
  Path: badgerPath,
})
if err != nil {
  log.Fatal(err)
}

For more information on how the badger hook works, or how to use it, see the examples/persistence/badger/main.go or hooks/storage/badger code.

There is also a BoltDB hook which has been deprecated in favour of Badger, but if you need it, check examples/persistence/bolt/main.go.

Developing with Event Hooks

Many hooks are available for interacting with the broker and client lifecycle. The function signatures for all the hooks and mqtt.Hook interface can be found in hooks.go.

The most flexible event hooks are OnPacketRead, OnPacketEncode, and OnPacketSent - these hooks be used to control and modify all incoming and outgoing packets.

Function Usage
OnStarted Called when the server has successfully started.
OnStopped Called when the server has successfully stopped.
OnConnectAuthenticate Called when a user attempts to authenticate with the server. An implementation of this method MUST be used to allow or deny access to the server (see hooks/auth/allow_all or basic). It can be used in custom hooks to check connecting users against an existing user database. Returns true if allowed.
OnACLCheck Called when a user attempts to publish or subscribe to a topic filter. As above.
OnSysInfoTick Called when the $SYS topic values are published out.
OnConnect Called when a new client connects
OnSessionEstablished Called when a new client successfully establishes a session (after OnConnect)
OnDisconnect Called when a client is disconnected for any reason.
OnAuthPacket Called when an auth packet is received. It is intended to allow developers to create their own mqtt v5 Auth Packet handling mechanisms. Allows packet modification.
OnPacketRead Called when a packet is received from a client. Allows packet modification.
OnPacketEncode Called immediately before a packet is encoded to be sent to a client. Allows packet modification.
OnPacketSent Called when a packet has been sent to a client.
OnPacketProcessed Called when a packet has been received and successfully handled by the broker.
OnSubscribe Called when a client subscribes to one or more filters. Allows packet modification.
OnSubscribed Called when a client successfully subscribes to one or more filters.
OnSelectSubscribers Called when subscribers have been collected for a topic, but before shared subscription subscribers have been selected. Allows receipient modification.
OnUnsubscribe Called when a client unsubscribes from one or more filters. Allows packet modification.
OnUnsubscribed Called when a client successfully unsubscribes from one or more filters.
OnPublish Called when a client publishes a message. Allows packet modification.
OnPublished Called when a client has published a message to subscribers.
OnRetainMessage Called then a published message is retained.
OnQosPublish Called when a publish packet with Qos >= 1 is issued to a subscriber.
OnQosComplete Called when the Qos flow for a message has been completed.
OnQosDropped Called when an inflight message expires before completion.
OnWill Called when a client disconnects and intends to issue a will message. Allows packet modification.
OnWillSent Called when an LWT message has been issued from a disconnecting client.
OnClientExpired Called when a client session has expired and should be deleted.
OnRetainedExpired Called when a retained message has expired and should be deleted.
StoredClients Returns clients, eg. from a persistent store.
StoredSubscriptions Returns client subscriptions, eg. from a persistent store.
StoredInflightMessages Returns inflight messages, eg. from a persistent store.
StoredRetainedMessages Returns retained messages, eg. from a persistent store.
StoredSysInfo Returns stored system info values, eg. from a persistent store.

If you are building a persistent storage hook, see the existing persistent hooks for inspiration and patterns. If you are building an auth hook, you will need OnACLCheck and OnConnectAuthenticate.

Direct Publish

To publish basic message to a topic from within the embedding application, you can use the server.Publish(topic string, payload []byte, retain bool, qos byte) error method.

err := server.Publish("direct/publish", []byte("packet scheduled message"), false, 0)

The Qos byte in this case is only used to set the upper qos limit available for subscribers, as per MQTT v5 spec.

Packet Injection

If you want more control, or want to set specific MQTT v5 properties and other values you can create your own publish packets from a client of your choice. This method allows you to inject MQTT packets (no just publish) directly into the runtime as though they had been received by a specific client. Most of the time you'll want to use the special client flag inline=true, as it has unique privileges: it bypasses all ACL and topic validation checks, meaning it can even publish to $SYS topics.

Packet injection can be used for any MQTT packet, including ping requests, subscriptions, etc. And because the Clients structs and methods are now exported, you can even inject packets on behalf of a connected client (if you have a very custom requirements).

cl := server.NewClient(nil, "local", "inline", true)
server.InjectPacket(cl, packets.Packet{
  FixedHeader: packets.FixedHeader{
    Type: packets.Publish,
  },
  TopicName: "direct/publish",
  Payload: []byte("scheduled message"),
})

MQTT packets still need to be correctly formed, so refer our the test packets catalogue and MQTTv5 Specification for inspiration.

See the hooks example to see this feature in action.

Testing
Unit Tests

Mochi MQTT tests over a thousand scenarios with thoughtfully hand written unit tests to ensure each function does exactly what we expect. You can run the tests using go:

go run --cover ./...
Paho Interoperability Test

You can check the broker against the Paho Interoperability Test by starting the broker using examples/paho/main.go, and then running the mqtt v5 and v3 tests with python3 client_test5.py from the interoperability folder.

Note that there are currently a number of outstanding issues regarding false negatives in the paho suite, and as such, certain compatibility modes are enabled in the paho/main.go example.

Performance Benchmarks

Mochi MQTT performance is comparable with popular brokers such as Mosquitto, EMQX, and others.

Performance benchmarks were tested using MQTT-Stresser on a Apple Macbook Air M2, using cmd/main.go default settings. Taking into account bursts of high and low throughput, the median scores are the most useful. Higher is better.

The values presented in the benchmark are not representative of true messages per second throughput. They rely on an unusual calculation by mqtt-stresser, but are usable as they are consistent across all brokers. Benchmarks are provided as a general performance expectation guideline only.

mqtt-stresser -broker tcp://localhost:1883 -num-clients=2 -num-messages=10000

Broker publish fastest median slowest receive fastest median slowest
Mochi v2.0.0 139,860 135,960 132,059 217,499 211,027 204,555
Mosquitto v2.0.15 155,920 155,919 155,918 185,485 185,097 184,709
EMQX v5.0.11 156,945 156,257 155,568 17,918 17,783 17649

mqtt-stresser -broker tcp://localhost:1883 -num-clients=10 -num-messages=10000

Broker publish fastest median slowest receive fastest median slowest
Mochi v2.0.0 55,189 34,840 21,298 56,980 28,557 23,781
Mosquitto v2.0.15 42,729 38,633 29,879 23,241 19,714 18,806
EMQX v5.0.11 21,553 17,418 14,356 4,257 3,980 3756

Million Message Challenge (hit the server with 1 million messages immediately):

mqtt-stresser -broker tcp://localhost:1883 -num-clients=100 -num-messages=10000

Broker publish fastest median slowest receive fastest median slowest
Mochi v2.0.0 13,573 3,678 1,848 34,309 2,470 5,636
Mosquitto v2.0.15 3,826 3,395 3,032 1,200 1,150 1,118
EMQX v5.0.11 4,086 2,432 2,274 434 333 311

Not sure what's going on with EMQX here, perhaps the docker out-of-the-box settings are not optimal, so take it with a pinch of salt as we know for a fact it's a solid piece of software.

Stargazers over time 🥰

Stargazers over time Are you using Mochi MQTT in a project? Let us know!

Contributions

Contributions and feedback are both welcomed and encouraged! Open an issue to report a bug, ask a question, or make a feature request.

Documentation ¶

Overview ¶

package mqtt provides a high performance, fully compliant MQTT v5 broker server with v3.1.1 backward compatibility.

Index ¶

Constants ¶

View Source 
const (
	SetOptions byte = iota
	OnSysInfoTick
	OnStarted
	OnStopped
	OnConnectAuthenticate
	OnACLCheck
	OnConnect
	OnSessionEstablished
	OnDisconnect
	OnAuthPacket
	OnPacketRead
	OnPacketEncode
	OnPacketSent
	OnPacketProcessed
	OnSubscribe
	OnSubscribed
	OnSelectSubscribers
	OnUnsubscribe
	OnUnsubscribed
	OnPublish
	OnPublished
	OnRetainMessage
	OnQosPublish
	OnQosComplete
	OnQosDropped
	OnWill
	OnWillSent
	OnClientExpired
	OnRetainedExpired
	StoredClients
	StoredSubscriptions
	StoredInflightMessages
	StoredRetainedMessages
	StoredSysInfo
)
View Source 
const (
	Version = "2.1.6" // the current server version.

)

Variables ¶

View Source 
var (
	// DefaultServerCapabilities defines the default features and capabilities provided by the server.
	DefaultServerCapabilities = &Capabilities{
		MaximumSessionExpiryInterval: math.MaxUint32,
		MaximumMessageExpiryInterval: 60 * 60 * 24,
		ReceiveMaximum:               1024,
		MaximumQos:                   2,
		RetainAvailable:              1,
		MaximumPacketSize:            0,
		TopicAliasMaximum:            math.MaxUint16,
		WildcardSubAvailable:         1,
		SubIDAvailable:               1,
		SharedSubAvailable:           1,
		ServerKeepAlive:              10,
		MinimumProtocolVersion:       3,
	}

	ErrListenerIDExists = errors.New("listener id already exists") // a listener with the same id already exists.
	ErrConnectionClosed = errors.New("connection not open")        // connection is closed
)
View Source 
var (
	SharePrefix = "$SHARE" // the prefix indicating a share topic
	SysPrefix   = "$SYS"   // the prefix indicating a system info topic
)
View Source 
var (
	// ErrInvalidConfigType indicates a different Type of config value was expected to what was received.
	ErrInvalidConfigType = errors.New("invalid config type provided")
)

Functions ¶

func AtomicItoa ¶ 

func AtomicItoa(ptr *int64) string

AtomicItoa converts an int64 point to a string.

func IsSharedFilter ¶ 

func IsSharedFilter(filter string) bool

IsSharedFilter returns true if the filter uses the share prefix.

func IsValidFilter ¶ 

func IsValidFilter(filter string, forPublish bool) bool

IsValidFilter returns true if the filter is valid.

Types ¶

type Capabilities ¶ 

type Capabilities struct {
	MaximumMessageExpiryInterval int64
	MaximumSessionExpiryInterval uint32
	MaximumPacketSize            uint32
	ReceiveMaximum               uint16
	TopicAliasMaximum            uint16
	ServerKeepAlive              uint16
	SharedSubAvailable           byte
	MinimumProtocolVersion       byte
	Compatibilities              Compatibilities
	MaximumQos                   byte
	RetainAvailable              byte
	WildcardSubAvailable         byte
	SubIDAvailable               byte
}

Capabilities indicates the capabilities and features provided by the server.

type Client ¶ 

type Client struct {
	Properties ClientProperties // client properties
	State      ClientState      // the operational state of the client.
	Net        ClientConnection // network connection state of the clinet
	ID         string           // the client id.

	sync.RWMutex // mutex
	// contains filtered or unexported fields
}

Client contains information about a client known by the broker.

func (*Client) ClearInflights ¶ 

func (cl *Client) ClearInflights(now, maximumExpiry int64) []uint16

ClearInflights deletes all inflight messages for the client, eg. for a disconnected user with a clean session.

func (*Client) Closed ¶ 

func (cl *Client) Closed() bool

Closed returns true if client connection is closed.

func (*Client) NextPacketID ¶ 

func (cl *Client) NextPacketID() (i uint32, err error)

NextPacketID returns the next available (unused) packet id for the client. If no unused packet ids are available, an error is returned and the client should be disconnected.

func (*Client) ParseConnect ¶ 

func (cl *Client) ParseConnect(lid string, pk packets.Packet)

ParseConnect parses the connect parameters and properties for a client.

func (*Client) Read ¶ 

func (cl *Client) Read(packetHandler ReadFn) error

Read reads incoming packets from the connected client and transforms them into packets to be handled by the packetHandler.

func (*Client) ReadFixedHeader ¶ 

func (cl *Client) ReadFixedHeader(fh *packets.FixedHeader) error

ReadFixedHeader reads in the values of the next packet's fixed header.

func (*Client) ReadPacket ¶ 

func (cl *Client) ReadPacket(fh *packets.FixedHeader) (pk packets.Packet, err error)

ReadPacket reads the remaining buffer into an MQTT packet.

func (*Client) ResendInflightMessages ¶ 

func (cl *Client) ResendInflightMessages(force bool) error

ResendInflightMessages attempts to resend any pending inflight messages to connected clients.

func (*Client) Stop ¶ 

func (cl *Client) Stop(err error)

Stop instructs the client to shut down all processing goroutines and disconnect.

func (*Client) StopCause ¶ 

func (cl *Client) StopCause() error

StopCause returns the reason the client connection was stopped, if any.

func (*Client) WritePacket ¶ 

func (cl *Client) WritePacket(pk packets.Packet) error

WritePacket encodes and writes a packet to the client.

type ClientConnection ¶ 

type ClientConnection struct {
	Conn net.Conn // the net.Conn used to establish the connection

	Remote   string // the remote address of the client
	Listener string // listener id of the client
	Inline   bool   // client is an inline programmetic client
	// contains filtered or unexported fields
}

ClientConnection contains the connection transport and metadata for the client.

type ClientProperties ¶ 

type ClientProperties struct {
	Props           packets.Properties
	Will            Will
	Username        []byte
	ProtocolVersion byte
	Clean           bool
}

ClientProperties contains the properties which define the client behaviour.

type ClientState ¶ 

type ClientState struct {
	TopicAliases TopicAliases // a map of topic aliases

	Inflight      *Inflight      // a map of in-flight qos messages
	Subscriptions *Subscriptions // a map of the subscription filters a client maintains
	// contains filtered or unexported fields
}

State tracks the state of the client.

type ClientSubscriptions ¶ 

type ClientSubscriptions map[string]packets.Subscription

ClientSubscriptions is a map of aggregated subscriptions for a client.

type Clients ¶ 

type Clients struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

Clients contains a map of the clients known by the broker.

func NewClients ¶ 

func NewClients() *Clients

NewClients returns an instance of Clients.

func (*Clients) Add ¶ 

func (cl *Clients) Add(val *Client)

Add adds a new client to the clients map, keyed on client id.

func (*Clients) Delete ¶ 

func (cl *Clients) Delete(id string)

Delete removes a client from the internal map.

func (*Clients) Get ¶ 

func (cl *Clients) Get(id string) (*Client, bool)

Get returns the value of a client if it exists.

func (*Clients) GetAll ¶ 

func (cl *Clients) GetAll() map[string]*Client

GetAll returns all the clients.

func (*Clients) GetByListener ¶ 

func (cl *Clients) GetByListener(id string) []*Client

GetByListener returns clients matching a listener id.

func (*Clients) Len ¶ 

func (cl *Clients) Len() int

Len returns the length of the clients map.

type Compatibilities ¶ 

type Compatibilities struct {
	ObscureNotAuthorized     bool // return unspecified errors instead of not authorized
	PassiveClientDisconnect  bool // don't disconnect the client forcefully after sending disconnect packet (paho)
	AlwaysReturnResponseInfo bool // always return response info (useful for testing)
	RestoreSysInfoOnRestart  bool // restore system info from store as if server never stopped
}

Compatibilities provides flags for using compatibility modes.

type FanPool ¶ 

type FanPool struct {
	Mutex sync.Mutex
	// contains filtered or unexported fields
}

FanPool is a fixed-sized fan-style worker pool with multiple working 'columns'. Instead of a single queue channel processed by many goroutines, this fan pool uses many queue channels each processed by a single goroutine. Very special thanks are given to the authors of HMQ in particular @chowyu08 and @muXxer for their work on the fixpool worker pool https://github.com/fhmq/hmq/blob/master/pool/fixpool.go from which this fan-pool is heavily inspired.

func NewFanPool ¶ 

func NewFanPool(fanSize, queueSize uint64) *FanPool

New returns a new instance of FanPool. fanSize controls the number of 'columns' of the fan, whereas queueSize controls the size of each column's queue.

func (*FanPool) Close ¶ 

func (p *FanPool) Close()

Close issues a shutdown signal to the workers.

func (*FanPool) Enqueue ¶ 

func (p *FanPool) Enqueue(id string, task func())

Enqueue adds a new task to the queue to be processed.

func (*FanPool) Size ¶ 

func (p *FanPool) Size() uint64

Size returns the current number of workers in the pool.

func (*FanPool) Wait ¶ 

func (p *FanPool) Wait()

Wait blocks until all the workers in the pool have completed.

type Hook ¶ 

type Hook interface {
	ID() string
	Provides(b byte) bool
	Init(config interface{}) error
	Stop() error
	SetOpts(l *zerolog.Logger, o *HookOptions)
	OnStarted()
	OnStopped()
	OnConnectAuthenticate(cl *Client, pk packets.Packet) bool
	OnACLCheck(cl *Client, topic string, write bool) bool
	OnSysInfoTick(*system.Info, map[string]string)
	OnConnect(cl *Client, pk packets.Packet) bool
	OnSessionEstablished(cl *Client, pk packets.Packet)
	OnDisconnect(cl *Client, err error, expire bool)
	OnAuthPacket(cl *Client, pk packets.Packet) (packets.Packet, error)
	OnPacketRead(cl *Client, pk packets.Packet) (packets.Packet, error) // triggers when a new packet is received by a client, but before packet validation
	OnPacketEncode(cl *Client, pk packets.Packet) packets.Packet        // modify a packet before it is byte-encoded and written to the client
	OnPacketSent(cl *Client, pk packets.Packet, b []byte)               // triggers when packet bytes have been written to the client
	OnPacketProcessed(cl *Client, pk packets.Packet, err error)         // triggers after a packet from the client been processed (handled)
	OnSubscribe(cl *Client, pk packets.Packet) packets.Packet
	OnSubscribed(cl *Client, pk packets.Packet, reasonCodes []byte)
	OnSelectSubscribers(subs *Subscribers, pk packets.Packet) *Subscribers
	OnUnsubscribe(cl *Client, pk packets.Packet) packets.Packet
	OnUnsubscribed(cl *Client, pk packets.Packet)
	OnPublish(cl *Client, pk packets.Packet) (packets.Packet, error)
	OnPublished(cl *Client, pk packets.Packet)
	OnRetainMessage(cl *Client, pk packets.Packet, r int64)
	OnQosPublish(cl *Client, pk packets.Packet, sent int64, resends int)
	OnQosComplete(cl *Client, pk packets.Packet)
	OnQosDropped(cl *Client, pk packets.Packet)
	OnWill(cl *Client, will Will) (Will, error)
	OnWillSent(cl *Client, pk packets.Packet)
	OnClientExpired(cl *Client)
	OnRetainedExpired(filter string)
	StoredClients() ([]storage.Client, error)
	StoredSubscriptions() ([]storage.Subscription, error)
	StoredInflightMessages() ([]storage.Message, error)
	StoredRetainedMessages() ([]storage.Message, error)
	StoredSysInfo() (storage.SystemInfo, error)
}

Hook provides an interface of handlers for different events which occur during the lifecycle of the broker.

type HookBase ¶ 

type HookBase struct {
	Hook
	Log  *zerolog.Logger
	Opts *HookOptions
}

HookBase provides a set of default methods for each hook. It should be embedded in all hooks.

func (*HookBase) ID ¶ 

func (h *HookBase) ID() string

ID returns the ID of the hook.

func (*HookBase) Init ¶ 

func (h *HookBase) Init(config interface{}) error

Init performs any pre-start initializations for the hook, such as connecting to databases or opening files.

func (*HookBase) OnACLCheck ¶ 

func (h *HookBase) OnACLCheck(cl *Client, topic string, write bool) bool

OnACLCheck is called when a user attempts to subscribe or publish to a topic.

func (*HookBase) OnAuthPacket ¶ 

func (h *HookBase) OnAuthPacket(cl *Client, pk packets.Packet) (packets.Packet, error)

OnAuthPacket is called when an auth packet is received from the client.

func (*HookBase) OnClientExpired ¶ 

func (h *HookBase) OnClientExpired(cl *Client)

OnClientExpired is called when a client session has expired.

func (*HookBase) OnConnect ¶ 

func (h *HookBase) OnConnect(cl *Client, pk packets.Packet) bool

OnConnect is called when a new client connects.

func (*HookBase) OnConnectAuthenticate ¶ 

func (h *HookBase) OnConnectAuthenticate(cl *Client, pk packets.Packet) bool

OnConnectAuthenticate is called when a user attempts to authenticate with the server.

func (*HookBase) OnDisconnect ¶ 

func (h *HookBase) OnDisconnect(cl *Client, err error, expire bool)

OnDisconnect is called when a client is disconnected for any reason.

func (*HookBase) OnPacketEncode ¶ 

func (h *HookBase) OnPacketEncode(cl *Client, pk packets.Packet) packets.Packet

OnPacketEncode is called before a packet is byte-encoded and written to the client.

func (*HookBase) OnPacketProcessed ¶ 

func (h *HookBase) OnPacketProcessed(cl *Client, pk packets.Packet, err error)

OnPacketProcessed is called immediately after a packet from a client is processed.

func (*HookBase) OnPacketRead ¶ 

func (h *HookBase) OnPacketRead(cl *Client, pk packets.Packet) (packets.Packet, error)

OnPacketRead is called when a packet is received.

func (*HookBase) OnPacketSent ¶ 

func (h *HookBase) OnPacketSent(cl *Client, pk packets.Packet, b []byte)

OnPacketSent is called immediately after a packet is written to a client.

func (*HookBase) OnPublish ¶ 

func (h *HookBase) OnPublish(cl *Client, pk packets.Packet) (packets.Packet, error)

OnPublish is called when a client publishes a message.

func (*HookBase) OnPublished ¶ 

func (h *HookBase) OnPublished(cl *Client, pk packets.Packet)

OnPublished is called when a client has published a message to subscribers.

func (*HookBase) OnQosComplete ¶ 

func (h *HookBase) OnQosComplete(cl *Client, pk packets.Packet)

OnQosComplete is called when the Qos flow for a message has been completed.

func (*HookBase) OnQosDropped ¶ 

func (h *HookBase) OnQosDropped(cl *Client, pk packets.Packet)

OnQosDropped is called the Qos flow for a message expires.

func (*HookBase) OnQosPublish ¶ 

func (h *HookBase) OnQosPublish(cl *Client, pk packets.Packet, sent int64, resends int)

OnQosPublish is called when a publish packet with Qos > 1 is issued to a subscriber.

func (*HookBase) OnRetainMessage ¶ 

func (h *HookBase) OnRetainMessage(cl *Client, pk packets.Packet, r int64)

OnRetainMessage is called then a published message is retained.

func (*HookBase) OnRetainedExpired ¶ 

func (h *HookBase) OnRetainedExpired(topic string)

OnRetainedExpired is called when a retained message for a topic has expired.

func (*HookBase) OnSelectSubscribers ¶ 

func (h *HookBase) OnSelectSubscribers(subs *Subscribers, pk packets.Packet) *Subscribers

OnSelectSubscribers is called when selecting subscribers to receive a message.

func (*HookBase) OnSessionEstablished ¶ 

func (h *HookBase) OnSessionEstablished(cl *Client, pk packets.Packet)

OnSessionEstablished is called when a new client establishes a session (after OnConnect).

func (*HookBase) OnStarted ¶ 

func (h *HookBase) OnStarted()

OnStarted is called when the server starts.

func (*HookBase) OnStopped ¶ 

func (h *HookBase) OnStopped()

OnStopped is called when the server stops.

func (*HookBase) OnSubscribe ¶ 

func (h *HookBase) OnSubscribe(cl *Client, pk packets.Packet) packets.Packet

OnSubscribe is called when a client subscribes to one or more filters.

func (*HookBase) OnSubscribed ¶ 

func (h *HookBase) OnSubscribed(cl *Client, pk packets.Packet, reasonCodes []byte)

OnSubscribed is called when a client subscribes to one or more filters.

func (*HookBase) OnSysInfoTick ¶ 

func (h *HookBase) OnSysInfoTick(*system.Info, map[string]string)

OnSysInfoTick is called when the server publishes system info.

func (*HookBase) OnUnsubscribe ¶ 

func (h *HookBase) OnUnsubscribe(cl *Client, pk packets.Packet) packets.Packet

OnUnsubscribe is called when a client unsubscribes from one or more filters.

func (*HookBase) OnUnsubscribed ¶ 

func (h *HookBase) OnUnsubscribed(cl *Client, pk packets.Packet)

OnUnsubscribed is called when a client unsubscribes from one or more filters.

func (*HookBase) OnWill ¶ 

func (h *HookBase) OnWill(cl *Client, will Will) (Will, error)

OnWill is called when a client disconnects and publishes an LWT message.

func (*HookBase) OnWillSent ¶ 

func (h *HookBase) OnWillSent(cl *Client, pk packets.Packet)

OnWillSent is called when an LWT message has been issued from a disconnecting client.

func (*HookBase) Provides ¶ 

func (h *HookBase) Provides(b byte) bool

Provides indicates which methods a hook provides. The default is none - this method should be overridden by the embedding hook.

func (*HookBase) SetOpts ¶ 

func (h *HookBase) SetOpts(l *zerolog.Logger, opts *HookOptions)

SetOpts is called by the server to propagate internal values and generally should not be called manually.

func (*HookBase) Stop ¶ 

func (h *HookBase) Stop() error

Stop is called to gracefully shutdown the hook.

func (*HookBase) StoredClients ¶ 

func (h *HookBase) StoredClients() (v []storage.Client, err error)

StoredClients returns all clients from a store.

func (*HookBase) StoredInflightMessages ¶ 

func (h *HookBase) StoredInflightMessages() (v []storage.Message, err error)

StoredInflightMessages returns all inflight messages from a store.

func (*HookBase) StoredRetainedMessages ¶ 

func (h *HookBase) StoredRetainedMessages() (v []storage.Message, err error)

StoredRetainedMessages returns all retained messages from a store.

func (*HookBase) StoredSubscriptions ¶ 

func (h *HookBase) StoredSubscriptions() (v []storage.Subscription, err error)

StoredSubscriptions returns all subcriptions from a store.

func (*HookBase) StoredSysInfo ¶ 

func (h *HookBase) StoredSysInfo() (v storage.SystemInfo, err error)

StoredSysInfo returns a set of system info values.

type HookOptions ¶ 

type HookOptions struct {
	Capabilities *Capabilities
}

HookOptions contains values which are inherited from the server on initialisation.

type Hooks ¶ 

type Hooks struct {
	Log *zerolog.Logger // a logger for the hook (from the server)

	sync.Mutex // a mutex for locking when adding hooks
	// contains filtered or unexported fields
}

Hooks is a slice of Hook interfaces to be called in sequence.

func (*Hooks) Add ¶ 

func (h *Hooks) Add(hook Hook, config interface{}) error

Add adds and initializes a new hook.

func (*Hooks) GetAll ¶ 

func (h *Hooks) GetAll() []Hook

GetAll returns a slice of all the hooks.

func (*Hooks) Len ¶ 

func (h *Hooks) Len() int64

Len returns the number of hooks added.

func (*Hooks) OnACLCheck ¶ 

func (h *Hooks) OnACLCheck(cl *Client, topic string, write bool) bool

OnACLCheck is called when a user attempts to publish or subscribe to a topic filter. An implementation of this method MUST be used to allow or deny access to the (see hooks/auth/allow_all or basic). It can be used in custom hooks to check publishing and subscribing users against an existing permissions or roles database.

func (*Hooks) OnAuthPacket ¶ 

func (h *Hooks) OnAuthPacket(cl *Client, pk packets.Packet) (pkx packets.Packet, err error)

OnAuthPacket is called when an auth packet is received. It is intended to allow developers to create their own auth packet handling mechanisms.

func (*Hooks) OnClientExpired ¶ 

func (h *Hooks) OnClientExpired(cl *Client)

OnClientExpired is called when a client session has expired and should be deleted.

func (*Hooks) OnConnect ¶ 

func (h *Hooks) OnConnect(cl *Client, pk packets.Packet) bool

OnConnect is called when a new client connects.

func (*Hooks) OnConnectAuthenticate ¶ 

func (h *Hooks) OnConnectAuthenticate(cl *Client, pk packets.Packet) bool

OnConnectAuthenticate is called when a user attempts to authenticate with the server. An implementation of this method MUST be used to allow or deny access to the server (see hooks/auth/allow_all or basic). It can be used in custom hooks to check connecting users against an existing user database.

func (*Hooks) OnDisconnect ¶ 

func (h *Hooks) OnDisconnect(cl *Client, err error, expire bool)

OnDisconnect is called when a client is disconnected for any reason.

func (*Hooks) OnPacketEncode ¶ 

func (h *Hooks) OnPacketEncode(cl *Client, pk packets.Packet) packets.Packet

OnPacketEncode is called immediately before a packet is encoded to be sent to a client.

func (*Hooks) OnPacketProcessed ¶ 

func (h *Hooks) OnPacketProcessed(cl *Client, pk packets.Packet, err error)

OnPacketProcessed is called when a packet has been received and successfully handled by the broker.

func (*Hooks) OnPacketRead ¶ 

func (h *Hooks) OnPacketRead(cl *Client, pk packets.Packet) (pkx packets.Packet, err error)

OnPacketRead is called when a packet is received from a client.

func (*Hooks) OnPacketSent ¶ 

func (h *Hooks) OnPacketSent(cl *Client, pk packets.Packet, b []byte)

OnPacketSent is called when a packet has been sent to a client. It takes a bytes parameter containing the bytes sent.

func (*Hooks) OnPublish ¶ 

func (h *Hooks) OnPublish(cl *Client, pk packets.Packet) (pkx packets.Packet, err error)

OnPublish is called when a client publishes a message. This method differs from OnPublished in that it allows you to modify you to modify the incoming packet before it is processed. The return values of the hook methods are passed-through in the order the hooks were attached.

func (*Hooks) OnPublished ¶ 

func (h *Hooks) OnPublished(cl *Client, pk packets.Packet)

OnPublished is called when a client has published a message to subscribers.

func (*Hooks) OnQosComplete ¶ 

func (h *Hooks) OnQosComplete(cl *Client, pk packets.Packet)

OnQosComplete is called when the Qos flow for a message has been completed. In other words, when an inflight message is resolved. It is typically used to delete an inflight message from a store.

func (*Hooks) OnQosDropped ¶ 

func (h *Hooks) OnQosDropped(cl *Client, pk packets.Packet)

OnQosDropped is called the Qos flow for a message expires. In other words, when an inflight message expires or is abandoned. It is typically used to delete an inflight message from a store.

func (*Hooks) OnQosPublish ¶ 

func (h *Hooks) OnQosPublish(cl *Client, pk packets.Packet, sent int64, resends int)

OnQosPublish is called when a publish packet with Qos >= 1 is issued to a subscriber. In other words, this method is called when a new inflight message is created or resent. It is typically used to store a new inflight message.

func (*Hooks) OnRetainMessage ¶ 

func (h *Hooks) OnRetainMessage(cl *Client, pk packets.Packet, r int64)

OnRetainMessage is called then a published message is retained.

func (*Hooks) OnRetainedExpired ¶ 

func (h *Hooks) OnRetainedExpired(filter string)

OnRetainedExpired is called when a retained message has expired and should be deleted.

func (*Hooks) OnSelectSubscribers ¶ 

func (h *Hooks) OnSelectSubscribers(subs *Subscribers, pk packets.Packet) *Subscribers

OnSelectSubscribers is called when subscribers have been collected for a topic, but before shared subscription subscribers have been selected. This hook can be used to programmatically remove or add clients to a publish to subscribers process, or to select the subscriber for a shared group in a custom manner (such as based on client id, ip, etc).

func (*Hooks) OnSessionEstablished ¶ 

func (h *Hooks) OnSessionEstablished(cl *Client, pk packets.Packet)

OnSessionEstablished is called when a new client establishes a session (after OnConnect).

func (*Hooks) OnStarted ¶ 

func (h *Hooks) OnStarted()

OnStarted is called when the server has successfully started.

func (*Hooks) OnStopped ¶ 

func (h *Hooks) OnStopped()

OnStopped is called when the server has successfully stopped.

func (*Hooks) OnSubscribe ¶ 

func (h *Hooks) OnSubscribe(cl *Client, pk packets.Packet) packets.Packet

OnSubscribe is called when a client subscribes to one or more filters. This method differs from OnSubscribed in that it allows you to modify the subscription values before the packet is processed. The return values of the hook methods are passed-through in the order the hooks were attached.

func (*Hooks) OnSubscribed ¶ 

func (h *Hooks) OnSubscribed(cl *Client, pk packets.Packet, reasonCodes []byte)

OnSubscribed is called when a client subscribes to one or more filters.

func (*Hooks) OnSysInfoTick ¶ 

func (h *Hooks) OnSysInfoTick(sys *system.Info, topics map[string]string)

OnSysInfoTick is called when the $SYS topic values are published out.

func (*Hooks) OnUnsubscribe ¶ 

func (h *Hooks) OnUnsubscribe(cl *Client, pk packets.Packet) packets.Packet

OnUnsubscribe is called when a client unsubscribes from one or more filters. This method differs from OnUnsubscribed in that it allows you to modify the unsubscription values before the packet is processed. The return values of the hook methods are passed-through in the order the hooks were attached.

func (*Hooks) OnUnsubscribed ¶ 

func (h *Hooks) OnUnsubscribed(cl *Client, pk packets.Packet)

OnUnsubscribed is called when a client unsubscribes from one or more filters.

func (*Hooks) OnWill ¶ 

func (h *Hooks) OnWill(cl *Client, will Will) Will

OnWill is called when a client disconnects and publishes an LWT message. This method differs from OnWillSent in that it allows you to modify the LWT message before it is published. The return values of the hook methods are passed-through in the order the hooks were attached.

func (*Hooks) OnWillSent ¶ 

func (h *Hooks) OnWillSent(cl *Client, pk packets.Packet)

OnWillSent is called when an LWT message has been issued from a disconnecting client.

func (*Hooks) Provides ¶ 

func (h *Hooks) Provides(b ...byte) bool

Provides returns true if any one hook provides any of the requested hook methods.

func (*Hooks) Stop ¶ 

func (h *Hooks) Stop()

Stop indicates all attached hooks to gracefully end.

func (*Hooks) StoredClients ¶ 

func (h *Hooks) StoredClients() (v []storage.Client, err error)

StoredClients returns all clients, e.g. from a persistent store, is used to populate the server clients list before start.

func (*Hooks) StoredInflightMessages ¶ 

func (h *Hooks) StoredInflightMessages() (v []storage.Message, err error)

StoredInflightMessages returns all inflight messages, e.g. from a persistent store, and is used to populate the restored clients with inflight messages before start.

func (*Hooks) StoredRetainedMessages ¶ 

func (h *Hooks) StoredRetainedMessages() (v []storage.Message, err error)

StoredRetainedMessages returns all retained messages, e.g. from a persistent store, and is used to populate the server topics with retained messages before start.

func (*Hooks) StoredSubscriptions ¶ 

func (h *Hooks) StoredSubscriptions() (v []storage.Subscription, err error)

StoredSubscriptions returns all subcriptions, e.g. from a persistent store, and is used to populate the server subscriptions list before start.

func (*Hooks) StoredSysInfo ¶ 

func (h *Hooks) StoredSysInfo() (v storage.SystemInfo, err error)

StoredSysInfo returns a set of system info values.

type InboundTopicAliases ¶ 

type InboundTopicAliases struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

InboundTopicAliases contains a map of topic aliases received from the client.

func NewInboundTopicAliases ¶ 

func NewInboundTopicAliases(topicAliasMaximum uint16) *InboundTopicAliases

NewInboundTopicAliases returns a pointer to InboundTopicAliases.

func (*InboundTopicAliases) Set ¶ 

func (a *InboundTopicAliases) Set(id uint16, topic string) string

Set sets a new alias for a specific topic.

type Inflight ¶ 

type Inflight struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

Inflight is a map of InflightMessage keyed on packet id.

func NewInflights ¶ 

func NewInflights() *Inflight

NewInflights returns a new instance of an Inflight packets map.

func (*Inflight) Delete ¶ 

func (i *Inflight) Delete(id uint16) bool

Delete removes an in-flight message from the map. Returns true if the message existed.

func (*Inflight) Get ¶ 

func (i *Inflight) Get(id uint16) (packets.Packet, bool)

Get returns an inflight packet by packet id.

func (*Inflight) GetAll ¶ 

func (i *Inflight) GetAll(immediate bool) []packets.Packet

GetAll returns all the inflight messages.

func (*Inflight) Len ¶ 

func (i *Inflight) Len() int

Len returns the size of the inflight messages map.

func (*Inflight) NextImmediate ¶ 

func (i *Inflight) NextImmediate() (packets.Packet, bool)

NextImmediate returns the next inflight packet which is indicated to be sent immediately. This typically occurs when the quota has been exhausted, and we need to wait until new quota is free to continue sending.

func (*Inflight) ResetReceiveQuota ¶ 

func (i *Inflight) ResetReceiveQuota(n int32)

ResetReceiveQuota resets the receive quota to the maximum allowed value.

func (*Inflight) ResetSendQuota ¶ 

func (i *Inflight) ResetSendQuota(n int32)

ResetSendQuota resets the send quota to the maximum allowed value.

func (*Inflight) ReturnReceiveQuota ¶ 

func (i *Inflight) ReturnReceiveQuota()

TakeRecieveQuota increases the receive quota by 1.

func (*Inflight) ReturnSendQuota ¶ 

func (i *Inflight) ReturnSendQuota()

ReturnSendQuota increases the send quota by 1.

func (*Inflight) Set ¶ 

func (i *Inflight) Set(m packets.Packet) bool

Set adds or updates an inflight packet by packet id.

func (*Inflight) TakeReceiveQuota ¶ 

func (i *Inflight) TakeReceiveQuota()

TakeRecieveQuota reduces the receive quota by 1.

func (*Inflight) TakeSendQuota ¶ 

func (i *Inflight) TakeSendQuota()

TakeSendQuota reduces the send quota by 1.

type Options ¶ 

type Options struct {
	// Capabilities defines the server features and behaviour.
	Capabilities *Capabilities

	// Logger specifies a custom configured implementation of zerolog to override
	// the servers default logger configuration. If you wish to change the log level,
	// of the default logger, you can do so by setting
	// 	server := mqtt.New(nil)
	// 	l := server.Log.Level(zerolog.DebugLevel)
	// 	server.Log = &l
	Logger *zerolog.Logger

	// FanPoolSize is the number of individual workers and queues to initialize.
	// Bigger is not necessarily better, and you should rely on defaults unless
	// you have know what you are doing.
	FanPoolSize uint64

	// FanPoolQueueSize is the size of the queue per worker. Increase this value
	// accordingly if you anticipate having intermittent but massive numbers of
	// messages. Cluster support is roadmapped.
	FanPoolQueueSize uint64

	// SysTopicResendInterval specifies the interval between $SYS topic updates in seconds.
	SysTopicResendInterval int64
}

Options contains configurable options for the server.

type OutboundTopicAliases ¶ 

type OutboundTopicAliases struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

OutboundTopicAliases contains a map of topic aliases sent from the broker to the client.

func NewOutboundTopicAliases ¶ 

func NewOutboundTopicAliases(topicAliasMaximum uint16) *OutboundTopicAliases

NewOutboundTopicAliases returns a pointer to OutboundTopicAliases.

func (*OutboundTopicAliases) Set ¶ 

func (a *OutboundTopicAliases) Set(topic string) (uint16, bool)

Set sets a new topic alias for a topic and returns the alias value, and a boolean indicating if the alias already existed.

type ReadFn ¶ 

type ReadFn func(*Client, packets.Packet) error

ReadFn is the function signature for the function used for reading and processing new packets.

type Server ¶ 

type Server struct {
	Options   *Options             // configurable server options
	Listeners *listeners.Listeners // listeners are network interfaces which listen for new connections
	Clients   *Clients             // clients known to the broker
	Topics    *TopicsIndex         // an index of topic filter subscriptions and retained messages
	Info      *system.Info         // values about the server commonly known as $SYS topics

	Log *zerolog.Logger // minimal no-alloc logger
	// 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(opts *Options) *Server

New returns a new instance of mochi mqtt broker. Optional parameters can be specified to override some default settings (see Options).

func (*Server) AddHook ¶ 

func (s *Server) AddHook(hook Hook, config interface{}) error

AddHook attaches a new Hook to the server. Ideally, this should be called before the server is started with s.Serve().

func (*Server) AddListener ¶ 

func (s *Server) AddListener(l listeners.Listener) error

AddListener adds a new network listener to the server, for receiving incoming client connections.

func (*Server) Close ¶ 

func (s *Server) Close() error

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

func (*Server) DisconnectClient ¶ 

func (s *Server) DisconnectClient(cl *Client, code packets.Code) error

DisconnectClient sends a Disconnect packet to a client and then closes the client connection.

func (*Server) EstablishConnection ¶ 

func (s *Server) EstablishConnection(listener string, c net.Conn) error

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

func (*Server) InjectPacket ¶ 

func (s *Server) InjectPacket(cl *Client, pk packets.Packet) error

InjectPacket injects a packet into the broker as if it were sent from the specified client. InlineClients using this method can publish packets to any topic (including $SYS) and bypass ACL checks.

func (*Server) NewClient ¶ 

func (s *Server) NewClient(c net.Conn, listener string, id string, inline bool) *Client

NewClient returns a new Client instance, populated with all the required values and references to be used with the server. If you are using this client to directly publish messages from the embedding application, set the inline flag to true to bypass ACL and topic validation checks.

func (*Server) Publish ¶ 

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

Publish publishes a publish packet into the broker as if it were sent from the speicfied client. This is a convenience function which wraps InjectPacket. As such, this method can publish packets to any topic (including $SYS) and bypass ACL checks. The qos byte is used for limiting the outbound qos (mqtt v5) rather than issuing to the broker (we assume qos 2 complete).

func (*Server) Serve ¶ 

func (s *Server) Serve() error

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

func (*Server) UnsubscribeClient ¶ 

func (s *Server) UnsubscribeClient(cl *Client)

UnsubscribeClient unsubscribes a client from all of their subscriptions.

type SharedSubscriptions ¶ 

type SharedSubscriptions struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

SharedSubscriptions contains a map of subscriptions to a shared filter, keyed on share group then client id.

func NewSharedSubscriptions ¶ 

func NewSharedSubscriptions() *SharedSubscriptions

NewSharedSubscriptions returns a new instance of Subscriptions.

func (*SharedSubscriptions) Add ¶ 

func (s *SharedSubscriptions) Add(group, id string, val packets.Subscription)

Add creates a new shared subscription for a group and client id pair.

func (*SharedSubscriptions) Delete ¶ 

func (s *SharedSubscriptions) Delete(group, id string)

Delete deletes a client id from a shared subscription group.

func (*SharedSubscriptions) Get ¶ 

func (s *SharedSubscriptions) Get(group, id string) (val packets.Subscription, ok bool)

Get returns the subscription properties for a client id in a share group, if one exists.

func (*SharedSubscriptions) GetAll ¶ 

func (s *SharedSubscriptions) GetAll() map[string]map[string]packets.Subscription

GetAll returns all shared subscription groups and their subscriptions.

func (*SharedSubscriptions) GroupLen ¶ 

func (s *SharedSubscriptions) GroupLen() int

GroupLen returns the number of groups subscribed to the filter.

func (*SharedSubscriptions) Len ¶ 

func (s *SharedSubscriptions) Len() int

Len returns the total number of shared subscriptions to a filter across all groups.

type Subscribers ¶ 

type Subscribers struct {
	Shared         map[string]map[string]packets.Subscription
	SharedSelected map[string]packets.Subscription
	Subscriptions  map[string]packets.Subscription
}

Subscribers contains the shared and non-shared subscribers matching a topic.

func (*Subscribers) MergeSharedSelected ¶ 

func (s *Subscribers) MergeSharedSelected()

MergeSharedSelected merges the selected subscribers for a shared subscription group and the non-shared subscribers, to ensure that no subscriber gets multiple messages due to have both types of subscription matching the same filter.

func (*Subscribers) SelectShared ¶ 

func (s *Subscribers) SelectShared()

SelectShared returns one subscriber for each shared subscription group.

type Subscriptions ¶ 

type Subscriptions struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

Subscriptions is a map of subscriptions keyed on client.

func NewSubscriptions ¶ 

func NewSubscriptions() *Subscriptions

NewSubscriptions returns a new instance of Subscriptions.

func (*Subscriptions) Add ¶ 

func (s *Subscriptions) Add(id string, val packets.Subscription)

Add adds a new subscription for a client. ID can be a filter in the case this map is client state, or a client id if particle state.

func (*Subscriptions) Delete ¶ 

func (s *Subscriptions) Delete(id string)

Delete removes a subscription by client or filter id.

func (*Subscriptions) Get ¶ 

func (s *Subscriptions) Get(id string) (val packets.Subscription, ok bool)

Get returns a subscriptions for a specific client or filter id.

func (*Subscriptions) GetAll ¶ 

func (s *Subscriptions) GetAll() map[string]packets.Subscription

GetAll returns all subscriptions.

func (*Subscriptions) Len ¶ 

func (s *Subscriptions) Len() int

Len returns the number of subscriptions.

type TopicAliases ¶ 

type TopicAliases struct {
	Inbound  *InboundTopicAliases
	Outbound *OutboundTopicAliases
}

TopicAliases contains inbound and outbound topic alias registrations.

func NewTopicAliases ¶ 

func NewTopicAliases(topicAliasMaximum uint16) TopicAliases

NewTopicAliases returns an instance of TopicAliases.

type TopicsIndex ¶ 

type TopicsIndex struct {
	Retained *packets.Packets
	// contains filtered or unexported fields
}

TopicsIndex is a prefix/trie tree containing topic subscribers and retained messages.

func NewTopicsIndex ¶ 

func NewTopicsIndex() *TopicsIndex

NewTopicsIndex returns a pointer to a new instance of Index.

func (*TopicsIndex) Messages ¶ 

func (x *TopicsIndex) Messages(filter string) []packets.Packet

Messages returns a slice of any retained messages which match a filter.

func (*TopicsIndex) RetainMessage ¶ 

func (x *TopicsIndex) RetainMessage(pk packets.Packet) int64

RetainMessage saves a message payload to the end of a topic address. Returns 1 if a retained message was added, and -1 if the retained message was removed. 0 is returned if sequential empty payloads are received.

func (*TopicsIndex) Subscribe ¶ 

func (x *TopicsIndex) Subscribe(client string, subscription packets.Subscription) bool

Subscribe adds a new subscription for a client to a topic filter, returning true if the subscription was new.

func (*TopicsIndex) Subscribers ¶ 

func (x *TopicsIndex) Subscribers(topic string) *Subscribers

Subscribers returns a map of clients who are subscribed to matching filters, their subscription ids and highest qos.

func (*TopicsIndex) Unsubscribe ¶ 

func (x *TopicsIndex) Unsubscribe(filter, client string) bool

Unsubscribe removes a subscription filter for a client, returning true if the subscription existed.

type Will ¶ 

type Will struct {
	Payload           []byte                 // -
	User              []packets.UserProperty // -
	TopicName         string                 // -
	Flag              uint32                 // 0,1
	WillDelayInterval uint32                 // -
	Qos               byte                   // -
	Retain            bool                   // -
}

Will contains the last will and testament details for a client connection.

Source Files ¶

View all Source files

Directories ¶

Path Synopsis
examples
auth/basic
auth/encoded
debug
hooks
paho.testing
persistence/badger
persistence/bolt
persistence/redis
tcp
tls
websocket
hooks
auth
debug
storage
storage/badger
storage/bolt
SPDX-License-Identifier: MIT SPDX-FileCopyrightText: 2022 mochi-co SPDX-FileContributor: mochi-co package bolt is provided for historical compatibility and may not be actively updated, you should use the badger hook instead.
 SPDX-License-Identifier: MIT SPDX-FileCopyrightText: 2022 mochi-co SPDX-FileContributor: mochi-co package bolt is provided for historical compatibility and may not be actively updated, you should use the badger hook instead.
storage/redis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.