The Things Network MQTT & Azure IoT Part1

Side by Side

In my last few posts I have built Proof of Concept(PoC) The Things Network(TTN) V3 Hypertext Transfer Protocol(HTTP) API Integration and Message Queue Telemetry Transport (MQTT) Data API Integrations.

While building these PoCs I have learnt a lot about the way that the TTN V3 RESTful and MQTT APIs work and this is the first in a series of posts about linking them together. My plan is to start with yet another .NetCore Console application which hosts both the MQTT and Azure IoT Hub DeviceClient (using the Advanced Message Queueing Protocol(AMQP)) client implementations. I’m using MQTTnet to build my data API client and used NSwag by Richo Suter to generate my RESTful client from the TTN provided swagger file.

In this PoC I’m using the commandlineParser NuGet package to the reduce the amount of code required to process command line parameters and make it more robust. This PoC has a lot of command line parameters which would have been painful to manually parse and validate.

public class CommandLineOptions
{
	[Option('u', "APIbaseURL", Required = false, HelpText = "TTN Restful API URL.")]
	public string ApiBaseUrl { get; set; }

	[Option('K', "APIKey", Required = true, HelpText = "TTN Restful API APIkey")]
	public string ApiKey { get; set; }

	[Option('P', "APIApplicationID", Required = true, HelpText = "TTN Restful API ApplicationID")]
	public string ApiApplicationID { get; set; }

	[Option('D', "DeviceListPageSize", Required = true, HelpText = "The size of the pages used to retrieve EndDevice configuration")]
	public int DevicePageSize { get; set; }

	[Option('S', "MQTTServerName", Required = true, HelpText = "TTN MQTT API server name")]
	public string MqttServerName { get; set; }

	[Option('A', "MQTTAccessKey", Required = true, HelpText = "TTN MQTT API access key")]
	public string MqttAccessKey { get; set; }

	[Option('Q', "MQTTApplicationID", Required = true, HelpText = "TTN MQTT API ApplicationID")]
	public string MqttApplicationID { get; set; }

	[Option('C', "MQTTClientName", Required = true, HelpText = "TTN MQTT API Client ID")]
	public string MqttClientID { get; set; }

	[Option('Z', "AzureIoTHubConnectionString", Required = true, HelpText = "Azure IoT Hub Connection string")]
	public string AzureIoTHubconnectionString { get; set; }
}

To keep things simple in this PoC I’m using an Azure IoT Hub specific (rather than a device specific connection string)

Azure IoT Hub Device shared access policy selection

After some trial and error I found the order of execution was important

  • Open MQTTnet connection to TTN host (but don’t configure any subscriptions)
  • Configure connection to TTN RESTful API
  • Retrieve list of V3EndDevices (paginated), then for each V3EndDevice
    • Open connection to Azure IoT Hub using command line connection string + TTN Device ID
    • Call DeviceClient.SetReceiveMessageHandlerAsync to specify ReceiveMessageCallback and additional context information for processing Azure IoT Hub downlink messages.
    • Store DeviceClient instance in ObjectCache using DeviceID as key
  • Configure the MQTTnet recived message handler
  • Subscribe to uplink messages from all the V3EndDevices in the specified application.
