Azure IoT Hub with MQTTnet

As I’m testing my Message Queue Telemetry Transport(MQTT) LoRa gateway I’m building a proof of concept(PoC) .Net core console application for each IoT platform I would like to support.

This PoC was to confirm that my device could connect to the Microsoft Azure IoT Hub MQTT API then format topics and payloads correctly.

Azure IoT Hub MQTT Console Client

I had tried with a couple of different MQTT libraries from micro controllers and embedded devices without success. With the benefit of hindsight (plus this article) I think I had the SAS key format wrong.

The Azure IoT Hub MQTT broker requires only a server name (fully resolved CName), device ID and SAS Key.

   class Program
   {
      private static IMqttClient mqttClient = null;
      private static IMqttClientOptions mqttOptions = null;
      private static string server;
      private static string username;
      private static string password;
      private static string clientId;
      private static string topicD2C;
      private static string topicC2D;

      static void Main(string[] args)
      {
         MqttFactory factory = new MqttFactory();
         mqttClient = factory.CreateMqttClient();

         if (args.Length != 3)
         {
            Console.WriteLine("[AzureIoTHubHostName] [deviceID] [SASKey]");
            Console.WriteLine("Press <enter> to exit");
            Console.ReadLine();
            return;
         }

         server = args[0];
         clientId = args[1];
         sasKey= args[2];

         username = $"{server}/{clientId}/api-version=2018-06-30";
         topicD2C = $"devices/{clientId}/messages/events/";
         topicC2D = $"devices/{clientId}/messages/devicebound/#";

         Console.WriteLine($"MQTT Server:{server} Username:{username} ClientID:{clientId}");

         mqttOptions = new MqttClientOptionsBuilder()
            .WithTcpServer(server, 8883)
            .WithCredentials(username, sasKey)
            .WithClientId(clientId)
            .WithTls()
            .Build();

         mqttClient.UseDisconnectedHandler(new MqttClientDisconnectedHandlerDelegate(e => MqttClient_Disconnected(e)));
         mqttClient.UseApplicationMessageReceivedHandler(new MqttApplicationMessageReceivedHandlerDelegate(e => MqttClient_ApplicationMessageReceived(e)));
         mqttClient.ConnectAsync(mqttOptions).Wait();

         mqttClient.SubscribeAsync(topicC2D, MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce).GetAwaiter().GetResult();

         while (true)
         {
            JObject payloadJObject = new JObject();

            payloadJObject.Add("OfficeTemperature", "22." + DateTime.UtcNow.Millisecond.ToString());
            payloadJObject.Add("OfficeHumidity", (DateTime.UtcNow.Second + 40).ToString());

            string payload = JsonConvert.SerializeObject(payloadJObject);
            Console.WriteLine($"Topic:{topicD2C} Payload:{payload}");

            var message = new MqttApplicationMessageBuilder()
               .WithTopic(topicD2C)
               .WithPayload(payload)
               .WithAtLeastOnceQoS()
            .Build();

            Console.WriteLine("PublishAsync start");
            mqttClient.PublishAsync(message).Wait();
            Console.WriteLine("PublishAsync finish");

            Thread.Sleep(30100);
         }
      }

      private static void MqttClient_ApplicationMessageReceived(MqttApplicationMessageReceivedEventArgs e)
      {
         Console.WriteLine($"ClientId:{e.ClientId} Topic:{e.ApplicationMessage.Topic} Payload:{e.ApplicationMessage.ConvertPayloadToString()}");
      }

      private static async void MqttClient_Disconnected(MqttClientDisconnectedEventArgs e)
      {
         Debug.WriteLine("Disconnected");
         await Task.Delay(TimeSpan.FromSeconds(5));

         try
         {
            await mqttClient.ConnectAsync(mqttOptions);
         }
         catch (Exception ex)
         {
            Debug.WriteLine("Reconnect failed {0}", ex.Message);
         }
      }
   }

Overall the initial configuration went smoothly after I figured out the required Quality of Service (QoS) settings, and the SAS Key format.

Using the approach described in the Microsoft documentation I manually generated the SAS Key.(In my Netduino samples I have code for generating a SAS Key in my HTTPS Azure IoT Hub Client)

Azure Device Explorer Device Management
Azure Device Explorer SAS Key Generator

Once I had the configuration correct I could see telemetry from the device and send it messages.

