natsmqtt5

README

natsmqtt5

An MQTT v5.0 broker that uses an existing NATS server as its messaging core.

nats-server speaks MQTT 3.1.1 and will not be adding v5: "we don't want to own the ongoing maintenance cost for the massive API surface increase that comes with MQTT 5 at this time." This is the other option raised in that thread — a separate broker that keeps NATS as the messaging engine, so you point it at the NATS cluster you already run rather than deploying a second, unrelated message bus.

MQTT v5 clients ──TCP/TLS──▶ natsmqtt5 ──▶ your existing NATS server
                             (a NATS client, not a fork)

What it does

• Speaks MQTT v5 to clients. All fifteen control packets, all 27 properties. Verified against the Eclipse Paho v5 client over a real socket.
• Uses NATS for everything underneath. Topics become NATS subjects, subscriptions become NATS subscriptions, shared subscriptions become NATS queue groups, retained messages live in a JetStream stream.
• Interoperates on the subject level. The topic-to-subject mapping is byte-for-byte the one nats-server uses for its own MQTT support, so sensors/7/temp lands on mqtt.sensors.7.temp either way. A NATS-native publisher reaches MQTT subscribers and vice versa, with no bridge.
• Scales horizontally by adding brokers. Two brokers against one NATS server are one logical broker: a client on either reaches subscribers on both.
• Embeds as a library. natsmqtt5.New returns a *Broker you run inside your own process, sharing your *nats.Conn if you want.

Install

As a binary:

go install github.com/taumatix/natsmqtt5/cmd/natsmqtt5@v0.1.0
natsmqtt5 -nats nats://localhost:4222 -listen :1883

As a library:

go get github.com/taumatix/natsmqtt5@v0.1.0

Requires Go 1.25 or newer, and a NATS server with JetStream enabled (or -no-retained / Options.DisableRetained to run without it).

Use

package main

import (
	"context"
	"log"

	"github.com/taumatix/natsmqtt5"
)

func main() {
	broker, err := natsmqtt5.New(natsmqtt5.Options{
		NATSURL: "nats://localhost:4222",
		Listen:  ":1883",
	})
	if err != nil {
		log.Fatal(err)
	}
	defer broker.Close()

	if err := broker.Serve(context.Background()); err != nil {
		log.Fatal(err)
	}
}

Authentication and authorization are two small interfaces:

opts := natsmqtt5.Options{
	Authenticator: natsmqtt5.AuthenticatorFunc(
		func(ctx context.Context, req *natsmqtt5.AuthRequest) (*natsmqtt5.AuthResult, error) {
			if !credentialsOK(req.Username, req.Password) {
				// Choose the CONNACK Reason Code the client sees.
				return nil, &natsmqtt5.ConnectError{Code: packet.BadUserNameOrPassword}
			}
			return &natsmqtt5.AuthResult{Identity: req.Username}, nil
		}),

	Authorizer: natsmqtt5.AuthorizerFunc(
		func(ctx context.Context, req *natsmqtt5.AuthzRequest) error {
			if req.Action == natsmqtt5.ActionPublish && !mayWrite(req.Identity, req.Topic) {
				return errors.New("denied") // becomes 0x87 Not authorized in the PUBACK
			}
			return nil
		}),
}

AuthRequest carries the TLS ConnectionState, so client-certificate authentication is a few lines.

How MQTT maps onto NATS

MQTT	NATS
Topic sensors/7/temp	Subject mqtt.sensors.7.temp
Topic level /	Subject token separator .
A literal . in a topic	//
An empty topic level	./ or /. depending on position
Filter wildcard +	*
Filter wildcard #	>, plus a second subscription on the parent
Shared subscription $share/g/f	Queue group on the subject for f
Retained message	JetStream stream, MaxMsgsPerSubject: 1
v5 properties	NATS headers (Mqtt5-Content-Type, Mqtt5-User, …)

