<p align="center">

</p>

Comqtt

A lightweight, high-performance MQTT server in Go (v3.0｜v3.1.1｜v5.0)

Comqtt is an embeddable high-performance MQTT broker server written in Go, and supporting distributed cluster, and compliant with the MQTT v3.0 and v3.1.1 and v5.0 specification for the development of IoT and smarthome projects. The server can be used either as a standalone binary or embedded as a library in your own projects. Comqtt message throughput is comparable with everyone's favourites such as Mosquitto, Mosca, and VerneMQ.

:+1: Comqtt code is cleaner, easier to read, customize, and extend than other Mqtt Broker code! :heart_eyes:

:+1: If you like this project or it's useful to you, please give it a STAR, let more people know about it, and contribute in it's maintenance together! :muscle:

📦 💬 See Github Discussions for discussions about releases

Ongoing discussion about current and future releases can be found at https://github.com/wind-c/comqtt/discussions

Developers in China can join wechat group discussions at https://github.com/wind-c/comqtt/discussions/32

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

When is this repo updated?

Unless it's a critical issue, new releases typically go out over the weekend. At some point in the future this repo may be converted to an organisation, or collaborators added if the project continues to grow.

Comqtt Features

• Full MQTTv5 Feature Compliance, compatibility for MQTT v3.1.1 and v3.0.0.
• TCP, Websocket, QUIC (including SSL/TLS) and Dashboard listeners.
• File-based server, auth, storage and bridge configuration, Click to see config examples.
• Auth and ACL Plugin is supported Redis, HTTP, Mysql and PostgreSql.
• Packets are bridged to kafka according to the configured rule.
• Single-machine mode supports local storage BBolt, Badger and Redis.
• Hook design pattern makes it easy to develop plugins for Auth, Bridge, and Storage.
• Cluster support is based on Gossip and Raft, Click to Cluster README.

Roadmap

• Dashboard.
• Rule engine.
• Bridge(Other Mqtt Broker、RocketMQ、RabbitMQ).
• Enhanced Metrics support.
• CoAP.

Restful API

• GET /api/v1/mqtt/config : [single] get configuration parameters of mqtt server
• GET /api/v1/mqtt/stat/overall : [single] get mqtt server info
• GET /api/v1/mqtt/stat/online : [single] get online number
• GET /api/v1/mqtt/clients/{id} : [single] get a client info
• GET /api/v1/mqtt/blacklist : [single/cluster] get blacklist, each node in the cluster has the same blacklist
• POST /api/v1/mqtt/blacklist/{id} : [single] disconnect the client and add it to the blacklist
• DELETE api/v1/mqtt/blacklist/{id} : [single] remove from the blacklist
• POST /api/v1/mqtt/message : [single/cluster] publish message to subscribers in the cluster, body {"topic_name": "xxx", "payload": "xxx", "retain": true/false, "qos": 1}
• GET /api/v1/node/config : [cluster] get configuration parameters of node
• DELETE /api/v1/node/{name} : [cluster] leave local node gracefully exits the cluster.Call this API on the node to be deleted, exiting the cluster actively can prevent other nodes from constantly attempting to connect to that node.
• GET /api/v1/cluster/nodes : [cluster] get all nodes in the cluster
• POST /api/v1/cluster/nodes : [cluster] add a node to the cluster, body {"name": "xx", "addr": "ip:port"}.If the configuration file sets "members: [ip:port]", then the node will automatically join the cluster upon startup and there is no need to call this API.
• GET /api/v1/cluster/stat/online : [cluster] online number from all nodes in the cluster
• GET /api/v1/cluster/clients/{id} : [cluster] get a client information, search from all nodes in the cluster
• POST /api/v1/cluster/blacklist/{id} : [cluster] add clientId to the blacklist on all nodes in the cluster
• DELETE /api/v1/cluster/blacklist/{id} : [cluster] remove from the blacklist on all nodes in the cluster

<!-- POST /api/v1/cluster/peers : [cluster] add peer to raft cluster, body {"name": "xx", "addr": "ip:port"} --> <!-- DELETE /api/v1/cluster/peers/{name} : [cluster] remove peer from raft cluster -->

Quick Start

Running the Broker with Go

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

Build

cd cmd
go build -o comqtt ./single/main.go

Start

./comqtt
or
./comqtt --conf=./config/single.yml

If you want to obtain the bridge and multiple authentication capabilities, you need to use the configuration file to start.Click to config example.

Using Docker

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

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

Developing with Comqtt

Importing as a package

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

import (
  "log"

  "github.com/wind-c/comqtt/v2/mqtt"
  "github.com/wind-c/comqtt/v2/mqtt/hooks/auth"
  "github.com/wind-c/comqtt/v2/mqtt/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 mqtt/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.NewNet | A net.Listener listener | | listeners.NewWebsocket | A Websocket listener | | listeners.NewHTTPStats | An HTTP $SYS info dashboard | | listeners.NewHTTPHealthCheck | An HTTP healthcheck listener to provide health check responses for e.g. cloud infrastructure |

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 mqtt/examples folder or cmd/single/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,
    },
  },
  ClientNetWriteBufferSize: 1024,
  ClientNetReadBufferSize: 1024,
  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 | mqtt/hooks/auth.AllowHook | Allow access to all connecting clients and read/write to all topics. | | Access Control | mqtt/hooks/auth.Auth | Rule-based access control ledger. | | Persistence | mqtt/hooks/storage/bolt | Persistent storage using BoltDB (deprecated). | | Persistence | mqtt/hooks/storage/badger | Persistent storage using BadgerDB. | | Persistence | mqtt/hooks/storage/redis | Persistent storage using Redis. | | Debugging | [mqtt/hooks/debug]