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)
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}");
}
}
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
devMobile's blog 
Pingback: The Things Network MQTT & Azure IoT Part2 | devMobile's blog