Azure Device Explorer Data View

In a future post I will upload data to the Azure IoT Central for display. Then explore using a “module” attached to a device which maybe useful for my field gateway.

Thingspeak with MQTTnet

As I’m testing my Message Queue Telemetry Transport(MQTT) LoRa gateway I’m building a proof of concept(PoC) .Net core console application for each IoT platform I would like to support.

This PoC was to confirm that I could connect to the Thingspeak MQTT API then format topics and payloads correctly.

MQTT Console Client

The Thingspeak MQTT broker, username, password, and client ID, Channel, writeAPIKey, ClientID, channelData, are the minimum command line options required. The channelSubscribe, readAPIkey and field are optional.

   class Program
   {
      private static IMqttClient mqttClient = null;
      private static IMqttClientOptions mqttOptions = null;
      private static string server;
      private static string username;
      private static string password;
      private static string writeApiKey;
      private static string clientId;
      private static string channelData;
      private static string channelSubscribe;
      private static string readApiKey;
      private static string field;

      static void Main(string[] args)
      {
         MqttFactory factory = new MqttFactory();
         mqttClient = factory.CreateMqttClient();

         if ((args.Length != 6) && (args.Length != 8) && (args.Length != 9))
         {
            Console.WriteLine("[MQTT Server] [UserName] [Password] [WriteAPIKey] [ClientID] [Channel]");
            Console.WriteLine("[MQTT Server] [UserName] [Password] [WriteAPIKey] [ClientID] [Channel] [channelSubscribe] [ReadApiKey]");
            Console.WriteLine("[MQTT Server] [UserName] [Password] [WriteAPIKey] [ClientID] [Channel] [channelSubscribe] [ReadApiKey] [Field]");
            Console.WriteLine("Press <enter> to exit");
            Console.ReadLine();
            return;
         }

         server = args[0];
         username = args[1];
         password = args[2];
         writeApiKey = args[3];
         clientId = args[4];
         channelData = args[5];

         if (args.Length == 6)
         {
            Console.WriteLine($"MQTT Server:{server} Username:{username} ClientID:{clientId} ChannelData:{channelData}");
         }

         if (args.Length == 8)
         {
            channelSubscribe = args[6];
            readApiKey = args[7];
            Console.WriteLine($"MQTT Server:{server} Username:{username} ClientID:{clientId} ChannelData:{channelData} ChannelSubscribe:{channelSubscribe}");
         }

         if (args.Length == 9)
         {
            channelSubscribe = args[6];
            readApiKey = args[7];
            field = args[8];
            Console.WriteLine($"MQTT Server:{server} Username:{username} ClientID:{clientId} ChannelData:{channelData} ChannelSubscribe:{channelSubscribe} Field:{field}");
         }

         mqttOptions = new MqttClientOptionsBuilder()
            .WithTcpServer(server)
            .WithCredentials(username, password)
            .WithClientId(clientId)
            .WithTls()
            .Build();

         mqttClient.ConnectAsync(mqttOptions).Wait();

         if (args.Length == 8)
         {
            string topic = $"channels/{channelSubscribe}/subscribe/fields/+/{readApiKey}";
            
            Console.WriteLine($"Subscribe Topic:{topic}");

            mqttClient.SubscribeAsync(topic, MQTTnet.Protocol.MqttQualityOfServiceLevel.AtMostOnce).Wait(); 
            mqttClient.UseApplicationMessageReceivedHandler(new MqttApplicationMessageReceivedHandlerDelegate(e => MqttClient_ApplicationMessageReceived(e)));
         }
         if (args.Length == 9)
         {
            string topic = $"channels/{channelSubscribe}/subscribe/fields/{field}/{readApiKey}";

            Console.WriteLine($"Subscribe Topic:{topic}");

            mqttClient.SubscribeAsync(topic, MQTTnet.Protocol.MqttQualityOfServiceLevel.AtMostOnce).Wait();
            mqttClient.UseApplicationMessageReceivedHandler(new MqttApplicationMessageReceivedHandlerDelegate(e => MqttClient_ApplicationMessageReceived(e)));
         }

         mqttClient.UseDisconnectedHandler(new MqttClientDisconnectedHandlerDelegate(e => MqttClient_Disconnected(e)));

         string topicTemperatureData = $"channels/{channelData}/publish/{writeApiKey}";

         Console.WriteLine();

         while (true)
         {
            string value = "field1=22." + DateTime.UtcNow.Millisecond.ToString() + "&field2=60." + DateTime.UtcNow.Millisecond.ToString();
            Console.WriteLine($"Publish Topic {topicTemperatureData}  Value {value}");

            var message = new MqttApplicationMessageBuilder()
               .WithTopic(topicTemperatureData)
               .WithPayload(value)
               .WithQualityOfServiceLevel(MQTTnet.Protocol.MqttQualityOfServiceLevel.AtMostOnce)
//               .WithRetainFlag()
            .Build();

            Console.WriteLine("PublishAsync start");
            mqttClient.PublishAsync(message).Wait();
            Console.WriteLine("PublishAsync finish");
            Console.WriteLine();

            Thread.Sleep(30100);
         }
      }

      private static void MqttClient_ApplicationMessageReceived(MqttApplicationMessageReceivedEventArgs e)
      {
         Console.WriteLine($"ApplicationMessageReceived ClientId:{e.ClientId} Topic:{e.ApplicationMessage.Topic} Qos:{e.ApplicationMessage.QualityOfServiceLevel} Payload:{e.ApplicationMessage.ConvertPayloadToString()}");
         Console.WriteLine();
      }

      private static async void MqttClient_Disconnected(MqttClientDisconnectedEventArgs e)
      {
         Debug.WriteLine("Disconnected");
         await Task.Delay(TimeSpan.FromSeconds(5));

         try
         {
            await mqttClient.ConnectAsync(mqttOptions);
         }
         catch (Exception ex)
         {
            Debug.WriteLine("Reconnect failed {0}", ex.Message);
         }
      }
   }