private static async Task ApplicationCore(CommandLineOptions options)
{
	MqttFactory factory = new MqttFactory();
	mqttClient = factory.CreateMqttClient();

#if DIAGNOSTICS
	Console.WriteLine($"baseURL: {options.ApiBaseUrl}");
	Console.WriteLine($"APIKey: {options.ApiKey}");
	Console.WriteLine($"ApplicationID: {options.ApiApplicationID}");
	Console.WriteLine($"AazureIoTHubconnectionString: {options.AzureIoTHubconnectionString}");
	Console.WriteLine();
#endif

	try
	{
		// First configure MQTT, open connection and wire up disconnection handler. 
		// Can't wire up MQTT received handler as at this stage AzureIoTHub devices not connected.
		mqttOptions = new MqttClientOptionsBuilder()
			.WithTcpServer(options.MqttServerName)
			.WithCredentials(options.MqttApplicationID, options.MqttAccessKey)
			.WithClientId(options.MqttClientID)
			.WithTls()
			.Build();

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

		await mqttClient.ConnectAsync(mqttOptions);

		// Prepare the HTTP client to be used in the TTN device enumeration
		using (HttpClient httpClient = new HttpClient())
		{
			EndDeviceRegistryClient endDeviceRegistryClient = new EndDeviceRegistryClient(options.ApiBaseUrl, httpClient)
			{
				ApiKey = options.ApiKey
			};

			// Retrieve list of devices page by page
			V3EndDevices endDevices = await endDeviceRegistryClient.ListAsync(
				options.ApiApplicationID, 
				field_mask_paths: DevicefieldMaskPaths, 
				limit: options.DevicePageSize);
			if ((endDevices != null) && (endDevices.End_devices != null)) // If no devices returns null rather than empty list
			{
				foreach (V3EndDevice endDevice in endDevices.End_devices)
				{
					// Display the device info+attributes then connect device to Azure IoT Hub
#if DEVICE_FIELDS_MINIMUM
					Console.WriteLine($"EndDevice ID: {endDevice.Ids.Device_id}");
#else
					Console.WriteLine($"Device ID: {endDevice.Ids.Device_id} Name: {endDevice.Name} Description: {endDevice.Description}");
					Console.WriteLine($"  CreatedAt: {endDevice.Created_at:dd-MM-yy HH:mm:ss} UpdatedAt: {endDevice.Updated_at:dd-MM-yy HH:mm:ss}");
#endif

#if DEVICE_ATTRIBUTES_DISPLAY
					if (endDevice.Attributes != null)
					{
						Console.WriteLine("  EndDevice attributes");

						foreach (KeyValuePair<string, string> attribute in endDevice.Attributes)
						{
							Console.WriteLine($"    Key: {attribute.Key} Value: {attribute.Value}");
						}
					}
#endif
					try
					{
						DeviceClient deviceClient = DeviceClient.CreateFromConnectionString(
							options.AzureIoTHubconnectionString, 
							endDevice.Ids.Device_id, 
							TransportType.Amqp_Tcp_Only);

						await deviceClient.OpenAsync();

						await deviceClient.SetReceiveMessageHandlerAsync(
							AzureIoTHubClientReceiveMessageHandler,
							new AzureIoTHubReceiveMessageHandlerContext()
							{
								DeviceId = endDevice.Ids.Device_id,
								ApplicationId = endDevice.Ids.Application_ids.Application_id,
							});

						DeviceClients.Add(endDevice.Ids.Device_id, deviceClient, cacheItemPolicy);
					}
					catch( Exception ex)
					{
						Console.WriteLine($"Azure IoT Hub OpenAsync failed {ex.Message}");
					}
				}
			}
		}

		// At this point all the AzureIoT Hub deviceClients setup and ready to go so can enable MQTT receive
		mqttClient.UseApplicationMessageReceivedHandler(new MqttApplicationMessageReceivedHandlerDelegate(e => MqttClientApplicationMessageReceived(e)));

		// This may shift to individual device subscriptions
		string uplinktopic = $"v3/{options.MqttApplicationID}/devices/+/up";

		await mqttClient.SubscribeAsync(uplinktopic, MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce);
	}
	catch(Exception ex)
	{
		Console.WriteLine($"Main {ex.Message}");
		Console.WriteLine("Press any key to exit");
		Console.ReadLine();
		return;
	}

	while (!Console.KeyAvailable)
	{
		Console.Write(".");
		await Task.Delay(1000);
	}

	// Consider ways to mop up connections

	Console.WriteLine("Press any key to exit");
	Console.ReadLine();
}

When I was initially looking at Azure Deviceclient I would of had to have created a thread (which would have been blocked most of the time) for each device. This implementation issued was removed by the introduction of the DeviceClient SetReceiveMessageHandlerAsync method in release 1.33.0.

Currently the application just displays the Cloud to Device(C2D) message payload plus diagnostic information, and the CompleteAsync method is called so the message is dequeued.