# gets two NATS subscriptions because MQTT's # matches the parent level (sport/# matches sport) and NATS's > does not.

The mqtt subject prefix is configurable with SubjectPrefix; brokers that should share traffic must agree on it, and brokers that must not share traffic must differ.

A NATS message with no Mqtt5-* headers is delivered as a QoS 0 message with no properties, which is what makes plain nats pub reach MQTT subscribers.

What v0.1.0 does not do

Stated plainly, because a broker you cannot trust the limits of is worse than one with fewer features:

• Sessions do not survive a broker restart, and are not shared between brokers. A session is held in the broker process. A client reconnecting to the same broker within its Session Expiry Interval resumes its subscriptions; reconnecting to a different broker, or after a restart, gets a fresh session and must subscribe again. Retained messages and live traffic are unaffected — those go through NATS.
• No offline message queue. Messages published while a session is disconnected are not stored for it.
• No QoS 1/2 retransmission on reconnect, which follows from the above.
• Shared subscriptions are granted QoS 1 at most. A QoS 2 request on a $share/ filter is answered with Granted QoS 1 in the SUBACK, which [MQTT-3.8.4-7] permits. Exactly-once delivery to one member of a group needs per-member ownership of a half-finished PUBREC/PUBREL exchange, and JetStream's unit of ownership is a consumer, not a queue-group member.
• No enhanced authentication (AUTH packets). A CONNECT carrying an Authentication Method is refused with 0x8C Bad authentication method, which is the conforming answer from a server that does not support it.
• The broker never sends topic aliases, though it accepts them from clients.
• Message Expiry Interval is forwarded but not enforced on stored retained messages.
• Topics containing a space, a tab, *, > or DEL are refused with 0x90 Topic Name invalid. MQTT permits them; a NATS subject cannot carry them. nats-server lets * and > through, which silently turns an MQTT subscription to a/*/b into a NATS wildcard that receives topics it must not see. Failing loudly is the deliberate choice here; see ROADMAP.md for the escaping option.

Correctness

The MQTT layer is written against the OASIS MQTT v5.0 Standard of 07 March 2019, and the tests cite it:

• The codec tests assert the literal bytes from the spec's own worked examples (Figures 3-6, 3-9, 3-19, 3-21) and every boundary in Table 1-1, rather than only round-tripping our own encoder against our own decoder.
• The topic mapping is checked against all 40 conversion fixtures from nats-server's TestMQTTTopicAndSubjectConversion and TestMQTTFilterConversion, so a divergence from nats-server fails the build.
• The matching tests use the examples in §4.7.1.2, §4.7.1.3, §4.7.2 and §4.8.2.
• Integration tests run a real NATS server in-process with JetStream, a real broker, and the Eclipse Paho v5 client over a real TCP socket.
• The packet decoder and the topic mapping are fuzzed in CI.

Packages

Package	What it is
github.com/taumatix/natsmqtt5	The broker
.../packet	MQTT v5 wire codec — standard library only, no NATS
.../topic	Topic validation, matching and the NATS subject mapping

packet and topic are usable on their own; neither imports NATS.

Licence

Apache 2.0. See LICENSE.

Documentation

Overview

Package natsmqtt5 is an MQTT version 5.0 broker that uses an existing NATS server as its messaging core.

The broker is a NATS client, not a fork of nats-server. Point it at a NATS URL you already run and it speaks MQTT v5 on a TCP port, translating topics to NATS subjects, fan-out to NATS subscriptions, shared subscriptions to NATS queue groups and retained messages to a JetStream stream. Two brokers against the same NATS server see each other's traffic, so an MQTT client on one reaches a subscriber on the other.

The subject mapping is byte-for-byte the one nats-server uses for its own MQTT 3.1.1 support, so topics land where a NATS-native subscriber expects.

Minimal use:

b, err := natsmqtt5.New(natsmqtt5.Options{NATSURL: "nats://localhost:4222"})
if err != nil {
	return err
}
defer b.Close()
return b.Serve(ctx)

Section numbers in the documentation refer to the OASIS MQTT v5.0 Standard of 07 March 2019.

Example

Run a broker against an existing NATS server.

package main

import (
	"context"
	"log"

	"github.com/taumatix/natsmqtt5"
)

func main() {
	broker, err := natsmqtt5.New(natsmqtt5.Options{
		NATSURL: "nats://localhost:4222",
		Listen:  ":1883",
	})
	if err != nil {
		log.Fatal(err)
	}
	defer broker.Close()

	if err := broker.Serve(context.Background()); err != nil {
		log.Fatal(err)
	}
}

Example (Authentication)

Reject unknown credentials with the CONNACK Reason Code of your choosing, and confine each user to its own topic subtree.

package main

import (
	"context"
	"errors"
	"log"
	"strings"

	"github.com/taumatix/natsmqtt5"
	"github.com/taumatix/natsmqtt5/packet"
)

func main() {
	passwords := map[string]string{"alice": "s3cret"}

	opts := natsmqtt5.Options{
		NATSURL: "nats://localhost:4222",

		Authenticator: natsmqtt5.AuthenticatorFunc(
			func(_ context.Context, req *natsmqtt5.AuthRequest) (*natsmqtt5.AuthResult, error) {
				if want, ok := passwords[req.Username]; !ok || want != string(req.Password) {
					return nil, &natsmqtt5.ConnectError{Code: packet.BadUserNameOrPassword}
				}
				return &natsmqtt5.AuthResult{Identity: req.Username}, nil
			}),

		Authorizer: natsmqtt5.AuthorizerFunc(
			func(_ context.Context, req *natsmqtt5.AuthzRequest) error {
				// A denied publish comes back as 0x87 in the PUBACK; a denied
				// subscription as 0x87 in the SUBACK. The connection stays up.
				if !strings.HasPrefix(req.Topic, "users/"+req.Identity+"/") {
					return errors.New("outside the user's subtree")
				}
				return nil
			}),
	}

	broker, err := natsmqtt5.New(opts)
	if err != nil {
		log.Fatal(err)
	}
	defer broker.Close()

	log.Fatal(broker.Serve(context.Background()))
}

Example (Embedded)

Share one NATS connection between the broker and the rest of an application, and keep MQTT traffic in its own subject namespace.

package main

import (
	"context"
	"log"

	"github.com/nats-io/nats.go"

	"github.com/taumatix/natsmqtt5"
)

func main() {
	// An existing connection the application already uses. The broker does not
	// close a connection it did not open.
	nc, err := nats.Connect("nats://localhost:4222")
	if err != nil {
		log.Fatal(err)
	}
	defer nc.Close()

	broker, err := natsmqtt5.New(natsmqtt5.Options{
		Conn:          nc,
		Listen:        "127.0.0.1:1883",
		SubjectPrefix: "iot",
		StreamPrefix:  "IOT",
	})
	if err != nil {
		log.Fatal(err)
	}
	defer broker.Close()

	// An MQTT client publishing to "sensors/7/temp" now reaches any NATS
	// subscriber on "iot.sensors.7.temp", and the other way round.
	log.Fatal(broker.Serve(context.Background()))
}

Index

• Constants
• Variables
• func Ptr[T any](v T) *T
• type Action
  • func (a Action) String() string
• type AuthRequest
• type AuthResult
• type Authenticator
• type AuthenticatorFunc
  • func (f AuthenticatorFunc) Authenticate(ctx context.Context, req *AuthRequest) (*AuthResult, error)
• type Authorizer
• type AuthorizerFunc
  • func (f AuthorizerFunc) Authorize(ctx context.Context, req *AuthzRequest) error
• type AuthzRequest
• type Broker
  • func New(opts Options) (*Broker, error)
  • func NewWithContext(ctx context.Context, opts Options) (*Broker, error)
  • func (b *Broker) Close() error
  • func (b *Broker) ListenAddr() net.Addr
  • func (b *Broker) NATS() *nats.Conn
  • func (b *Broker) Serve(ctx context.Context) error
• type ConnectError
  • func (e *ConnectError) Error() string
• type Options

Examples

• Package
• Package (Authentication)
• Package (Embedded)

Constants

const (
	DefaultListen        = ":1883"
	DefaultSubjectPrefix = "mqtt"
	DefaultStreamPrefix  = "MQTT5"
	// DefaultReceiveMaximum bounds how many QoS 1 and QoS 2 publications a
	// client may have in flight towards the broker (MQTT-5.0 §3.2.2.3.3).
	DefaultReceiveMaximum = 1024
	// DefaultTopicAliasMaximum is the highest Topic Alias the broker accepts
	// from a client (MQTT-5.0 §3.1.2.11.5).
	DefaultTopicAliasMaximum = 64
	// DefaultMaximumPacketSize caps an inbound packet at 64 MiB
	// (MQTT-5.0 §3.2.2.3.6).
	DefaultMaximumPacketSize = 64 << 20
	// DefaultConnectTimeout bounds how long a new connection may stay silent
	// before sending its CONNECT (MQTT-5.0 §3.1.4, step 1).
	DefaultConnectTimeout = 10 * time.Second
	// DefaultMaxSessionExpiry caps the Session Expiry Interval a client may
	// request. The broker returns the capped value in CONNACK, as
	// MQTT-5.0 §3.2.2.3.2 provides for.
	DefaultMaxSessionExpiry = time.Hour
)

Defaults applied to the zero value of the corresponding Options field.

Variables

var ErrInvalidOptions = errors.New("natsmqtt5: invalid options")

ErrInvalidOptions reports an Options value the broker cannot run with.

Functions

func Ptr

func Ptr[T any](v T) *T

Ptr returns a pointer to v. The optional Options fields are pointers so that "not set" is distinguishable from a meaningful zero, and this saves callers declaring a variable to take the address of.

natsmqtt5.Options{MaximumQoS: natsmqtt5.Ptr(uint8(1))}

Types

type Action

type Action int

Action is an operation an Authorizer decides on.

const (
	// ActionPublish is a client publishing to a Topic Name.
	ActionPublish Action = iota
	// ActionSubscribe is a client subscribing to a Topic Filter.
	ActionSubscribe
)

func (Action) String

func (a Action) String() string

type AuthRequest

type AuthRequest struct {
	// ClientID is the Client Identifier from CONNECT, before the broker
	// assigns one for an empty value.
	ClientID string
	// Username is empty when the User Name Flag was not set.
	Username string
	// Password is nil when the Password Flag was not set. MQTT v5 permits a
	// password with no username (MQTT-5.0 §3.1.2.9).
	Password []byte
	// RemoteAddr is the client's network address.
	RemoteAddr net.Addr
	// TLS is the connection's TLS state, or nil for a plaintext connection.
	// Use it for client-certificate authentication.
	TLS *tls.ConnectionState
	// Properties are the CONNECT properties, including any User Properties.
	Properties *packet.Properties
}

AuthRequest describes a connection attempt for an Authenticator.

type AuthResult

type AuthResult struct {
	// Identity, when non-empty, names the authenticated principal. It is
	// passed to the Authorizer and included in log records; it does not change
	// the Client Identifier.
	Identity string
	// AssignClientID, when non-empty, overrides the Client Identifier for the
	// session and is returned to the client as the Assigned Client Identifier
	// [MQTT-3.1.3-7].
	AssignClientID string
}

AuthResult is an Authenticator's verdict.

type Authenticator

type Authenticator interface {
	Authenticate(ctx context.Context, req *AuthRequest) (*AuthResult, error)
}

Authenticator decides whether a CONNECT is accepted.

Returning a nil error accepts the connection. Returning an error rejects it; wrap or return a *ConnectError to choose the CONNACK Reason Code, otherwise the broker answers 0x87 (Not authorized) and does not disclose the reason.

type AuthenticatorFunc

type AuthenticatorFunc func(ctx context.Context, req *AuthRequest) (*AuthResult, error)

AuthenticatorFunc adapts a function to the Authenticator interface.

func (AuthenticatorFunc) Authenticate

func (f AuthenticatorFunc) Authenticate(ctx context.Context, req *AuthRequest) (*AuthResult, error)

Authenticate calls f.

type Authorizer

type Authorizer interface {
	Authorize(ctx context.Context, req *AuthzRequest) error
}

Authorizer decides whether a session may publish to or subscribe to a topic. A nil error permits the operation.

A denied publish is answered with 0x87 (Not authorized) in the PUBACK or PUBREC, or dropped silently at QoS 0, as MQTT-5.0 §3.3.4 requires. A denied subscription is answered with 0x87 in the SUBACK for that filter.

type AuthorizerFunc

type AuthorizerFunc func(ctx context.Context, req *AuthzRequest) error

AuthorizerFunc adapts a function to the Authorizer interface.

func (AuthorizerFunc) Authorize

func (f AuthorizerFunc) Authorize(ctx context.Context, req *AuthzRequest) error

Authorize calls f.

type AuthzRequest

type AuthzRequest struct {
	Action Action
	// ClientID is the session's Client Identifier, after any assignment.
	ClientID string
	// Identity is the AuthResult.Identity from authentication, if any.
	Identity string
	// Username is the User Name from CONNECT, if any.
	Username string
	// Topic is the Topic Name for ActionPublish or the Topic Filter, shared
	// prefix included, for ActionSubscribe.
	Topic string
	// QoS is the requested QoS.
	QoS packet.QoS
	// Retain is set for a publish with the RETAIN flag.
	Retain bool
}

AuthzRequest describes an operation for an Authorizer.

type Broker

type Broker struct {
	// contains filtered or unexported fields
}

Broker is an MQTT v5 broker backed by a NATS server. Create one with New, run it with Serve, and release its resources with Close.

A Broker is safe for concurrent use.

func New

func New(opts Options) (*Broker, error)

New connects to NATS, provisions the JetStream assets the broker needs and binds the MQTT listener. It does not accept connections until Serve is called, so a caller can read ListenAddr first.

The returned Broker must be closed with Close.

func NewWithContext

func NewWithContext(ctx context.Context, opts Options) (*Broker, error)

NewWithContext is New with a context bounding the NATS connection and JetStream setup.

func (*Broker) Close

func (b *Broker) Close() error

Close stops the broker: it closes the listener, disconnects every client with 0x8B (Server shutting down) and releases the NATS resources it owns. A NATS connection passed in through Options.Conn is left open.

func (*Broker) ListenAddr

func (b *Broker) ListenAddr() net.Addr

ListenAddr reports the address the broker accepts MQTT connections on. It is valid as soon as New returns, which lets a test bind :0 and discover the port.

func (*Broker) NATS

func (b *Broker) NATS() *nats.Conn

NATS returns the NATS connection the broker uses, so an embedding application can share it rather than opening a second one.

func (*Broker) Serve

func (b *Broker) Serve(ctx context.Context) error

Serve accepts MQTT connections until ctx is cancelled or Close is called. It returns nil on a clean shutdown.

type ConnectError

type ConnectError struct {
	Code   packet.ReasonCode
	Reason string
}

ConnectError carries the CONNACK Reason Code an Authenticator wants the broker to send. Any code of 0x80 or greater from MQTT-5.0 Table 3-1 is valid; common choices are BadUserNameOrPassword (0x86), NotAuthorized (0x87), Banned (0x8A) and ClientIdentifierNotValid (0x85).

func (*ConnectError) Error

func (e *ConnectError) Error() string

type Options

type Options struct {
	// NATSURL is the NATS server to use. Defaults to nats.DefaultURL. Ignored
	// when Conn is set.
	NATSURL string
	// NATSOptions are passed to nats.Connect for credentials, TLS and so on.
	// Ignored when Conn is set.
	NATSOptions []nats.Option
	// Conn is an already-established NATS connection to use instead of
	// dialling. The broker does not close a connection it did not open.
	Conn *nats.Conn

	// Listen is the TCP address to accept MQTT connections on. Defaults to
	// DefaultListen. Ignored when Listener is set.
	Listen string
	// Listener is an already-open listener to accept on instead of binding
	// Listen. The broker closes it on shutdown.
	Listener net.Listener
	// TLSConfig, when non-nil, wraps the listener in TLS.
	TLSConfig *tls.Config

	// SubjectPrefix namespaces every NATS subject the broker publishes MQTT
	// traffic on, keeping it clear of NATS system subjects. Defaults to
	// DefaultSubjectPrefix. Set it to the same value on every broker that
	// should share traffic.
	SubjectPrefix string
	// StreamPrefix names the JetStream assets the broker creates. Defaults to
	// DefaultStreamPrefix. Brokers sharing a SubjectPrefix must share this.
	StreamPrefix string
	// DisableRetained turns off retained-message support, so no JetStream
	// assets are created and CONNACK advertises Retain Available = 0
	// (MQTT-5.0 §3.2.2.3.5). Use it against a NATS server without JetStream.
	DisableRetained bool
	// RetainedStorage selects file or memory storage for the retained-message
	// stream. Defaults to file storage.
	RetainedStorage jetstream.StorageType
	// RetainedReplicas is the JetStream replica count for the retained-message
	// stream. Defaults to 1.
	RetainedReplicas int

	// Authenticator, when non-nil, decides whether a CONNECT is accepted. When
	// nil every connection is accepted, which is only appropriate on a trusted
	// network.
	Authenticator Authenticator
	// Authorizer, when non-nil, decides whether a session may publish to or
	// subscribe to a topic. When nil everything is permitted.
	Authorizer Authorizer

	// ReceiveMaximum is advertised in CONNACK and bounds the client's in-flight
	// QoS 1 and QoS 2 publications. Defaults to DefaultReceiveMaximum.
	ReceiveMaximum uint16
	// MaximumQoS is the highest QoS the broker accepts in a PUBLISH and grants
	// on a SUBSCRIBE, advertised in CONNACK [MQTT-3.2.2-9]. Values above 2 are
	// clamped. Leave it nil for the default of 2; it is a pointer so that
	// Ptr(uint8(0)) can express a broker that only does QoS 0, which the zero
	// value would otherwise make unreachable.
	MaximumQoS *uint8
	// TopicAliasMaximum is advertised in CONNACK. Zero disables inbound topic
	// aliases; leave it unset for DefaultTopicAliasMaximum.
	TopicAliasMaximum *uint16
	// MaximumPacketSize caps an inbound packet. Defaults to
	// DefaultMaximumPacketSize.
	MaximumPacketSize uint32
	// ServerKeepAlive, when non-zero, overrides the client's requested Keep
	// Alive and is returned in CONNACK [MQTT-3.1.2-21].
	ServerKeepAlive uint16
	// ConnectTimeout bounds the wait for a CONNECT on a new connection.
	// Defaults to DefaultConnectTimeout.
	ConnectTimeout time.Duration
	// MaxSessionExpiry caps a requested Session Expiry Interval. Defaults to
	// DefaultMaxSessionExpiry.
	MaxSessionExpiry time.Duration

	// Logger receives broker events. Defaults to slog.Default().
	Logger *slog.Logger
}

Options configures a Broker. The zero value is usable: it connects to nats.DefaultURL and listens on :1883.

Fields are only ever added to this struct, never removed or repurposed, so a composite literal with field names keeps compiling across versions.