To add MQTT support to our sensor, we will use the MqttClient class defined in the Clayster.Library.Internet.MQTT namespace. We start by adding the following namespace to our using section in the code:
using Clayster.Library.Internet.MQTT;
Communication with the MQTT server will be done from a separate thread in our example. This is to assure that we avoid timing problems with the measurement logic. When new values are available, we flag this fact to the MQTT thread using auto-reset event objects. So, we need the following static variables:
private static Thread mqttThread = null; private static AutoResetEvent mqttNewTemp = new AutoResetEvent(false); private static AutoResetEvent mqttNewLight = new AutoResetEvent(false); private static AutoResetEvent mqttNewMotion = new AutoResetEvent(false);
The application will then publish values to MQTT topics if a significant change has occurred, or when the given time has passed since its last publication. So, we can create variables for the last published value as well as the time when the last publication occurred.
private static double mqttLastTemp = 0; private static double mqttLastLight = 0; private static bool mqttLastMotion = false; private static DateTime mqttLastTempPublished = DateTime.MinValue; private static DateTime mqttLastLightPublished = DateTime.MinValue; private static DateTime mqttLastMotionPublished = DateTime.MinValue;
Before the main loop, we start the MQTT thread in the usual manner, making sure to set the thread priority to BelowNormal so that it does not affect the way it is normally executed:
mqttThread = new Thread (MqttThread); mqttThread.Name = "MQTT"; mqttThread.Priority = ThreadPriority.BelowNormal; mqttThread.Start();
Note
We create our thread using the thread priority just below the normal (BelowNormal) priority. This means the thread will not interfere with normal operation in case the device CPU usage reaches 100 percent. When CPU usage is below 100 percent, this thread will work as a normal thread. Since communication is normally dropped when this happens, this does not imply loss of real functionality.
When we close the application, we must also make sure the thread is closed properly. We use the Abort() method on the thread to accomplish this:
if(mqttThread != null)
{
mqttThread.Abort ();
mqttThread = null;
}In the SampleSensorValues() method where we sample new sensor values, we need to detect significant events that the MQTT thread needs to react to. We can start with the motion detector. After it notifies any CoAP subscribers, we also need to signal the MQTT thread that the Boolean value has changed:
if (MotionChanged)
{
if (motionTxt != null)
motionTxt.NotifySubscribers ();
mqttNewMotion.Set ();
mqttLastMotionPublished = Now;
mqttLastMotion = motionDetected;However, we also need to republish the value if it has been a long time since a value was published, which can be done with the following code:
}
else if((Now - mqttLastMotionPublished).TotalMinutes >= 10)
{
mqttNewMotion.Set ();
mqttLastMotionPublished = Now;
mqttLastMotion = motionDetected;
}Significant events for Boolean values are easy to define. But what is a significant event for a numerical value? In our implementation, a significant event is if the temperature change is more than half a degree centigrade or if more than ten minutes has passed. Here, what constitutes the word "significant" depends on what type of temperature we are measuring. This limit could be configurable if the context is not clear.
In the same way, we define a significant event for the light sensor as a change in one unit of a percent or ten minutes since it was last published, whichever comes first:
if ((Now - mqttLastLightPublished).TotalMinutes >= 10 || System.Math.Abs (lightPercent - mqttLastLight) >= 1.0)
{
mqttNewLight.Set ();
mqttLastLightPublished = Now;
mqttLastLight = lightPercent;
}Tip
Since the request/response pattern is difficult to implement using MQTT (you would have to invent separate topics to send requests on), you need a method to notify subscribers of the current status of the sensor as well as tell them that you are alive and well. One way to accomplish this is to, with some regularity, publish the current status, even if the change from the last published value is not great or doesn't exist at all.
From our communication thread, we use the MqttClient class to communicate with the MQTT server:
MqttClient Client = null;
In the following example, we will use a publicly available MQTT server hosted by eclipse.org. It allows anybody to connect and publish information, so we simply provide a username for appearance's sake and leave the password empty. The last Boolean parameter specifies whether we want to use SSL/TLS encryption or not. In our case, we will not bother with encryption of the connection since the data will be publicly available on the Internet anyway:
if (Client == null)
{
Client = new MqttClient ("iot.eclipse.org", MqttClient.DefaultPort, "LearningIoTSensor", string.Empty, false);We then open the connection and log in to the server. In the CONNECT() method, you need to specify the keepalive time in seconds and whether the connection is a clean connection or not. A clean connection discards any pending notifications stored on the server. If you wish to reconnect to the server, you can choose not to use a clean connection. The server will then send you any notifications it has stored in the session, while you were not connected, if the session has not been timed out and removed.
Client.Open (); Client.CONNECT (20, true);
Finally, it's a good idea to log an event in the event log indicating that the MQTT connection is active:
Log.Information ("Publishing via MQTT to " + "Clayster/LearningIoT/Sensor @ ", EventLevel.Minor, Client.Host + ":" + Client.PortNumber.ToString ());
}Note
You can install and host your own MQTT message brokers if you want to. You can find several of these to choose from via https://github.com/mqtt/mqtt.github.io/wiki/servers.
The application flags significant events to consider using event objects. So we first need to create an array of the available events to monitor:
WaitHandle[] Events = new WaitHandle[]
{
mqttNewTemp, mqttNewLight, mqttNewMotion
};In our infinite loop, we then wait for any of the events to occur:
switch (WaitHandle.WaitAny (Events, 1000))
{
}We begin by publishing temperature information, if such an event is detected. Here we publish the current temperature using the acknowledged message service as a string with one decimal, suffixed by a space and the unit C:
case 0:// New temperature
Client.PUBLISH("Clayster/LearningIoT/Sensor/Temperature", FieldNumeric.Format (temperatureC, "C", 1), MqttQoS.QoS1_Acknowledged, true);
break;Tip
MQTT is a binary protocol, and it does not support the encoding or decoding of content. We must keep track of encoding and decoding ourselves. The MqttClient library provides you with a PUBLISH() method that allows you to publish binary content. It also has overrides that allow you to publish text and XML content using simple UTF-8 encoding.
Similarly, we will publish the current light density as a string with one decimal suffixed by a space and a percent character (%):
case 1:// New light
Client.PUBLISH ("Clayster/LearningIoT/Sensor/Light", FieldNumeric.Format (lightPercent, "%", 1),MqttQoS.QoS1_Acknowledged, true);
break;The motion detector only contains a Boolean value. We publish this value as either a string containing the digit 1 if motion is detected or 0 if not.
case 2:// New motion
Client.PUBLISH ("Clayster/LearningIoT/Sensor/Motion", motionDetected ? "1" : "0", MqttQoS.QoS1_Acknowledged, true);
break;