soypat/natiu-mqtt

A dead-simple, extensible MQTT implementation well suited for embedded systems.

natiu-mqtt

A dead-simple, extensible and correct MQTT implementation.

Natiu: Means mosquito in the Guaraní language, a language spoken primarily in Paraguay. Commonly written as ñati'û or ñati'ũ.

Highlights

• Modular

  • Client implementation leaves allocating parts up to the Decoder interface type. Users can choose to use non-allocating or allocating implementations of the 3 method interface.
  • RxTx type lets one build an MQTT implementation from scratch for any transport. No server/client logic defined at this level.

• No uneeded allocations: The PUBLISH application message is not handled by this library, the user receives an io.Reader with the underlying transport bytes. This prevents allocations on natiu-mqtt side.

• V3.1.1: Compliant with MQTT version 3.1.1 for QoS0 interactions. QoS1 and QoS2 are WIP.

• No external dependencies: Nada. Nope.

• Data oriented design: Minimizes abstractions or objects for the data on the wire.

• Fuzz tested, robust: Decoding implementation fuzzed to prevent adversarial user input from crashing application (95% coverage).

• Simplicity: A simple base package yields simple implementations for different transports. See Implementations section.

• Runtime-what?: Unlike other MQTT implementations. No channels, no interface conversions, no goroutines- as little runtimey stuff as possible. You get the best of Go's concrete types when using Natiu's API. Why? Because MQTT deserialization and serialization are an embarrassingly serial and concrete problem.

Goals

This implementation will have a simple embedded-systems implementation in the package top level. This implementation will be transport agnostic and non-concurrent. This will make it far easier to modify and reason about. The transport dependent implementations will have their own subpackage, so one package for TCP transport, another for UART, PPP etc.

• Minimal, if any, heap allocations.
• Support for TCP transport.
• User owns payload bytes.

Implementations

• natiu-wsocket: MQTT via Websockets. Tested with moscajs/aedes broker server..

Examples

API subject to before v1.0.0 release.

Example use of Client

	// Create new client.
	client := mqtt.NewClient(mqtt.ClientConfig{
		Decoder: mqtt.DecoderNoAlloc{make([]byte, 1500)},
		OnPub: func(_ mqtt.Header, _ mqtt.VariablesPublish, r io.Reader) error {
			message, _ := io.ReadAll(r)
			log.Println("received message:", string(message))
			return nil
		},
	})

	// Get a transport for MQTT packets.
	const defaultMQTTPort = ":1883"
	conn, err := net.Dial("tcp", "127.0.0.1"+defaultMQTTPort)
	if err != nil {
		fmt.Println(err)
		return
	}

	// Prepare for CONNECT interaction with server.
	var varConn mqtt.VariablesConnect
	varConn.SetDefaultMQTT([]byte("salamanca"))
	ctx, cancel := context.WithTimeout(context.Background(), 4*time.Second)
	err = client.Connect(ctx, conn, &varConn) // Connect to server.
	cancel()
	if err != nil {
		// Error or loop until connect success.
		log.Fatalf("connect attempt failed: %v\n", err)
	}

	// Ping forever until error.
	for {
		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
		pingErr := client.Ping(ctx)
		cancel()
		if pingErr != nil {
			log.Fatal("ping error: ", pingErr, " with disconnect reason:", client.Err())
		}
		log.Println("ping success!")
	}

Why not just use paho?

Some issues with Eclipse's Paho implementation:

• Inherent data races on API side. The implementation is so notoriously hard to modify this issue has been in a frozen state.
• Calling Client.Disconnect when client is already disconnected blocks indefinetely and can cause deadlock or spin with Paho's implementation.
• If there is an issue with the network and Reconnect is enabled then then Paho's Reconnect spins. There is no way to prevent this.
• Interfaces used for ALL data types. This is not necessary and makes it difficult to work with since there is no in-IDE documentation on interface methods.
• No lower level abstraction of MQTT for use in embedded systems with non-TCP transport.
• Uses any interface for the payload, which could simply be a byte slice...

I found these issues after a 2 hour dev session. There will undoubtedly be more if I were to try to actually get it working...

natiu-mqtt's Issues

Over the wire

Hey @soypat

wow this looks pretty nice

it looks like it can act as a control plane between different devices on a board ?

I want to build a control plane gui so anyone can sniff the messages going on between the devices on a board.

I would normally use nats which does support mqtt also btw but I saw hat someone has build a pure mqtt Server also that is pluggable ( Redis, mongo dB ).

looking for some advice from the expert dude :)

I can also share my gui so you get a feel for what I am trying to do. The gui is meant to act as a debugger . It is multi window and streams all changes that it “hears” to the screen.

QoS Receive Example

How would a QoS 2 [exactly once] receive look like? An example in the README and/or a godoc example would be handy.

Calling Disconnect() fails with error "not connected"

After connecting to a MQTT broker, I call Disconnect() and get the error "not connected". I then checked the implementation of the Disconnect() method.

if !c.lastRx.IsZero() {
    return errors.New("not connected")
}

Shouldn't the if-statement be like if c.lastRx.IsZero() { ... }?

Subscription Example

Following along with youre amazing work with the PicoW wifi. Thanks a ton for everything you're doing for the community.

I have been able to successfully write messages over MQTT, but I am stuggling to get a MQTT subscription running...was wondering if you would mind posting an example.

Thanks again!

Best practice when using with go routines

My MQTT client is using this library to abstract the communication with a broker. At first this library looks pretty straight forward to use, but I do find it hard to use with go routines.

Why go routines?
Some parts of the MQTT protocol requires you to read and write simultaneously. For example once you are connected to a broker, you have to send a Ping message after n seconds to keep the connection alive. But also when you're subscribed to a topic, you have to keep reading from the connection using ReadNextPacket(). This means you have to use a go routine to either send or receive messages.

The problem is that when you call Ping() on a different go routine, you will get a data race – the Ping() implementation calls ReadNextPacket(). Within this method the LastReceivedHeader field is written – you get a data race because this field is written on 2 different go routines.

Any ideas on how to avoid that?