gott

README

An MQTT Broker written in pure Go. Aims to be a high-performance, easy-to-use pluggable broker with most of the features that could be embedded in a broker available out of the box (either embedded in the Broker itself or as plugins) at no cost.

Hopefully with your contribution we could build something great!

Project Status

In BETA

• GOTT is currently in a BETA stage. All planned features fully work as tested while developed. Still, there is room for improvement as more tests and optimizations are needed.

Planned for v1 (MQTT v3.1.1)

• Ping (client -> server)
• Topic filtering with wildcards support
• Subscriptions
• QoS 0 messages
• QoS 1 messages
• QoS 2 messages
• Retained messages
• Will messages
• Sessions
• Plugins
• Logging to disk (with levels and log rotation)
• TLS/SSL
• WebSockets

Planned for v2

• MQTT v5
• Clustering

Known Issues

• Restarting the broker will reset existing subscriptions. They are not saved to disk (sessions are saved to disk but subscriptions are not restored on broker restart).

Quick Start

1. Install dependencies:

$ go get github.com/google/uuid
$ go get github.com/dgraph-io/badger
$ go get github.com/json-iterator/go
$ go get go.uber.org/zap
$ go get gopkg.in/natefinch/lumberjack.v2
$ go get gopkg.in/yaml.v2
$ go get github.com/gorilla/websocket

1. Clone/download this repo and place it in $GOPATH/src.
2. Run cd main inside the project's directory.
3. Run go run main.go.

Alternatively, you can download and run the install script that will handle dependency installation and repo cloning for you.

Plugins

GOTT implements a plugin system that is very easy to work with. You can easily build your own plugin that does whatever you want.

Start by reading the plugins documentation.

License

Apache License 2.0, see LICENSE.

Contribution

You are very welcome to submit a new feature, fix a bug, an optimization to the code, report a bug or even a benchmark would be helpful.

To Contribute:

Open an issue or:

1. Fork this repo.
2. Create a new branch with a descriptive name (example: feature/some-new-function or fix/something-somewhere).
3. Commit and push your code to your new branch.
4. Create a new Pull Request here.

Documentation

Index

• Constants
• func NewLogger(cnf loggingConfig) *zap.Logger
• func Recover(callback func(err, stack string))
• func Retry(packetID uint16, msg *clientMessage)
• type Broker
  • func NewBroker() (*Broker, error)
  • func (b *Broker) Listen() error
  • func (b *Broker) Publish(topic, payload []byte, flags publishFlags) bool
  • func (b *Broker) PublishRetained(msg *message, sub *subscription)
  • func (b *Broker) PublishToClient(client *Client, packetID uint16, cm *clientMessage)
  • func (b *Broker) Retain(msg *message, topic []byte)
  • func (b *Broker) Subscribe(client *Client, filter []byte, qos byte) bool
  • func (b *Broker) Unsubscribe(client *Client, filter []byte) bool
  • func (b *Broker) UnsubscribeAll(client *Client)
• type Client
• type Config

Constants

const (
	ConnectAccepted byte = iota
	ConnectUnacceptableProto
	ConnectIDRejected
	ConnectServerUnavailable
	ConnectBadUsernamePassword
	ConnectNotAuthorized
)

Connect Ack return codes

const (
	TypeReserved = iota
	TypeConnect
	TypeConnAck
	TypePublish
	TypePubAck
	TypePubRec
	TypePubRel
	TypePubComp
	TypeSubscribe
	TypeSubAck
	TypeUnsubscribe
	TypeUnsubAck
	TypePingReq
	TypePingResp
	TypeDisconnect
)

Packet types.

const (
	ConnectRemLen       = 2 // 2 is constant remaining len as per [3.2.1]
	ConnectVarHeaderLen = 10
	PubackRemLen        = 2 // 2 is constant remaining len as per [3.4.1]
	PubrecRemLen        = 2 // 2 is constant remaining len as per [3.5.1]
	PubrelRemLen        = 2 // 2 is constant remaining len as per [3.6.1]
	PubcompRemLen       = 2 // 2 is constant remaining len as per [3.7.1]
	SubackRemLen        = 2 // 2 is constant remaining len as per [3.9.2]
	UnsubackRemLen      = 2 // 2 is constant remaining len as per [3.11.1]
	PingrespRemLen      = 0 // 0 is constant remaining len as per [3.13.1]
)

Remaining lengths according to the MQTT spec.

const (
	StatusUnacknowledged int32 = iota
	StatusPubackReceived
	StatusPubrecReceived
	StatusPubrelReceived
	StatusPubcompReceived
)

Application Message statuses.

const (
	SubackFailureCode byte = 128
)

Subscribe Ack return codes.

Functions

func NewLogger

func NewLogger(cnf loggingConfig) *zap.Logger

NewLogger initializes a new zap.Logger with lumberjack support to write to log files with rotation.

func Recover

func Recover(callback func(err, stack string))

Recover calls built-in recover function. Can call a provided callback function passing in the error reported and the stack of it.

func Retry

func Retry(packetID uint16, msg *clientMessage)

Retry will check for a msg's status and resend it if it hasn't been acknowledged within 20 seconds.

Types

type Broker

type Broker struct {
	TopicFilterStorage *topicStorage
	MessageStore       *messageStore
	SessionStore       *sessionStore
	// contains filtered or unexported fields
}

Broker is the main broker struct. Should not be used directly. Use the global GOTT var instead.

var GOTT *Broker

GOTT is a singleton used to save memory instead of keeping a reference inside each Client

func NewBroker

func NewBroker() (*Broker, error)

NewBroker initializes a new object of type Broker. You can either use the returned pointer or the global GOTT var. Returns a pointer of type Broker which is also assigned to the global GOTT var and an error. It creates/opens an on-disk session store.

func (*Broker) Listen

func (b *Broker) Listen() error

Listen starts the broker and listens for new connections on the address provided in the config file.

func (*Broker) Publish

func (b *Broker) Publish(topic, payload []byte, flags publishFlags) bool

Publish sends out a payload to all clients with subscriptions on a provided topic given the passed publish flags.

func (*Broker) PublishRetained

func (b *Broker) PublishRetained(msg *message, sub *subscription)

PublishRetained is used to publish retained messages to a subscribing client.

func (*Broker) PublishToClient

func (b *Broker) PublishToClient(client *Client, packetID uint16, cm *clientMessage)

PublishToClient is used to publish a message with a provided packetId to a specific client.

func (*Broker) Retain

func (b *Broker) Retain(msg *message, topic []byte)

Retain stores a msg in a specific topic as a retained message.

func (*Broker) Subscribe

func (b *Broker) Subscribe(client *Client, filter []byte, qos byte) bool

Subscribe receives a client, a filter and qos level to create or update a subscription.

func (*Broker) Unsubscribe

func (b *Broker) Unsubscribe(client *Client, filter []byte) bool

Unsubscribe receives a client and a filter to remove a subscription.

func (*Broker) UnsubscribeAll

func (b *Broker) UnsubscribeAll(client *Client)

UnsubscribeAll is used to remove all subscriptions of a client. Currently used when the Client disconnects.

type Client

type Client struct {
	ClientID           string
	WillMessage        *message
	Username, Password string
	Session            *session
	// contains filtered or unexported fields
}

Client is the main struct for every client that connects to GOTT. Holds all the info needed to process its messages and maintain state.

type Config

type Config struct {
	ConfigPath string
	Listen     string
	Tls        tlsConfig
	WebSockets webSocketsConfig `yaml:"websockets"`
	Logging    loggingConfig
	Plugins    []interface{}
	// contains filtered or unexported fields
}

Config holds the parsed config file