jeffallen/mqtt

MQTT Clients and Servers in Go

mqtt

MQTT Clients, Servers and Load Testers in Go

For docs, see: http://godoc.org/github.com/jeffallen/mqtt

For a little discussion of this code see: http://blog.nella.org/mqtt-code-golf

Limitations

At this time, the following limitations apply:

• QoS level 0 only; messages are only stored in RAM
• Retain works, but at QoS level 0. Retained messages are lost on server restart.
• Last will messages are not implemented.
• Keepalive and timeouts are not implemented.

Servers

The example MQTT servers are in directories mqttsrv and smqttsrv (secured with TLS).

Benchmarking Tools

To use the benchmarking tools, cd into pingtest, loadtest, or many and type "go build". The tools have reasonable defaults, but you'll also want to use the -help flag to find out what can be tuned.

All benchmarks suck, and these three suck in different ways.

pingtest simulates a number of pairs of clients who are bouncing messages between them as fast as possible. It aims to measure latency of messages through the system when under load.

loadtest simulates a number of pairs of clients where one is sending as fast as possible to the other. Realistically, this ends up testing the ability of the system to decode and queue messages, because any slight inbalance in scheduling of readers causes a pile up of messages from the writer slamming them down the throat of the server.

many simulates a large number of clients who send a low transaction rate. The goal is to eventually use this to achieve 1 million (and more?) concurrent, active MQTT sessions in one server. So far, mqttsrv has survived a load of 40k concurrent connections from many.

mqtt's Issues

range over incoming eventually quits

I stole this chunk of code from one of your examples and then added a Println() call:

	// Receiver
	for m := range mqttClient.Incoming {
		fmt.Print(m.TopicName, "\t")
		m.Payload.WritePayload(os.Stdout)
		fmt.Println("\tr: ", m.Header.Retain)
	}

	fmt.Println("ran out of incoming")

and saw that if there weren't any messages for some period of time it would exit the for loop and my program would end.

Is this because of something on the mosquitto server closing my connection for being idle?

mqttsrv can't handle loadtest

--conns=500 --messages=1000 causes it to do this:

015/10/17 20:02:23 {IncomingConn: sub245}: failed to submit message
2015/10/17 20:02:23 {IncomingConn: sub245}: failed to submit message
2015/10/17 20:02:23 {IncomingConn: sub245}: failed to submit message
2015/10/17 20:02:23 {IncomingConn: sub245}: failed to submit message
2015/10/17 20:02:23 {IncomingConn: sub245}: failed to submit message
2015/10/17 20:02:23 {IncomingConn: sub245}: failed to submit message
2015/10/17 20:02:23 {IncomingConn: sub245}: failed to submit message
2015/10/17 20:02:23 {IncomingConn: sub245}: failed to submit message
2015/10/17 20:02:23 {IncomingConn: sub245}: failed to submit message
2015/10/17 20:02:23 {IncomingConn: sub77}: failed to submit message
2015/10/17 20:02:23 {IncomingConn: sub245}: failed to submit message
2015/10/17 20:02:23 {IncomingConn: sub77}: failed to submit message
2015/10/17 20:02:23 {IncomingConn: sub77}: failed to submit message
2015/10/17 20:02:23 {IncomingConn: sub245}: failed to submit message
2015/10/17 20:02:23 {IncomingConn: pub101}: failed to submit message
2015/10/17 20:02:23 {IncomingConn: pub101}: failed to submit message
2015/10/17 20:02:23 {IncomingConn: pub101}: failed to submit message
2015/10/17 20:02:23 {IncomingConn: pub101}: failed to submit message
2015/10/17 20:02:23 {IncomingConn: pub101}: failed to submit message
2015/10/17 20:02:23 {IncomingConn: pub101}: failed to submit message
2015/10/17 20:02:23 {IncomingConn: pub101}: failed to submit message
2015/10/17 20:02:23 {IncomingConn: pub101}: failed to submit message
2015/10/17 20:02:23 {IncomingConn: pub101}: failed to submit message
2015/10/17 20:02:23 {IncomingConn: pub101}: failed to submit message
2015/10/17 20:02:23 {IncomingConn: pub101}: failed to submit message

Fewer messages are fine.

mqtt.go, have a bug

at line 586 : m.Header.QosLevel != proto.QosAtMostOnce
here may be : m.Header.QosLevel != proto.QosAtLeastOnce

Invalid subscription packet being sent

I am experiencing invalid subscription error messages while load testing this mqtt v3.1/v3.1.1 broker. Could you point/assist me get the tests working. Mainly I would like to get its limits on a single 8GB server.

error "failed to submit message" under heavy load

The current queueing mechanism behaves badly when the input rate of messages which should depart on a certain connection is higher than the rate that the connection writer is managing to drain the queue and send them out.

This shows up in some configurations of loadtest.

The right solution is a variable-sized queue which grows up to a maximum size. Message drops after that point should be logged in a $SYS stat.

Are you planning to maintain this?

Gday

Just started hacking on something similar and found the work you have done so far.

Wanted to know if you plan on continuing it?

Cheers

Resources occupied after client connections aborted

the server holds the resources of memories and threads while connections aborted by Ctrl+C, the resources release only happens while "incomingConn" reader or writer goroutine returned. is there a good way to notify server when connection closed?

Error: Invalid subscriptions - invalid qos

Hello Jeff,
I am getting above issue whenever I am trying to subscribe to any topic. Here are the steps which I followed:

1. Installation:

git clone https://github.com/aws/aws-iot-device-sdk-js.git
cd aws-iot-device-sdk-js
npm install