Channel configuration

The ThingSpeak channel setup has many attributes (link to an external site, link to a GitHub repository, fields for elevation, latitude, longitude etc. ) but only 8 data fields which seems a bit limiting (some of my sensor nodes report more than 8 values).

Dashboard

The Thingspeak dashboard configuration was fairly simple, though the maximum of eight sequentially numbered of inputs (Fields) might require some mapping code.

Overall the initial configuration went smoothly after I figured out the required Quality of Service (QoS) settings, retain flag usage and the different APIKeys on the publish vs. subscribe topics.

I have not explored the advanced analysis enabled by the tight integration with MATLAB which could be quite an advantage for applications requiring that sort of functionality.

Ask Sensors with MQTTnet

After a 6 month pause I’m back working on my Message Queue Telemetry Transport(MQTT) LoRa gateway.

As I’m testing my Message Queue Telemetry Transport(MQTT) LoRa gateway I’m building a proof of concept(PoC) .Net core console application for each IoT platform I would like to support.

This PoC was to confirm that I could connect to the Ask Sensors MQTT API then format topics and payloads correctly.

Console test application

The MQTT broker, username, password, and client ID are command line options.

class Program
{
	private static IMqttClient mqttClient = null;
	private static IMqttClientOptions mqttOptions = null;
	private static string server;
	private static string username;
	private static string apiKey;
	private static string clientID;

	static void Main(string[] args)
	{
		MqttFactory factory = new MqttFactory();
		mqttClient = factory.CreateMqttClient();
		bool heatPumpOn = false;

		if (args.Length != 4)
		{
			Console.WriteLine("[MQTT Server] [UserName] [APIKey] [ClientID]");
			Console.WriteLine("Press <enter> to exit");
			Console.ReadLine();
			return;
		}

		server = args[0];
		username = args[1];
		apiKey = args[2];
		clientID = args[3];

		Console.WriteLine($"MQTT Server:{server} Username:{username} ClientID:{clientID}");

		mqttOptions = new MqttClientOptionsBuilder()
			.WithTcpServer(server)
			.WithCredentials(username, "")
			.WithClientId(clientID)
			//.WithTls() // This is a bit of a worry
			.Build();

		mqttClient.ApplicationMessageReceived += MqttClient_ApplicationMessageReceived;
		mqttClient.Disconnected += MqttClient_Disconnected;
		mqttClient.ConnectAsync(mqttOptions).Wait();

		// AskSensors formatted client state update topic
		string stateTopic = $"{username}/{apiKey}";

		while (true)
		{
			string payloadText;
			double temperature = 22.0 + (DateTime.UtcNow.Millisecond / 1000.0);
			double humidity = 50 + (DateTime.UtcNow.Millisecond / 100.0);
			double speed = 10 + (DateTime.UtcNow.Millisecond / 100.0);
			Console.WriteLine($"Topic:{stateTopic} Temperature:{temperature:0.00} Humidity:{humidity:0} HeatPumpOn:{heatPumpOn}");

			// First JSON attempt didn't work
			payloadText = @"{""Humidity"":55}";

            // Second attempt worked
            payloadText = $"module1=22";

            // Third attempt with "real" values injected
            payloadText = $"module1={temperature}&m2={humidity}";

            var message = new MqttApplicationMessageBuilder()
					.WithTopic(stateTopic)
					.WithPayload(payloadText)
					.WithQualityOfServiceLevel(global::MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce)
				   .WithExactlyOnceQoS()
				   //.WithAtLeastOnceQoS()
				   //.WithRetainFlag()
				   .Build();

				Console.WriteLine("PublishAsync start");
				mqttClient.PublishAsync(message).Wait();
				Console.WriteLine("PublishAsync finish");

				Thread.Sleep(30100);
			}
		}

