Uplink with decoded_payload & frm_payload
The next functionality added to my Proof of Concept(PoC) Azure IoT Hub, The Things Network(TTN) V3 Hypertext Transfer Protocol(HTTP) client API Integration, and Message Queue Telemetry Transport (MQTT) Data API Integration is sending of raw and decoded uplink messages to an Azure IoT Hub.
// 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
private static async void MqttClientApplicationMessageReceived(MqttApplicationMessageReceivedEventArgs e)
{
if (e.ApplicationMessage.Topic.EndsWith("/up", StringComparison.InvariantCultureIgnoreCase))
{
await UplinkMessageReceived(e);
}
/*
if (e.ApplicationMessage.Topic.EndsWith("/queued", StringComparison.InvariantCultureIgnoreCase))
{
await DownlinkMessageQueued(e);
}
...
*/
}
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
Currently the code has a lots of diagnostic Console.Writeline statements, doesn’t support Uplink messages, has no Advanced Message Queuing Protocol(AMQP) client connection pooling, can’t run as an Azure Webjob, and a number of other features which I plan on adding in future blog posts.
devMobile's blog 