The Things Network MQTT & Azure IoT Part3A

Cloud to Device with frm_payload no confirmation

An Azure IoT Hub supports three kinds for Cloud to Device(C2D) messaging and my gateway will initially support only Direct Methods and Cloud-to-device messages.

The first step was to add the The Things Network(TTN) V3 Tennant ID to the context information as it is required for the downlink Message Queue Telemetry Transport (MQTT) publish topic.

namespace devMobile.TheThingsNetwork.Models
{
   public class AzureIoTHubReceiveMessageHandlerContext
   {
      public string TenantId { get; set; }
      public string DeviceId { get; set; }
      public string ApplicationId { get; set; }
   }
}

The object is passed as the context parameter of the SetReceiveMessageHandlerAsync method.

try
{
	DeviceClient deviceClient = DeviceClient.CreateFromConnectionString(
		options.AzureIoTHubconnectionString,
		endDevice.Ids.Device_id,
		TransportType.Amqp_Tcp_Only);

	await deviceClient.OpenAsync();

	AzureIoTHubReceiveMessageHandlerContext context = new AzureIoTHubReceiveMessageHandlerContext()
	{
		TenantId = options.Tenant,
		DeviceId = endDevice.Ids.Device_id,
		ApplicationId = options.ApiApplicationID,
	};

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

To send a message to a LoRaWAN device in addition to the payload, TTN needs the port number and optionally a confirmation required flag, message priority, queueing type and correlation ids.

With my implementation the confirmation required flag, message priority, and queueing type are Azure IoT Hub message properties and the messageid is used as a correlation id.

private async static Task AzureIoTHubClientReceiveMessageHandler(Message message, object userContext)
{
	bool confirmed;
	byte port;
	DownlinkPriority priority;
	string downlinktopic;

	try
	{
		AzureIoTHubReceiveMessageHandlerContext receiveMessageHandlerConext = (AzureIoTHubReceiveMessageHandlerContext)userContext;

		DeviceClient deviceClient = (DeviceClient)DeviceClients.Get(receiveMessageHandlerConext.DeviceId);
		if (deviceClient == null)
		{
			Console.WriteLine($" UplinkMessageReceived unknown DeviceID: {receiveMessageHandlerConext.DeviceId}");
			await deviceClient.RejectAsync(message);
			return;
		}

		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
			if (!message.Properties.ContainsKey("Confirmed"))
			{
				Console.WriteLine(" UplinkMessageReceived missing confirmed property");
				await deviceClient.RejectAsync(message);
				return;
			}

			if (!bool.TryParse(message.Properties["Confirmed"], out confirmed))
			{
				Console.WriteLine(" UplinkMessageReceived confirmed property invalid");
				await deviceClient.RejectAsync(message);
				return;
			}

			if (!message.Properties.ContainsKey("Priority"))
			{
				Console.WriteLine(" UplinkMessageReceived missing priority property");
				await deviceClient.RejectAsync(message);
				return;
			}

			if (!Enum.TryParse(message.Properties["Priority"], true, out priority))
			{
				Console.WriteLine(" UplinkMessageReceived priority property invalid");
				await deviceClient.RejectAsync(message);
				return;
			}

			if (priority == DownlinkPriority.Undefined)
			{
				Console.WriteLine(" UplinkMessageReceived priority property undefined value invalid");
				await deviceClient.RejectAsync(message);
				return;
			}

			if (!message.Properties.ContainsKey("Port"))
			{
				Console.WriteLine(" UplinkMessageReceived missing port number property");
				await deviceClient.RejectAsync(message);
				return;
			}

			if (!byte.TryParse( message.Properties["Port"], out port))
			{
				Console.WriteLine(" UplinkMessageReceived port number property invalid");
				await deviceClient.RejectAsync(message);
				return;
			}

			if ((port < Constants.PortNumberMinimum) || port > (Constants.PortNumberMaximum))
			{
				Console.WriteLine($" UplinkMessageReceived port number property invalid value must be between {Constants.PortNumberMinimum} and {Constants.PortNumberMaximum}");
				await deviceClient.RejectAsync(message);
				return;
			}

			if (!message.Properties.ContainsKey("Queue"))
			{
				Console.WriteLine(" UplinkMessageReceived missing queue property");
				await deviceClient.RejectAsync(message);
				return;
			}

			switch(message.Properties["Queue"].ToLower())
			{
				case "push":
					downlinktopic = $"v3/{receiveMessageHandlerConext.ApplicationId}@{receiveMessageHandlerConext.TenantId}/devices/{receiveMessageHandlerConext.DeviceId}/down/push";
					break;
				case "replace":
					downlinktopic = $"v3/{receiveMessageHandlerConext.ApplicationId}@{receiveMessageHandlerConext.TenantId}/devices/{receiveMessageHandlerConext.DeviceId}/down/replace";
					break;
				default:
					Console.WriteLine(" UplinkMessageReceived missing queue property invalid value");
					await deviceClient.RejectAsync(message);
					return;
               }

			DownlinkPayload Payload = new DownlinkPayload()
			{
				Downlinks = new List<Downlink>()
				{ 
					new Downlink()
					{
						Confirmed = confirmed,
						PayloadRaw = messageBody,
						Priority = priority,
						Port = port,
						CorrelationIds = new List<string>()
						{
							message.MessageId
						}
					}
				}
			};

			var mqttMessage = new MqttApplicationMessageBuilder()
					.WithTopic(downlinktopic)
					.WithPayload(JsonConvert.SerializeObject(Payload))
					.WithAtLeastOnceQoS()
					.Build();

			await mqttClient.PublishAsync(mqttMessage);

			// Need to look at confirmation requirement ack, nack maybe failed & sent
			await deviceClient.CompleteAsync(message);

			Console.WriteLine();
		}
	}
	catch (Exception ex)
	{
		Debug.WriteLine("UplinkMessageReceived failed: {0}", ex.Message);
	}
}

To “smoke test”” my implementation I used Azure IoT Explorer to send a C2D telemetry message

Azure IoT Hub Explorer send message form with payload and message properties

The PoC console application then forwarded the message to TTN using MQTT to be sent(which fails)

PoC application sending message then displaying result

The TTN live data display shows the message couldn’t be delivered because my test LoRaWAN device has not been activiated.

TTN Live Data display with message delivery failure

Now that my PoC application can receive and transmit message to devices I need to reconfigure my RAK Wisgate Developer D+ gateway and Seeeduino LoRaWAN and RAK Wisnode 7200 Track Lite devices on The Things Industries Network so I can test my approach with more realistic setup.

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.