	private static async void MqttClient_Disconnected(object sender, MqttClientDisconnectedEventArgs e)
	{
		Debug.WriteLine("Disconnected");
		await Task.Delay(TimeSpan.FromSeconds(5));

		try
		{
			await mqttClient.ConnectAsync(mqttOptions);
		}
		catch (Exception ex)
		{
			Debug.WriteLine("Reconnect failed {0}", ex.Message);
		}
	}

The Ask Sensors screen designer has 8 different types of Graph (line, bar, scatter, gauge, table, binary. digital, map)

Ask sensors dashboard configuration was fairly simple, though sequential numbering of inputs (modules) might require some mapping code.

Overall the initial configuration went smoothly after I figured out the payload format (not JSON), though the functionality (of a free subscription) did appear to be quite limited.

Since I first started building my MQTT gateway there have been several breaking updates to the MQTTNet API which so I will have to refresh all the applications in my solution.

Ubidots with MQTTnet

As I’m testing my Message Queue Telemetry Transport(MQTT) LoRa gateway I’m building a proof of concept(PoC) .Net core console application for each IoT platform I would like to support.

This PoC was to confirm that I could connect to the ubidots MQTT API then format the topics and payloads correctly. The ubidots screen designer has “variables” (both actual sensors & synthetic calculated ones) which present as topics so I built a client which could subscribe to these.

.Net Core V2 MQTTnet client

The MQTT broker, username, password, and client ID are command line options.

class Program
{
	private static IMqttClient mqttClient = null;
	private static IMqttClientOptions mqttOptions = null;
	private static string server;
	private static string username;
	private static string deviceLabel;

