ESP8266, NodeMcu and MQTT: Event publishing queueing

The MQTT code that I’ve copied and changed that is on the following post: https://primalcortex.wordpress.com/2015/02/06/nodemcu-and-mqtt-how-to-start/ works fine for subscribing topics as long that you don’t overload it: https://primalcortex.wordpress.com/2015/02/13/esp8266-nodemcu-and-mqtt-event-subscription-load-testing/

The code subscribes an topic array for receiving and this works, more or less fine. But it also creates a timer to call periodically a publish_data functions, for testing purposes. In reality imagine that we want to publish something when an Input pin changes value. We can do that with the trig function. For example:

gpio.trig(5, "down",publish_data1)
gpio.trig(4, "down",publish_data2)

Where publish_data# are functions that publish something to a MQTT topic. The problem is that it might happen that sometimes both triggers (two buttons pressed at the same time, for example) are activated and try to publish to a MQTT topic at the same time.
Let’s simulate this ( please note that this code is a section of the code in the MQTT Getting Started post.

-- Sample publish functions:
function publish_data(topic, message)
if pub_sem == 0 then
pub_sem = 1
m:publish( topic, message ,0 ,0 , function(conn)
print("Sending to " .. topic ..": " .. message)
pub_sem = 0
id1 = id1 +1
end)
end
end

function publish_data1()
publish_data("Out1","Mensagem para out1" )
publish_data("Out2","Mensagem para out2" )
end

In the above code sample, the Out2 topic message is never published because it takes time to the message for the previous topic to go out (Out1), and so the message gets lost. And this happens because we can’t call the m:publish for MQTT publishing before it has ended the previous call, hence the semaphore to avoid this.

So we are going to do some queueing for publishing:

I’m using an unbound Lua array, with two pointers, one for the Head, and another for the Tail.
Please note that this code is not perfect. The last message is only sent if adding a new message.
Also note that this simple code takes huge amounts of memory, even on “compile/interpretation” phase.
But it works for a while before crashing down with lack of memory.

-- Configuration to connect to the MQTT broker.
BROKER = "192.168.1.16" -- Ip/hostname of MQTT broker
BRPORT = 1883 -- MQTT broker port
BRUSER = "user" -- If MQTT authenitcation is used then define the user
BRPWD = "pwd" -- The above user password
CLIENTID = "ESP8266-" .. node.chipid() -- The MQTT ID. Change to something you like

-- MQTT topics to subscribe
topics = {"topic1","topic2","topic3","topic4"} -- Add/remove topics to the array

-- Control variables.
pub_sem = 0 -- MQTT Publish semaphore. Stops the publishing whne the previous hasn't ended
current_topic = 1 -- variable for one currently being subscribed to
topicsub_delay = 50 -- microseconds between subscription attempts, worked for me (local network) down to 5...YMMV

-- Publishing structures
pub_topic = {}
pub_message = {}
pub_head = 1
pub_tail = 1

-- connect to the broker
print("heap: " .. node.heap() )
m = mqtt.Client( CLIENTID, 120, BRUSER, BRPWD)

print "Connecting to MQTT broker. Please wait..."
print("heap: " .. node.heap() )
m:connect( BROKER , BRPORT, 0, function(conn)
print("Connected to MQTT:" .. BROKER .. ":" .. BRPORT .." as " .. CLIENTID )
mqtt_sub() --run the subscription function
end)

function mqtt_sub()
if table.getn(topics) < current_topic then
run_main_prog()
else
m:subscribe(topics[current_topic] , 0, function(conn)
end)
current_topic = current_topic + 1
tmr.alarm(5, topicsub_delay, 0, mqtt_sub )
end
end

function _publish_data()
if pub_head ~= pub_tail then
if pub_sem == 0 then
pub_sem = 1
m:publish( pub_topic[pub_head], pub_message[pub_head] ,0 ,0 , function(conn)
print("Sending to " .. pub_topic[pub_head] ..": " .. pub_message[pub_head])
pub_sem = 0
pub_topic[pub_head] = nil
pub_message[pub_head] = nil
pub_head = pub_head + 1
end)
end
end
end

function publish_data(topic , message )
pub_topic[pub_tail] = topic
pub_message[pub_tail] = message
pub_tail = pub_tail + 1
_publish_data()
end

function publish_data1()
publish_data("Out1","Mensagem para out1" )
publish_data("Out2","Mensagem para out2" )
end

function run_main_prog()
print("Main program")

tmr.alarm(2, 5000, 1, publish_data1 )

-- Callback to receive the subscribed topic messages.
m:on("message", function(conn, topic, data)
print(topic .. ":" )
if (data ~= nil ) then
print ( data )
end
end )
end

ESP8266, NodeMcu and MQTT: Event subscription load testing

I’ve being doing some testing with the latest NodeMcu regarding MQTT support. My experience is, more or less, documented here: https://primalcortex.wordpress.com/2015/02/06/nodemcu-and-mqtt-how-to-start/ .

I wanted to check if concurrent subscription seems to work fine, since I know that publishing is not straightforward. In the code from the above link, the esp8266 subscribes to four topics, named topic1, topic2 and so on.

This simple bash script, calls repeatedly the mosquitto broker publishing command:

#!/bin/bash
for i in {1..60}
do
/home/pcortex/01.Develop/mosquitto/mosquitto-1.3.5/client/mosquitto_pub -h 192.168.1.16 -m "Mensagem $i" -t topic1
/home/pcortex/01.Develop/mosquitto/mosquitto-1.3.5/client/mosquitto_pub -h 192.168.1.16 -m "Mensagem $i" -t topic2
done

But with low values on the loop count, the MQTT code on the esp8266 seems to work fine, but with higher values, as above (60), the esp8266 hangs and reboots. One interesting thing was that after some testing not even doing a reset or completely powering it off, I was able to bring the esp8266 back online. It just put out garbage on the serial output and that was it. Strange indeed.

So I’ve reflash it again with the latest NodeMcu version (lots of reflashing hey 🙂 ), but still the esp8266 did not recover! Only garbage out when connecting to the serial port (And no, it wasn’t the serial port rate issue). So I’ve reflashed again the original AT Firmware, and it came back to life again, and then reflash it again with the latest Nodemcu version.

Adding a delay on the bash script, before the done line, like sleep 1 seems to keep the esp8266 running without issue.

So indeed this shows, that bombarding the esp8266 with MQTT messages in burst mode is not a good idea…

Synology DS: Cross compiling Eclipse/IBM RSMB MQTT broker

RSMB: Really Small Message Broker is an MQTT broker/server, simpler than the alternatives like Mosquitto, RabiitMQ, and so on. While it has its origins in IBM labs (as the MQTT protocol), it is now a project/sub project on Eclipse Paho MQTT related software. While downloading the RSBM does provide some binaries for some common platforms, it doesn’t offer any binaries for, in my case, the DS212+ Marvell 88F628x ARM processor.

So let’s see how to cross compile the RSMB for Synology DS and in this process also learn how to cross compile in your desktop computer native software for the Diskstation.

Requirements:

Setting up the cross compiling environment:

First, a read of the following document The 3rd party developer guide from Synology located here https://www.synology.com/en-global/support/developer#tool  is recommended. Based on this document (page 6 and 7)and on this page http://forum.synology.com/wiki/index.php/What_kind_of_CPU_does_my_NAS_have, we can know what version of the Synology tool chain we need to download from here:  http://sourceforge.net/projects/dsgpl/files/DSM%205.0%20Beta%20Tool%20Chains/

Download the required tool chain for your Synology version. In my case I have the Synology DS212+ that has the Marvel 88F628x ARM processor, so download this file: http://sourceforge.net/projects/dsgpl/files/DSM%205.0%20Beta%20Tool%20Chains/Marvell%2088F628x%20Linux%202.6.32/

Uncompress the file into the /usr/local directory. DO USE this directory. The tool chain is configured to get all files, libraries and so on from the /usr/local/… directory:

sudo tar xvzf gcc464_glibc215_88f6281-GPL.tgz -C /usr/local/

(Note: It’s a CAPITAL C. Check Synology documentation).

We can now get the RSMB sources.

Cross compiling RSMB:

Open an shell terminal (preferably bash but other shells might work) and create and change to a working directory. Clone the RSMB repository located in http://git.eclipse.org/gitroot/mosquitto/org.eclipse.mosquitto.rsmb.git with git tools:

mkdir work_dir
cd workdir
git clone http://git.eclipse.org/gitroot/mosquitto/org.eclipse.mosquitto.rsmb.git
cd org.eclipse.mosquitto.rsmb/rsmb/src/

While, for this case, not all the below settings are needed, for documentation purposes I document them here:

export INSTALLDIR=/usr/local/arm-marvell-linux-gnueabi
export PATH=$INSTALLDIR/bin:$PATH
export TARGETMACH=arm-marvell-linux-gnueabi
export BUILDMACH=i686-pc-linux-gnu
export CROSS=arm-marvell-linux-gnueabi
export CC=${CROSS}-gcc
export LD=${CROSS}-ld
export AS=${CROSS}-as
export AR=${CROSS}-ar

Just make sure that the INSTALLDIR variable and TARGETMACH and CROSS variables point to the correct settings.

Also, in this case, for compiling RSMB, we need also to add the following variable:

export GCC=${CROSS}-gcc

Otherwise we need to change the Makefile and change the line GCC=gcc to point to the correct compiler. We can compile now:

make

And we should have the broker executable among others.

Let’s make sure that it is ok:

pcortex@dune:~/01.Develop/org.eclipse.mosquitto.rsmb/rsmb/src$ file broker
broker: ELF 32-bit LSB  executable, ARM, EABI5 version 1 (SYSV), dynamically linked (uses shared libs), for GNU/Linux 2.6.16, stripped

If the output is this:

broker: ELF 64-bit LSB  executable, x86-64, version 1 (SYSV), dynamically linked (uses shared libs), for GNU/Linux 2.6.24, BuildID[sha1]=815abb3a1aad7f430c6e825670601c4991b45bd5, stripped

The wrong compiler was called.

Synology installation:

Copy the following files to your synology station: broker and Messages.1.3.0.2. From your workstation:

scp broker root@diskstation:/usr/local/bin
scp Messages.1.3.0.2 root@diskstation:/usr/local/bin

Access through ssh the Synology terminal, and make sure that broker is executable and do a test run:

cd /usr/local/bin
chmod +x broker
./broker
20150104 190523.162 CWNAN9999I Really Small Message Broker
20150104 190523.162 CWNAN9998I Part of Project Mosquitto in Eclipse
(http://projects.eclipse.org/projects/technology.mosquitto)
20150104 190523.163 CWNAN0053I Version 1.3.0.2, Jan  2 2015 20:13:39
20150104 190523.163 CWNAN0054I Features included: bridge
20150104 190523.163 CWNAN9993I Authors: Ian Craggs (icraggs@uk.ibm.com), Nicholas O'Leary
20150104 190523.163 CWNAN0014I MQTT protocol starting, listening on port 1883

And success! We can now test with MQTT-Spy (https://code.google.com/p/mqtt-spy/), Android Client, or Eclipse Paho tools.

Configuration and start and stopping:

For configuring the RSMB, we really should really read the documentation… 🙂 that is provided…

A simple configuration file should be located at /usr/local/etc and named rsmb.conf with the following basic contents:

# sample configuration on port 1883 with retained persistence in /tmp
port 1883
max_inflight_messages 50
max_queued_messages 200
persistence_location /tmp/
retained_persistence true

And at the /usr/local/etc/rc.d create a file named S99rsmb.sh with the following content:

#!/bin/sh

case $1 in
start)
    nohup /usr/local/bin/broker /usr/local/etc/rsmb.conf >> /var/log/rsmb.log&
    /usr/bin/logger -p1 -t "rsmb: INFO  " " Service started."
;;
stop)
    /usr/bin/killall broker
    /usr/bin/logger -p1 -t "rsmb: INFO  " " Service stopped."
;;
esac

Save and chmod +x S99rsmb.sh

Now the broker should start and stop automatically.

Final notes:

To use the broker from the Internet the port 1883/TCP should be open/forward at your router/firewall, and you should add authentication to the MQTT broker.