The TTI V3 Connector Minimalist Cloud to Device only required a port number, and there was no way to specify whether delivery of message had to be confirmed, the way the message was queued, or the priority of message delivery. Like the port number these optional settings can be specified in message properties.
- Confirmation – True/False
- Queue – Push/Replace
- Priority – Lowest/Low/BelowNormal/Normal/AboveNormal/High/Highest
If any of these properties are incorrect DeviceClient.RejectAsync is called which deletes the message from the device queue and indicates to the server that the message could not be processed.
private async Task AzureIoTHubClientReceiveMessageHandler(Message message, object userContext)
{
try
{
Models.AzureIoTHubReceiveMessageHandlerContext receiveMessageHandlerContext = (Models.AzureIoTHubReceiveMessageHandlerContext)userContext;
if (!_DeviceClients.TryGetValue(receiveMessageHandlerContext.DeviceId, out DeviceClient deviceClient))
{
_logger.LogWarning("Downlink-DeviceID:{0} unknown", receiveMessageHandlerContext.DeviceId);
return;
}
using (message)
{
string payloadText = Encoding.UTF8.GetString(message.GetBytes()).Trim();
if (!AzureDownlinkMessage.PortTryGet(message.Properties, out byte port))
{
_logger.LogWarning("Downlink-Port property is invalid");
await deviceClient.RejectAsync(message);
return;
}
if (!AzureDownlinkMessage.ConfirmedTryGet(message.Properties, out bool confirmed))
{
_logger.LogWarning("Downlink-Confirmed flag is invalid");
await deviceClient.RejectAsync(message);
return;
}
if (!AzureDownlinkMessage.PriorityTryGet(message.Properties, out Models.DownlinkPriority priority))
{
_logger.LogWarning("Downlink-Priority value is invalid");
await deviceClient.RejectAsync(message);
return;
}
if (!AzureDownlinkMessage.QueueTryGet(message.Properties, out Models.DownlinkQueue queue))
{
_logger.LogWarning("Downlink-Queue value is invalid");
await deviceClient.RejectAsync(message.LockToken);
return;
}
Models.Downlink downlink = new Models.Downlink()
{
Confirmed = confirmed,
Priority = priority,
Port = port,
CorrelationIds = AzureLockToken.Add(message.LockToken),
};
// Split over multiple lines in an attempt to improve readability. In this scenario a valid JSON string should start/end with {/} for an object or [/] for an array
if ((payloadText.StartsWith("{") && payloadText.EndsWith("}"))
||
((payloadText.StartsWith("[") && payloadText.EndsWith("]"))))
{
try
{
downlink.PayloadDecoded = JToken.Parse(payloadText);
}
catch (JsonReaderException)
{
downlink.PayloadRaw = payloadText;
}
}
else
{
downlink.PayloadRaw = payloadText;
}
_logger.LogInformation("Downlink-IoT Hub DeviceID:{0} MessageID:{2} LockToken:{3} Port:{4} Confirmed:{5} Priority:{6} Queue:{7}",
receiveMessageHandlerContext.DeviceId,
message.MessageId,
message.LockToken,
downlink.Port,
downlink.Confirmed,
downlink.Priority,
queue);
Models.DownlinkPayload Payload = new Models.DownlinkPayload()
{
Downlinks = new List<Models.Downlink>()
{
downlink
}
};
string url = $"{receiveMessageHandlerContext.WebhookBaseURL}/{receiveMessageHandlerContext.ApplicationId}/webhooks/{receiveMessageHandlerContext.WebhookId}/devices/{receiveMessageHandlerContext.DeviceId}/down/{queue}".ToLower();
using (var client = new WebClient())
{
client.Headers.Add("Authorization", $"Bearer {receiveMessageHandlerContext.ApiKey}");
client.UploadString(new Uri(url), JsonConvert.SerializeObject(Payload));
}
_logger.LogInformation("Downlink-DeviceID:{0} LockToken:{1} success", receiveMessageHandlerContext.DeviceId, message.LockToken);
}
}
catch (Exception ex)
{
_logger.LogError(ex, "Downlink-ReceiveMessge processing failed");
}
}
A correlation identifier containing the Message LockToken is added to the downlink payload.
Unconfirmed Downlink Messages
For unconfirmed messages The TTI Connector calls the DeviceClient.CompletedAsync method (with the LockToken from the CorrelationIDs list) which deletes the message from the device queue.
[Function("Queued")]
public async Task<HttpResponseData> Queued([HttpTrigger(AuthorizationLevel.Function, "post")] HttpRequestData req, FunctionContext executionContext)
{
var logger = executionContext.GetLogger("Queued");
// Wrap all the processing in a try\catch so if anything blows up we have logged it.
try
{
string payloadText = await req.ReadAsStringAsync();
Models.DownlinkQueuedPayload payload = JsonConvert.DeserializeObject<Models.DownlinkQueuedPayload>(payloadText);
if (payload == null)
{
logger.LogInformation("Queued-Payload {0} invalid", payloadText);
return req.CreateResponse(HttpStatusCode.BadRequest);
}
string applicationId = payload.EndDeviceIds.ApplicationIds.ApplicationId;
string deviceId = payload.EndDeviceIds.DeviceId;
logger.LogInformation("Queued-ApplicationID:{0} DeviceID:{1} ", applicationId, deviceId);
if (!_DeviceClients.TryGetValue(deviceId, out DeviceClient deviceClient))
{
logger.LogInformation("Queued-Unknown device for ApplicationID:{0} DeviceID:{1}", applicationId, deviceId);
return req.CreateResponse(HttpStatusCode.Conflict);
}
// If the message is not confirmed "complete" it as soon as with network
if (!payload.DownlinkQueued.Confirmed)
{
if (!AzureLockToken.TryGet(payload.DownlinkQueued.CorrelationIds, out string lockToken))
{
logger.LogWarning("Queued-DeviceID:{0} LockToken missing from payload:{1}", payload.EndDeviceIds.DeviceId, payloadText);
return req.CreateResponse(HttpStatusCode.BadRequest);
}
try
{
await deviceClient.CompleteAsync(lockToken);
}
catch (DeviceMessageLockLostException)
{
logger.LogWarning("Queued-CompleteAsync DeviceID:{0} LockToken:{1} timeout", payload.EndDeviceIds.DeviceId, lockToken);
return req.CreateResponse(HttpStatusCode.Conflict);
}
logger.LogInformation("Queued-DeviceID:{0} LockToken:{1} success", payload.EndDeviceIds.DeviceId, lockToken);
}
}
catch (Exception ex)
{
logger.LogError(ex, "Queued message processing failed");
return req.CreateResponse(HttpStatusCode.InternalServerError);
}
return req.CreateResponse(HttpStatusCode.OK);
}
Confirmed Downlink Messages
If message delivery succeeds the deviceClient.CompleteAsync method (with the LockToken from the CorrelationIDs list) is called which removes the message from the device queue.
[Function("Ack")]
public async Task<HttpResponseData> Ack([HttpTrigger(AuthorizationLevel.Function, "post")] HttpRequestData req, FunctionContext executionContext)
{
var logger = executionContext.GetLogger("Queued");
// Wrap all the processing in a try\catch so if anything blows up we have logged it.
try
{
string payloadText = await req.ReadAsStringAsync();
Models.DownlinkAckPayload payload = JsonConvert.DeserializeObject<Models.DownlinkAckPayload>(payloadText);
if (payload == null)
{
logger.LogInformation("Ack-Payload {0} invalid", payloadText);
return req.CreateResponse(HttpStatusCode.BadRequest);
}
string applicationId = payload.EndDeviceIds.ApplicationIds.ApplicationId;
string deviceId = payload.EndDeviceIds.DeviceId;
logger.LogInformation("Ack-ApplicationID:{0} DeviceID:{1} ", applicationId, deviceId);
if (!_DeviceClients.TryGetValue(deviceId, out DeviceClient deviceClient))
{
logger.LogInformation("Ack-Unknown device for ApplicationID:{0} DeviceID:{1}", applicationId, deviceId);
return req.CreateResponse(HttpStatusCode.Conflict);
}
if (!AzureLockToken.TryGet(payload.DownlinkAck.CorrelationIds, out string lockToken))
{
logger.LogWarning("Ack-DeviceID:{0} LockToken missing from payload:{1}", payload.EndDeviceIds.DeviceId, payloadText);
return req.CreateResponse(HttpStatusCode.BadRequest);
}
try
{
await deviceClient.CompleteAsync(lockToken);
}
catch (DeviceMessageLockLostException)
{
logger.LogWarning("Ack-CompleteAsync DeviceID:{0} LockToken:{1} timeout", payload.EndDeviceIds.DeviceId, lockToken);
return req.CreateResponse(HttpStatusCode.Conflict);
}
logger.LogInformation("Ack-DeviceID:{0} LockToken:{1} success", payload.EndDeviceIds.DeviceId, lockToken);
}
catch (Exception ex)
{
logger.LogError(ex, "Ack message processing failed");
return req.CreateResponse(HttpStatusCode.InternalServerError);
}
return req.CreateResponse(HttpStatusCode.OK);
}
If message delivery fails the deviceClient.AbandonAsync method (with the LockToken from the CorrelationIDs list) is called which puts the downlink message back onto the device queue.
[Function("Failed")]
public async Task<HttpResponseData> Failed([HttpTrigger(AuthorizationLevel.Function, "post")] HttpRequestData req, FunctionContext executionContext)
{
var logger = executionContext.GetLogger("Queued");
// Wrap all the processing in a try\catch so if anything blows up we have logged it.
try
{
string payloadText = await req.ReadAsStringAsync();
Models.DownlinkFailedPayload payload = JsonConvert.DeserializeObject<Models.DownlinkFailedPayload>(payloadText);
if (payload == null)
{
logger.LogInformation("Failed-Payload {0} invalid", payloadText);
return req.CreateResponse(HttpStatusCode.BadRequest);
}
string applicationId = payload.EndDeviceIds.ApplicationIds.ApplicationId;
string deviceId = payload.EndDeviceIds.DeviceId;
logger.LogInformation("Failed-ApplicationID:{0} DeviceID:{1} ", applicationId, deviceId);
if (!_DeviceClients.TryGetValue(deviceId, out DeviceClient deviceClient))
{
logger.LogInformation("Failed-Unknown device for ApplicationID:{0} DeviceID:{1}", applicationId, deviceId);
return req.CreateResponse(HttpStatusCode.Conflict);
}
if (!AzureLockToken.TryGet(payload.DownlinkFailed.CorrelationIds, out string lockToken))
{
logger.LogWarning("Failed-DeviceID:{0} LockToken missing from payload:{1}", payload.EndDeviceIds.DeviceId, payloadText);
return req.CreateResponse(HttpStatusCode.BadRequest);
}
try
{
await deviceClient.RejectAsync(lockToken);
}
catch (DeviceMessageLockLostException)
{
logger.LogWarning("Failed-RejectAsync DeviceID:{0} LockToken:{1} timeout", payload.EndDeviceIds.DeviceId, lockToken);
return req.CreateResponse(HttpStatusCode.Conflict);
}
logger.LogInformation("Failed-DeviceID:{0} LockToken:{1} success", payload.EndDeviceIds.DeviceId, lockToken);
}
catch (Exception ex)
{
logger.LogError(ex, "Failed message processing failed");
return req.CreateResponse(HttpStatusCode.InternalServerError);
}
return req.CreateResponse(HttpStatusCode.OK);
}
If message delivery is unsuccessful the deviceClient.RejectAsync method (with the LockToken from the CorrelationIDs list) is called which deletes the message from the device queue and indicates to the server that the message could not be processed.
[Function("Nack")]
public async Task<HttpResponseData> Nack([HttpTrigger(AuthorizationLevel.Function, "post")] HttpRequestData req, FunctionContext executionContext)
{
var logger = executionContext.GetLogger("Queued");
// Wrap all the processing in a try\catch so if anything blows up we have logged it.
try
{
string payloadText = await req.ReadAsStringAsync();
Models.DownlinkNackPayload payload = JsonConvert.DeserializeObject<Models.DownlinkNackPayload>(payloadText);
if (payload == null)
{
logger.LogInformation("Nack-Payload {0} invalid", payloadText);
return req.CreateResponse(HttpStatusCode.BadRequest);
}
string applicationId = payload.EndDeviceIds.ApplicationIds.ApplicationId;
string deviceId = payload.EndDeviceIds.DeviceId;
logger.LogInformation("Nack-ApplicationID:{0} DeviceID:{1} ", applicationId, deviceId);
if (!_DeviceClients.TryGetValue(deviceId, out DeviceClient deviceClient))
{
logger.LogInformation("Nack-Unknown device for ApplicationID:{0} DeviceID:{1}", applicationId, deviceId);
return req.CreateResponse(HttpStatusCode.Conflict);
}
if (!AzureLockToken.TryGet(payload.DownlinkNack.CorrelationIds, out string lockToken))
{
logger.LogWarning("Nack-DeviceID:{0} LockToken missing from payload:{1}", payload.EndDeviceIds.DeviceId, payloadText);
return req.CreateResponse(HttpStatusCode.BadRequest);
}
try
{
await deviceClient.RejectAsync(lockToken);
}
catch (DeviceMessageLockLostException)
{
logger.LogWarning("Nack-RejectAsync DeviceID:{0} LockToken:{1} timeout", payload.EndDeviceIds.DeviceId, lockToken);
return req.CreateResponse(HttpStatusCode.Conflict);
}
logger.LogInformation("Nack-DeviceID:{0} LockToken:{1} success", payload.EndDeviceIds.DeviceId, lockToken);
}
catch (Exception ex)
{
logger.LogError(ex, "Nack message processing failed");
return req.CreateResponse(HttpStatusCode.InternalServerError);
}
return req.CreateResponse(HttpStatusCode.OK);
}
The way message Failed(Abandon), Ack(CompleteAsync) and Nack(RejectAsync) are handled needs some more testing to confirm my understanding of the sequencing of TTI confirmed message delivery.
BEWARE
The use of Confirmed messaging with devices that send uplink messages irregularly can cause weird problems if the Azure IoT hub downlink message times out.
devMobile's blog 