	static void Main(string[] args)
	{
		MqttFactory factory = new MqttFactory();
		mqttClient = factory.CreateMqttClient();
		bool heatPumpOn = false;

		if (args.Length != 3)
		{
			Console.WriteLine("[MQTT Server] [UserName] [Password] [ClientID]");
			Console.WriteLine("Press <enter> to exit");
			Console.ReadLine();
			return;
		}

		server = args[0];
		username = args[1];
		deviceLabel = args[2];

		Console.WriteLine($"MQTT Server:{server} Username:{username} DeviceLabel:{deviceLabel}");

		mqttOptions = new MqttClientOptionsBuilder()
			.WithTcpServer(server)
			.WithCredentials(username, "NotVerySecret")
			.WithClientId(deviceLabel)
			.WithTls()
			.Build();

		mqttClient.ApplicationMessageReceived += MqttClient_ApplicationMessageReceived;
		mqttClient.Disconnected += MqttClient_Disconnected;
		mqttClient.ConnectAsync(mqttOptions).Wait();

		// Setup a subscription for commands sent to client
		string commandTopic = $"/v1.6/devices/{deviceLabel}/officetemperaturedesired/lv";
		mqttClient.SubscribeAsync(commandTopic).GetAwaiter().GetResult();

		//// Ubidots formatted client state update topic
		string stateTopic = $"/v1.6/devices/{deviceLabel}";

		while (true)
		{
			string payloadText;
			double temperature = 22.0 + (DateTime.UtcNow.Millisecond / 1000.0);
			double humidity = 50 + (DateTime.UtcNow.Millisecond / 100.0);
			double speed = 10 + (DateTime.UtcNow.Millisecond / 100.0);
			Console.WriteLine($"Topic:{stateTopic} Temperature:{temperature:0.00} Humidity:{humidity:0} HeatPumpOn:{heatPumpOn}");

			// First attempt which worked
			//payloadText = @"{""OfficeTemperature"":22.5}";

			// Second attempt to work out data format with "real" values injected
			//payloadText = @"{ ""officetemperature"":"+ temperature.ToString("F2") + @",""officehumidity"":" + humidity.ToString("F0") + @"}";

			// Third attempt with Jobject which sort of worked but number serialisation was sub optimal
			JObject payloadJObject = new JObject(); 
			payloadJObject.Add("OfficeTemperature", temperature.ToString("F2"));
			payloadJObject.Add("OfficeHumidity", humidity.ToString("F0"));

			if (heatPumpOn)
			{
				payloadJObject.Add("HeatPumpOn", 1);
			}
			else
			{
				payloadJObject.Add("HeatPumpOn", 0);
			}
			heatPumpOn = !heatPumpOn;
			payloadText = JsonConvert.SerializeObject(payloadJObject);

			/*
			// Forth attempt with JOBject, timestamps and gps 
			JObject payloadJObject = new JObject();
			JObject context = new JObject();
			context.Add("lat", "-43.5309325");
			context.Add("lng", "172.637119");// Christchurch Cathederal
			//context.Add("timestamp", ((DateTimeOffset)(DateTime.UtcNow)).ToUnixTimeSeconds()); // This field is optional and can be commented out
			JObject position = new JObject();
			position.Add("context", context);
			position.Add("value", "0");
			payloadJObject.Add("postion", position);
			payloadText = JsonConvert.SerializeObject(payloadJObject);
			*/

			var message = new MqttApplicationMessageBuilder()
				.WithTopic(stateTopic)
				.WithPayload(payloadText)
				.WithQualityOfServiceLevel(global::MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce)
			//.WithExactlyOnceQoS()// With ubidots this caused the publish to hang
			.WithAtLeastOnceQoS()
			.WithRetainFlag() 
			.Build();

			Console.WriteLine("PublishAsync start");
			mqttClient.PublishAsync(message).Wait();
			Console.WriteLine("PublishAsync finish");

			Thread.Sleep(30100);
		}
	}

	private static void MqttClient_ApplicationMessageReceived(object sender, MqttApplicationMessageReceivedEventArgs e)
	{
		Console.WriteLine($"ClientId:{e.ClientId} Topic:{e.ApplicationMessage.Topic} Payload:{e.ApplicationMessage.ConvertPayloadToString()}");
	}

	private static async void MqttClient_Disconnected(object sender, MqttClientDisconnectedEventArgs e)
	{
		Debug.WriteLine("Disconnected");
		await Task.Delay(TimeSpan.FromSeconds(5));

		try
		{
			await mqttClient.ConnectAsync(mqttOptions);
		}
		catch (Exception ex)
		{
			Debug.WriteLine("Reconnect failed {0}", ex.Message);
		}
	}
}

For this PoC I used the MQTTnet package which is available via NuGet. It appeared to be reasonably well supported and has had recent updates.

Variable configuration with device location map

Overall the initial configuration went smoothly, I found the dragging of blocks onto the dashboard and configuring them worked as expected.

The configuration of a “synthetic” variable (converting a temperature to Fahrenheit for readers from the Unites States of America, Myanmar & Liberia ) took a couple of goes to get right.

I may have missed something (April 2019) but the lack of boolean datatype variables was a bit odd.

Synthetic (calculated) variable configuration

I put a slider control on my test dashboard, associated it with a variable and my client reliably received messages when the slider was moved.

Dashboard with slider for desired temperature

Overall the Ubidots experience was pretty good and I’m going to spend some more time working with the device, data, users configurations to see how well it works for a “real-world” project.

I found (April 2019) that to get MQTTS going I had to install a Ubidots provided certificate

MQTT with TLS guidance and certificate download link

When my .Net Core application didn’t work I tried one my MQTT debugging tools and they didn’t work either with the Ubitdots MQTT brokers. The Ubidots forum people were quite helpful, but making it not necessary to install a certificate or making it really obvious in the documentation that this was required would be a good thing.