Random wanderings through Microsoft Azure esp. PaaS plumbing, the IoT bits, AI on Micro controllers, AI on Edge Devices, .NET nanoFramework, .NET Core on *nix and ML.NET+ONNX
Azure IoT Central with consecutive duplicate PacketIds
Then I started to pay more attention and noticed that duplicate PacketIds could be interleaved
Azure IoT Central with interleaved duplicate PacketIds
Shortly after noticing the interleaved PacketIds I checked the Delivery Method and found there were message delivery timeouts.
Swarm Space Delivery with method timeouts
In Azure Application Insights I could see that the UplinkController was taking up to 15 seconds to execute which was longer than the bumblebee hive delivery timeout.
In Telerik Fiddler I could see calls to the UplinkController taking 16 seconds to execute. (I did see 30+ seconds)
Telerik Fiddler showing duration of Uplink controller calls
To see if the problem was loading CS-Script I added code to load a simple function as the application started. After averaging the duration over many executions there was little difference in the duration.
public interface IApplication
{
public DateTime Startup(DateTime utcNow);
}
...
protected override async Task ExecuteAsync(CancellationToken cancellationToken)
{
await Task.Yield();
_logger.LogInformation("StartUpService.ExecuteAsync start");
// Force the loading and startup of CS Script evaluator
dynamic application = CSScript.Evaluator
.LoadCode(
@"using System;
public class Application : IApplication
{
public DateTime Startup(DateTime utcNow)
{
return utcNow;
}
}");
DateTime result = application.Startup(DateTime.UtcNow);
try
{
await _swarmSpaceBumblebeeHive.Login(cancellationToken);
await _azureIoTDeviceClientCache.Load(cancellationToken);
}
catch (Exception ex)
{
_logger.LogError(ex, "StartUpService.ExecuteAsync error");
throw;
}
_logger.LogInformation("StartUpService.ExecuteAsync finish");
}
using System;
using System.Globalization;
using System.Text;
using Microsoft.Azure.Devices.Client;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
public class FormatterUplink : PayloadFormatter.IFormatterUplink
{
public Message Evaluate(int organisationId, int deviceId, int deviceType, int userApplicationId, JObject telemetryEvent, JObject payloadJson, string payloadText, byte[] payloadBytes)
{
if ((payloadText != "") && (payloadJson != null))
{
JObject location = new JObject();
location.Add("lat", payloadJson.GetValue("lt"));
location.Add("lon", payloadJson.GetValue("ln"));
location.Add("alt", payloadJson.GetValue("a"));
telemetryEvent.Add("DeviceLocation", location);
}
Message ioTHubmessage = new Message(Encoding.ASCII.GetBytes(JsonConvert.SerializeObject(telemetryEvent)));
ioTHubmessage.Properties.Add("iothub-creation-time-utc", DateTimeOffset.FromUnixTimeSeconds((long)payloadJson.GetValue("d")).ToString("s", CultureInfo.InvariantCulture));
return ioTHubmessage;
}
}
I then added code to load the most complex uplink and downlink formatters as the application started. There was a significant reduction in the UplinkController execution durations, but it could still take more than 30 seconds.
I then added detailed telemetry to the code and found that the duration (also variability) was a combination of Azure IoT Device Provisoning Service(DPS) registration, Azure IoT Hub connection establishment, CS-Script payload formatter loading/compilation/execution, application startup tasks and message uploading durations.
After much experimentation It looks like that “synchronously” calling the payload processing code from the Uplink controller is not a viable approach as the Swarm Space Bumblebee hive calls regularly timeout resulting in duplicate messages.
public interface IFormatterUplink
{
public Message Evaluate(int organisationId, int deviceId, int deviceType, int userApplicationId, JObject telemetryEvent, JObject payloadJson, string payloadText, byte[] payloadBytes);
}
public class FormatterUplink : PayloadFormatter.IFormatterUplink
{
public Message Evaluate(int organisationId, int deviceId, int deviceType, int userApplicationId, JObject telemetryEvent, JObject payloadJson, string payloadText, byte[] payloadBytes)
{
Message ioTHubmessage = new Message(Encoding.ASCII.GetBytes(JsonConvert.SerializeObject(telemetryEvent)));
return ioTHubmessage;
}
}
const string codeSwarmSpaceUplinkFormatterCode = @"
using Newtonsoft.Json.Linq;
public class UplinkFormatter : PayloadFormatter.ISwarmSpaceFormatterUplink
{
public JObject Evaluate(JObject telemetryEvent, string payloadBase64, byte[] payloadBytes, string payloadText, JObject payloadJson)
{
if ((payloadText != """" ) && ( payloadJson != null))
{
JObject location = new JObject() ;
location.Add(""Lat"", payloadJson.GetValue(""lt""));
location.Add(""Lon"", payloadJson.GetValue(""ln""));
location.Add(""Alt"", payloadJson.GetValue(""a""));
telemetryEvent.Add( ""location"", location);
};
return telemetryEvent;
}
}";
}
The PayloadFormatter namespace was added to reduce the length of the payload formatter C# interface declarations.
namespace PayloadFormatter
{
using Newtonsoft.Json.Linq;
public interface ISwarmSpaceFormatterUplink
{
public JObject Evaluate(JObject telemetry, string payloadBase64, byte[] payloadBytes, string payloadText, JObject payloadJson);
}
public interface ISwarmSpaceFormatterDownlink
{
public string Evaluate(JObject payloadJson, string payloadText, byte[] payloadBytes, string payloadBase64);
}
}
namespace devMobile.IoT.SwarmSpace.AzureIoT.Connector
{
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using CSScriptLib;
using PayloadFormatter;
public interface ISwarmSpaceFormatterCache
{
public Task<ISwarmSpaceFormatterUplink> PayloadFormatterGetOrAddAsync(int userApplicationId);
}
public class SwarmSpaceFormatterCache : ISwarmSpaceFormatterCache
{
private readonly ILogger<SwarmSpaceFormatterCache> _logger;
public SwarmSpaceFormatterCache(ILogger<SwarmSpaceFormatterCache>logger)
{
_logger = logger;
}
public async Task<ISwarmSpaceFormatterUplink> PayloadFormatterGetOrAddAsync(int deviceId)
{
return CSScript.Evaluator.LoadCode<PayloadFormatter.ISwarmSpaceFormatterUplink>(codeSwarmSpaceUplinkFormatterCode);
}
...
}
The parameters of the formatter are Base64 encoded, textual and a Newtonsoft JObject representations of the uplink payload and a telemetry event populated with some uplink message metadata.
Azure IoT Central uplink telemetry message payload
The initial “compile” of an uplink formatter was taking approximately 2.1 seconds so they will be “compiled” on demand and cached in a Dictionary with the UserApplicationId as the key. A default uplink formatter will be used when a UserApplicationId specific uplink formatter is not configured.
My Azure IoT Central client application displays the generated message including the decoded payload field which is used by the built in Low Power Protocol(LPP) decoder/encoder and other custom encoders/decoders.
Azure IoT Central commands for TTN/TTI integration
From the “Device Commands” form I can send commands and a queued commands which have float parameters or object parameters which contain one or more float values in a JSON payload.
For commands which call the methodHander which was been registered by calling SetMethodDefaultHandlerAsync the request payload can be JSON or plain text. If the payload is valid JSON it is “grafted”(couldn’t think of a better word) into the decoded_payload field. If the payload is not valid a JSON object with the method name as the “name” and the text payload as the value is added the decoded_payload.
private static async Task<MethodResponse> MethodCallbackDefaultHandler(MethodRequest methodRequest, object userContext)
{
AzureIoTMethodHandlerContext receiveMessageHandlerConext = (AzureIoTMethodHandlerContext)userContext;
Console.WriteLine($"Default handler method {methodRequest.Name} was called.");
Console.WriteLine($"Payload:{methodRequest.DataAsJson}");
Console.WriteLine();
if (string.IsNullOrWhiteSpace(methodRequest.Name))
{
Console.WriteLine($" Method Request Name null or white space");
return new MethodResponse(400);
}
string payloadText = Encoding.UTF8.GetString(methodRequest.Data);
if (string.IsNullOrWhiteSpace(payloadText))
{
Console.WriteLine($" Payload null or white space");
return new MethodResponse(400);
}
// At this point would check to see if Azure DeviceClient is in cache, this is so nasty
if ( String.Compare( methodRequest.Name, "Analog_Output_1", true) ==0 )
{
Console.WriteLine($" Device not found");
return new MethodResponse(UTF8Encoding.UTF8.GetBytes("Device not found"), 404);
}
JObject payload;
if (IsValidJSON(payloadText))
{
payload = JObject.Parse(payloadText);
}
else
{
payload = new JObject
{
{ methodRequest.Name, payloadText }
};
}
string downlinktopic = $"v3/{receiveMessageHandlerConext.ApplicationId}@{receiveMessageHandlerConext.TenantId}/devices/{receiveMessageHandlerConext.DeviceId}/down/push";
DownlinkPayload downlinkPayload = new DownlinkPayload()
{
Downlinks = new List<Downlink>()
{
new Downlink()
{
Confirmed = false,
//PayloadRaw = messageBody,
PayloadDecoded = payload,
Priority = DownlinkPriority.Normal,
Port = 10,
/*
CorrelationIds = new List<string>()
{
methodRequest.LockToken
}
*/
}
}
};
Console.WriteLine($"TTN Topic :{downlinktopic}");
Console.WriteLine($"TTN downlink JSON :{JsonConvert.SerializeObject(downlinkPayload, Formatting.Indented)}");
return new MethodResponse(200);
}
Configuration of unqueued Commands with a typed payloadThe output of my test harness for a Command for a typed payloadConfiguring fields of object payload(JSON)
A JSON request payload also supports downlink messages with more that one value.
The output of my test harness for a Command with an object payload(JSON)
For queued commands which call the ReceiveMessageHandler which has was registered by calling SetReceiveMessageHandler the request payload is JSON or plain text.
When I initiated an Analog queued command the message handler was invoked with the name of the command capability (Analog_Output_2) in a message property called “method-name”. For a typed parameter the message content was a string representation of the value. For an object parameter the payload contains a JSON representation of the request field(s)
The output of my test harness for a Queued Command with a typed payload
A JSON request payload supports downlink message with more that one value.
The output of my test harness for a Queued Command with an object payload(JSON)
The context information for both comments and queued commands provides additional information required to construct the MQTT topic for publishing the downlink messages.
If the device is not known the Abandon method will be called immediately. For command messages Completed will be called as soon as the message is “sent”
With object based parameters the request JSON could contain more than one value though the validation of user provided information didn’t appear to be as robust.
Object parameter schema definition
I “migrated” my third preconfigured device to the CommandRequest template to see how the commands with Request parameters interacted with my PoC application.
After “migrating” my device I went back and created a Template view so I could visualise the simulated telemetry from my PoC application and provide a way to initiate commands (Didn’t really need four command tiles as they all open the Device commands form).
CommandRequest device template default view
From the Device Commands form I could send commands and a queued commands which had analog or digital parameters.
Device Three Command Tab
When I initiated an Analog non-queued command the default method handler was invoked with the name of the command capability (Analog_Output_1) as the method name and the payload contained a JSON representation of the request values(s). With a typed parameter a string representation of the value was in the message payload. With a typed parameter a string representation of the value was in the message payload rather than JSON.
Console application displaying Analog request and Analog Request queued commands
When I initiated an Analog queued command the message handler was invoked with the name of the command capability (Analog_Output_2) in a message property called “method-name” and the payload contained a JSON representation of the request value(s). With a typed parameter a string representation of the value was in the message payload rather than JSON.
When I initiated a Digital non-queued command the default method handler was invoked with the name of the command capability (Digital_Output_1) as the method name and the payload contained a JSON representation of the request values(s). With a typed parameter a string representation of the value was in the message payload rather than JSON.
Console application displaying Digital request and Digital Request queued commands
When I initiated a Digital queued command the message handler was invoked with the name of the command capability(Digital_Output_2) in a message property called “method-name” and the payload contained a JSON representation of the request value(s). With a typed parameter a string representation of the value was in the message payload rather than JSON.
The validation of user input wasn’t as robust as I expected, with problems selecting checkboxes with a mouse when there were several Boolean fields. I often had to click on a nearby input field and use the TAB button to navigate to the desired checkbox. I also had problems with ISO 8601 format date validation as the built in Date Picker returned a month, day, year date which was not editable and wouldn’t pass validation.
The next logical step would be to look at commands with a Response parameter but as the MQTT interface is The Things Network(TTN) and The Things Industries(TTI) is asynchronous and devices reporting every 5 minutes to a couple of times a day there could be a significant delay between sending a message and receiving an optional delivery confirmation or response.
I have been struggling with making The Things Network(TTN) and The Things Industries(TTI) uplink/downlink messages work well Azure IoT Central. To explore different messaging approaches I have built a proof of Concept(PoC) application which simulates TTN/TTI connectivity to an Azure IoT Hub, or Azure IoT Central.
This blog post is about queued and non queued Cloud to Device(C2D) commands without request or response parameters. I have mostly used non queued commands in other projects (my Azure IoT HubLoRa and RF24L01 gateways) to “Restart” devices etc..
From the Device Commands tab I can could non queued and a queued commands
Device Two Commands tab
When I sent a non-queued command the default method handler was invoked with the name of the command capability (Digital_Output_0) as the method name and an empty payload. In the Azure IoT Central interface I couldn’t see any difference for successful (HTTP 200 OK) or failure (HTTP 400 Bad Request or HTTP 404 Not Found) responses. If the application was not running the command failed immediately.
When I sent a queued command the message handler was invoked with the name of the command capability(Digital_Output_1) in a message property called “method-name” and the payload contained only an “@” character.
Console application displaying queued call
If the application was not running the command was queued until the Console application was started. When the console application was running and AbandonAsync was called rather than CompleteAsync the message was retried 10 times. If RejectAsync was called rather than CompleteAsync the message was deleted from the queue and not retried. There didn’t appear to be any difference for the displayed Azure IoT Central or Azure IoT Hub explorer results when AbandonAsync or RejectAsync were called.
I also created a personal dashboard to visualise the telemetry data and initiate commands. The way the two commands were presented on the dashboard was quite limited so I will go back to the documentation and see what I missed
I have been struggling with making The Things Network(TTN) and The Things Industries(TTI) uplink/downlink messages Azure IoT Central compatible. To explore the messaging approaches used I have built a proof of Concept(PoC) application which simulates TTN/TTI connectivity to an Azure IoT Hub, or Azure IoT Central.
I then “migrated” the first device to my BasicTelemetry template
Migrating a device to TelemetryBasic template
I then went back and created a Template view to visualise the telemetry from my console application.
TelemetryBasic device template default view
Then I configured a preview device so the template view was populated with “realistic” data.
TelemetryBasic device template default view configuring a device as data source
The console application simulates a digital input (random true/false), analog input (random value between 0.0 and 1.0) and a Global Positioning System(GPS) location (Christchurch Anglican Cathedral with a random latitude, longitude and altitude offset) .
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; }
}
}
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
// 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);
//string queuedTopic = $"v3/{options.MqttApplicationID}/devices/+/queued";
//await mqttClient.SubscribeAsync(queuedTopic, MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce);
The additional commented out subscriptions are for the processing of downlink messages
The MQTTNet received message handler uses the last segment of the topic to route messages to a method for processing
The UplinkMessageReceived method deserialises the message payload, retrieves device context information from the local ObjectCache, adds relevant uplink messages fields (including the raw payload), then if the message has been unpacked by a TTN Decoder, the message fields are added as well.
static async Task UplinkMessageReceived(MqttApplicationMessageReceivedEventArgs e)
{
try
{
PayloadUplinkV3 payload = JsonConvert.DeserializeObject<PayloadUplinkV3>(e.ApplicationMessage.ConvertPayloadToString());
string applicationId = payload.EndDeviceIds.ApplicationIds.ApplicationId;
string deviceId = payload.EndDeviceIds.DeviceId;
int port = payload.UplinkMessage.Port;
...
DeviceClient deviceClient = (DeviceClient)DeviceClients.Get(deviceId);
if (deviceClient == null)
{
Console.WriteLine($" UplinkMessageReceived unknown DeviceID: {deviceId}");
return;
}
JObject telemetryEvent = new JObject();
telemetryEvent.Add("DeviceID", deviceId);
telemetryEvent.Add("ApplicationID", applicationId);
telemetryEvent.Add("Port", port);
telemetryEvent.Add("PayloadRaw", payload.UplinkMessage.PayloadRaw);
// If the payload has been unpacked in TTN backend add fields to telemetry event payload
if (payload.UplinkMessage.PayloadDecoded != null)
{
EnumerateChildren(telemetryEvent, payload.UplinkMessage.PayloadDecoded);
}
// Send the message to Azure IoT Hub/Azure IoT Central
using (Message ioTHubmessage = new Message(Encoding.ASCII.GetBytes(JsonConvert.SerializeObject(telemetryEvent))))
{
// Ensure the displayed time is the acquired time rather than the uploaded time.
//ioTHubmessage.Properties.Add("iothub-creation-time-utc", payloadObject.Metadata.ReceivedAtUtc.ToString("s", CultureInfo.InvariantCulture));
ioTHubmessage.Properties.Add("ApplicationId", applicationId);
ioTHubmessage.Properties.Add("DeviceId", deviceId);
ioTHubmessage.Properties.Add("port", port.ToString());
await deviceClient.SendEventAsync(ioTHubmessage);
}
}
catch( Exception ex)
{
Debug.WriteLine("UplinkMessageReceived failed: {0}", ex.Message);
}
}
private static void EnumerateChildren(JObject jobject, JToken token)
{
if (token is JProperty property)
{
if (token.First is JValue)
{
// Temporary dirty hack for Azure IoT Central compatibility
if (token.Parent is JObject possibleGpsProperty)
{
if (possibleGpsProperty.Path.StartsWith("GPS_", StringComparison.OrdinalIgnoreCase))
{
if (string.Compare(property.Name, "Latitude", true) == 0)
{
jobject.Add("lat", property.Value);
}
if (string.Compare(property.Name, "Longitude", true) == 0)
{
jobject.Add("lon", property.Value);
}
if (string.Compare(property.Name, "Altitude", true) == 0)
{
jobject.Add("alt", property.Value);
}
}
}
jobject.Add(property.Name, property.Value);
}
else
{
JObject parentObject = new JObject();
foreach (JToken token2 in token.Children())
{
EnumerateChildren(parentObject, token2);
jobject.Add(property.Name, parentObject);
}
}
}
else
{
foreach (JToken token2 in token.Children())
{
EnumerateChildren(jobject, token2);
}
}
}
There is also some basic reformatting of the messages for Azure IoT Central
TTN Simulate uplink message with GPS location payload.Nasty console application processing uplink messageMessage from LoRaWAN device displayed in Azure IoT Explorer
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 commandlineParserNuGet 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 DeviceClientSetReceiveMessageHandlerAsync 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.
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.