private async static Task AzureIoTHubClientReceiveMessageHandler(Message message, object userContext)
{
	AzureIoTHubReceiveMessageHandlerContext receiveMessageHandlerConext = (AzureIoTHubReceiveMessageHandlerContext)userContext;

	DeviceClient deviceClient = (DeviceClient)DeviceClients.Get(receiveMessageHandlerConext.DeviceId);

	using (message)
	{
		Console.WriteLine();
		Console.WriteLine();
		Console.WriteLine($"{DateTime.UtcNow:HH:mm:ss} Azure IoT Hub downlink message");
		Console.WriteLine($" ApplicationID: {receiveMessageHandlerConext.ApplicationId}");
		Console.WriteLine($" DeviceID: {receiveMessageHandlerConext.DeviceId}");
#if DIAGNOSTICS_AZURE_IOT_HUB
		Console.WriteLine($" Cached: {DeviceClients.Contains(receiveMessageHandlerConext.DeviceId)}");
		Console.WriteLine($" MessageID: {message.MessageId}");
		Console.WriteLine($" DeliveryCount: {message.DeliveryCount}");
		Console.WriteLine($" EnqueuedTimeUtc: {message.EnqueuedTimeUtc}");
		Console.WriteLine($" SequenceNumber: {message.SequenceNumber}");
		Console.WriteLine($" To: {message.To}");
#endif
		string messageBody = Encoding.UTF8.GetString(message.GetBytes());
		Console.WriteLine($" Body: {messageBody}");
#if DOWNLINK_MESSAGE_PROPERTIES_DISPLAY
		foreach (var property in message.Properties)
		{
			Console.WriteLine($"   Key:{property.Key} Value:{property.Value}");
		}
#endif

		await deviceClient.CompleteAsync(message);

		Console.WriteLine();
	}
}

Currently the application just displays the Cloud to Device(D2C) message payload plus diagnostic information, displaying the payload fields if the message format has been configured and successfully processed.

private static void MqttClientApplicationMessageReceived(MqttApplicationMessageReceivedEventArgs e)
{
	if (e.ApplicationMessage.Topic.EndsWith("/up"))
	{
		PayloadUplinkV3 payload = JsonConvert.DeserializeObject<PayloadUplinkV3>(e.ApplicationMessage.ConvertPayloadToString());

		Console.WriteLine();
		Console.WriteLine();
		Console.WriteLine($"{DateTime.UtcNow:HH:mm:ss} TTN Uplink message");
#if DIAGNOSTICS_MQTT
		Console.WriteLine($" ClientId:{e.ClientId} Topic:{e.ApplicationMessage.Topic}");
		Console.WriteLine($" Cached: {DeviceClients.Contains(payload.EndDeviceIds.DeviceId)}");
#endif
		Console.WriteLine($" ApplicationID: {payload.EndDeviceIds.ApplicationIds.ApplicationId}");
		Console.WriteLine($" DeviceID: {payload.EndDeviceIds.DeviceId}");
		Console.WriteLine($" Port: {payload.UplinkMessage.Port} ");
		Console.WriteLine($" Payload raw: {payload.UplinkMessage.PayloadRaw}");

		if (payload.UplinkMessage.PayloadDecoded != null)
		{
			Console.WriteLine($" Payload decoded: {payload.UplinkMessage.PayloadRaw}");
			EnumerateChildren(1, payload.UplinkMessage.PayloadDecoded);
		}

		Console.WriteLine();
	}
	else
	{
		Console.WriteLine($"{DateTime.UtcNow:HH:mm:ss} ClientId: {e.ClientId} Topic: {e.ApplicationMessage.Topic}");
	}
}
dotNet Core Console application displaying simulated uplink and downlink messages.
Simulating C2D messages with AzureIoTExplorer
Simulating D2C messages with TTN Device console

In the MQTT received message handler.

Console.WriteLine($" Cached: {DeviceClients.Contains(receiveMessageHandlerConext.DeviceId)}");

and Azure DeviceClient received message handler.

Console.WriteLine($" Cached: {DeviceClients.Contains(receiveMessageHandlerConext.DeviceId)}");

check that the specified TTN device ID is in the DeviceClients ObjectCache

One thought on “The Things Network MQTT & Azure IoT Part1

  1. Pingback: The Things Network MQTT & Azure IoT Part2 | devMobile's blog

Leave a Reply

Fill in your details below or click an icon to log in:

WordPress.com Logo

You are commenting using your WordPress.com account. Log Out /  Change )

Google photo

You are commenting using your Google account. Log Out /  Change )

Twitter picture

You are commenting using your Twitter account. Log Out /  Change )

Facebook photo

You are commenting using your Facebook account. Log Out /  Change )

Connecting to %s

This site uses Akismet to reduce spam. Learn how your comment data is processed.