Sample code:

var awsIot = require('aws-iot-device-sdk');
var device = awsIot.device({
   keyPath: './certs/private.pem.key',
  certPath: './certs/certificate.pem.crt',
    caPath: './certs/root-CA.crt',
  clientId: 'myThing',
    region: 'us-east-1'
});

  device
    .on('connect', function() {
      console.log('connected');
      device.subscribe('topic_1', function(error, result) {
        console.log(result);
      });
    });

it gets connected, however throws error on :

device.subscribe('topic_1', function(error, result) {
        console.log(result);

Any guideline?

thanks,

Disconnect existing connections does not work?

in mqtt.go

			// Disconnect existing connections.
			if existing := c.add(); existing != nil {
				disconnect := &proto.Disconnect{}
				r := existing.submitSync(disconnect)
				r.wait()
				existing.del()
			}
			c.add()

it seem "Disconnect existing connections."does not work.

func (c *incomingConn) add() *incomingConn {
	clientsMu.Lock()
	defer clientsMu.Unlock()

	existing, ok := clients[c.clientid]
	if !ok {
		// this client id already exists, return it
		return existing
	}

	clients[c.clientid] = c
	return nil
}

to modify
if !ok { --- > if ok {

in line 450:

// Delete a connection; the conection must be closed by the caller first.
func (c *incomingConn) del() {
	clientsMu.Lock()
	defer clientsMu.Unlock()
	delete(clients, c.clientid)
	return
}

How should I close the connection or the connection's goroutine ? thanks!

loadtest causes mqttsrv to try to use closed network connections

With Go 1.5 on Arch Linux, running the server then loadtest with --messages=500 --conns=500 never completes due to unreceived messages. The server prints:

2015/10/15 15:51:14 {IncomingConn: 242846536268381992}: failed to submit message
2015/10/15 15:51:14 {IncomingConn: 5207166632368852897}: failed to submit message
2015/10/15 15:51:14 {IncomingConn: 5207166632368852897}: failed to submit message
2015/10/15 15:51:14 {IncomingConn: 5207166632368852897}: failed to submit message
2015/10/15 15:51:14 {IncomingConn: 5207166632368852897}: failed to submit message
2015/10/15 15:51:14 {IncomingConn: 5207166632368852897}: failed to submit message
2015/10/15 15:51:14 {IncomingConn: 5207166632368852897}: failed to submit message
2015/10/15 15:51:14 {IncomingConn: 5207166632368852897}: failed to submit message
2015/10/15 15:51:14 {IncomingConn: 5207166632368852897}: failed to submit message
2015/10/15 15:51:14 {IncomingConn: 5207166632368852897}: failed to submit message
2015/10/15 15:51:14 {IncomingConn: 5207166632368852897}: failed to submit message
2015/10/15 15:51:14 {IncomingConn: 5207166632368852897}: failed to submit message
2015/10/15 15:51:14 {IncomingConn: 5207166632368852897}: failed to submit message
2015/10/15 15:51:14 writer: write tcp [::1]:1883->[::1]:46156: use of closed network connection
2015/10/15 15:51:14 writer: write tcp [::1]:1883->[::1]:46138: use of closed network connection
2015/10/15 15:51:14 writer: write tcp [::1]:1883->[::1]:46384: use of closed network connection
2015/10/15 15:51:14 writer: write tcp [::1]:1883->[::1]:46226: use of closed network connection
2015/10/15 15:51:14 writer: write tcp [::1]:1883->[::1]:46268: use of closed network connection
2015/10/15 15:51:14 writer: write tcp [::1]:1883->[::1]:46348: use of closed network connection
2015/10/15 15:51:14 writer: write tcp [::1]:1883->[::1]:46442: use of closed network connection

Race detect!!!

WARNING: DATA RACE
Read at 0x00c0000601f0 by goroutine 39:
runtime.chansend()
d:/coding/golang/src/runtime/chan.go:142 +0x0

  d:/coding/golang/_home/src/github.com/jeffallen/mqtt/mqtt.go:885 +0x23d

Previous write at 0x00c0000601f0 by goroutine 43:
runtime.closechan()
d:/coding/golang/src/runtime/chan.go:334 +0x0
github.com/jeffallen/mqtt.(*ClientConn).reader.func1()
d:/coding/golang/_home/src/github.com/jeffallen/mqtt/mqtt.go:743 +0x5b
github.com/jeffallen/mqtt.(*ClientConn).reader()
d:/coding/golang/_home/src/github.com/jeffallen/mqtt/mqtt.go:754 +0x630

Authentication of Clients by the Server ？

Authentication of Clients by the Server ？Where to verify, the test does not success。
pub/main.go and sub/main.go
var user = flag.String("user", "username", "username")
var pass = flag.String("pass", "12345", "password")
and
var user = flag.String("user", "", "username")
var pass = flag.String("pass", "", "password")

No difference was found

server: Connection Refused: identifier rejected

I'm failing to understand how the server handles incoming client connections or user auth. A pointer to example code or docs would be highly appreciated.

question about NAT

I am playing with this MQTT lib, but now i am wondering how to connect to a remote device that is behind a NAted environment. At the remote location outbound connections over 80, 443 are allowed.
I might be able to get 22 outbound open too.

i basically are aiming for:

• being able to connect with SSH
• upgrade the software remotely. I plan to make the remote software just use 80 / 443.

Any thoughts on best way to approach this ?

not support mqtt3.1.1？

when i use uuid as clientid , it can not connect with the mqtt server

not able to build the project

I tried to build sub folder, it tells me "cannot find package github.com/huin/mqtt ".