Skip to content

Commit

Permalink
feature,fix (device-client) Handle Twin failures using Amqp (#1796)
Browse files Browse the repository at this point in the history
  • Loading branch information
@bikamani
bikamani committed Feb 27, 2021
1 parent c95dc1f commit 5fcd166
Show file tree
Hide file tree
Showing 3 changed files with 79 additions and 60 deletions.
6 changes: 5 additions & 1 deletion e2e/test/iothub/twin/TwinE2ETests.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,6 @@
using Microsoft.Azure.Devices.Shared;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;

namespace Microsoft.Azure.Devices.E2ETests.Twins
{
Expand Down Expand Up @@ -367,6 +366,10 @@ public async Task Twin_ClientHandlesRejectionInvalidPropertyName_MqttWs()
.ConfigureAwait(false);
}

// These tests worked earlier for Amqp and AmqpWs since it was catching a wrong exception
// To Do: Fix Update reported properties method behavior (breaking change) to wait for response
// and we should be able to enable these tests then.
[Ignore]
[LoggedTestMethod]
public async Task Twin_ClientHandlesRejectionInvalidPropertyName_Amqp()
{
Expand All @@ -375,6 +378,7 @@ public async Task Twin_ClientHandlesRejectionInvalidPropertyName_Amqp()
.ConfigureAwait(false);
}

[Ignore]
[LoggedTestMethod]
public async Task Twin_ClientHandlesRejectionInvalidPropertyName_AmqpWs()
{
Expand Down
46 changes: 26 additions & 20 deletions iothub/device/src/Transport/Amqp/AmqpTransportHandler.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -324,7 +324,8 @@ public override async Task EnableTwinPatchAsync(CancellationToken cancellationTo
try
{
cancellationToken.ThrowIfCancellationRequested();
await _amqpUnit.SendTwinMessageAsync(AmqpTwinMessageType.Put, Guid.NewGuid().ToString(), null, _operationTimeout).ConfigureAwait(false);
string correlationId = AmqpTwinMessageType.Put + Guid.NewGuid().ToString();
await _amqpUnit.SendTwinMessageAsync(AmqpTwinMessageType.Put, correlationId, null, _operationTimeout).ConfigureAwait(false);
}
finally
{
Expand Down Expand Up @@ -355,7 +356,7 @@ public override async Task<Twin> SendTwinGetAsync(CancellationToken cancellation
try
{
await EnableTwinPatchAsync(cancellationToken).ConfigureAwait(false);
Twin twin = await RoundTripTwinMessage(AmqpTwinMessageType.Get, null, cancellationToken).ConfigureAwait(false);
Twin twin = await RoundTripTwinMessageAsync(AmqpTwinMessageType.Get, null, cancellationToken).ConfigureAwait(false);
if (twin == null)
{
throw new InvalidOperationException("Service rejected the message");
Expand All @@ -375,19 +376,19 @@ public override async Task SendTwinPatchAsync(TwinCollection reportedProperties,
try
{
await EnableTwinPatchAsync(cancellationToken).ConfigureAwait(false);
Twin twin = await RoundTripTwinMessage(AmqpTwinMessageType.Patch, reportedProperties, cancellationToken).ConfigureAwait(false);
await RoundTripTwinMessageAsync(AmqpTwinMessageType.Patch, reportedProperties, cancellationToken).ConfigureAwait(false);
}
finally
{
Logging.Exit(this, reportedProperties, cancellationToken, $"{nameof(SendTwinPatchAsync)}");
}
}

private async Task<Twin> RoundTripTwinMessage(AmqpTwinMessageType amqpTwinMessageType, TwinCollection reportedProperties, CancellationToken cancellationToken)
private async Task<Twin> RoundTripTwinMessageAsync(AmqpTwinMessageType amqpTwinMessageType, TwinCollection reportedProperties, CancellationToken cancellationToken)
{
Logging.Enter(this, cancellationToken, $"{nameof(RoundTripTwinMessage)}");
Logging.Enter(this, cancellationToken, $"{nameof(RoundTripTwinMessageAsync)}");

string correlationId = Guid.NewGuid().ToString();
string correlationId = amqpTwinMessageType + Guid.NewGuid().ToString();
Twin response = null;

try
Expand All @@ -405,10 +406,6 @@ private async Task<Twin> RoundTripTwinMessage(AmqpTwinMessageType amqpTwinMessag
// Consider that the task may have faulted or been canceled.
// We re-await the task so that any exceptions/cancellation is rethrown.
response = await receivingTask.ConfigureAwait(false);
if (response == null)
{
throw new InvalidOperationException("Service response is null");
}
}
else
{
Expand All @@ -419,7 +416,7 @@ private async Task<Twin> RoundTripTwinMessage(AmqpTwinMessageType amqpTwinMessag
finally
{
_twinResponseCompletions.TryRemove(correlationId, out _);
Logging.Exit(this, cancellationToken, $"{nameof(RoundTripTwinMessage)}");
Logging.Exit(this, cancellationToken, $"{nameof(RoundTripTwinMessageAsync)}");
}

return response;
Expand Down Expand Up @@ -522,19 +519,28 @@ private async Task DisposeMessageAsync(string lockToken, AmqpIoTDisposeActions o

private void TwinMessageListener(Twin twin, string correlationId, TwinCollection twinCollection)
{
if (correlationId != null)
if (correlationId == null)
{
// It is a GET, just complete the task.
TaskCompletionSource<Twin> task;
if (_twinResponseCompletions.TryRemove(correlationId, out task))
{
task.SetResult(twin);
}
// This is desired property updates, so call the callback with TwinCollection.
_onDesiredStatePatchListener(twinCollection);
}
else
{
// It is a PATCH, just call the callback with the TwinCollection
_onDesiredStatePatchListener(twinCollection);
if (correlationId.StartsWith(AmqpTwinMessageType.Get.ToString(), StringComparison.OrdinalIgnoreCase) ||
correlationId.StartsWith(AmqpTwinMessageType.Patch.ToString(), StringComparison.OrdinalIgnoreCase))
{
// For Get and Patch, complete the task.
TaskCompletionSource<Twin> task;
if (_twinResponseCompletions.TryRemove(correlationId, out task))
{
task.SetResult(twin);
}
else
{
// This can happen if we received a message from service with correlation Id that was not set by SDK or does not exist in dictionary.
Logging.Info("Could not remove correlation id to complete the task awaiter for a twin operation.", nameof(TwinMessageListener));
}
}
}
}

Expand Down
87 changes: 48 additions & 39 deletions iothub/device/src/Transport/AmqpIoT/AmqpIoTReceivingLink.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ internal class AmqpIoTReceivingLink
private Action<Message> _onEventsReceived;
private Action<Message> _onDeviceMessageReceived;
private Action<MethodRequestInternal> _onMethodReceived;
private Action<Twin, string, TwinCollection> _onDesiredPropertyReceived;
private Action<Twin, string, TwinCollection> _onTwinMessageReceived;

public AmqpIoTReceivingLink(ReceivingAmqpLink receivingAmqpLink)
{
Expand Down Expand Up @@ -257,89 +257,98 @@ private void DisposeDelivery(AmqpMessage amqpMessage, bool settled, Accepted acc

internal void RegisterTwinListener(Action<Twin, string, TwinCollection> onDesiredPropertyReceived)
{
_onDesiredPropertyReceived = onDesiredPropertyReceived;
_receivingAmqpLink.RegisterMessageListener(OnDesiredPropertyReceived);
_onTwinMessageReceived = onDesiredPropertyReceived;
_receivingAmqpLink.RegisterMessageListener(OnTwinChangesReceived);
}

private void OnDesiredPropertyReceived(AmqpMessage amqpMessage)
private void OnTwinChangesReceived(AmqpMessage amqpMessage)
{
if (Logging.IsEnabled)
{
Logging.Enter(this, amqpMessage, $"{nameof(OnDesiredPropertyReceived)}");
Logging.Enter(this, amqpMessage, $"{nameof(OnTwinChangesReceived)}");
}

try
{
_receivingAmqpLink.DisposeDelivery(amqpMessage, true, AmqpIoTConstants.AcceptedOutcome);
string correlationId = amqpMessage.Properties?.CorrelationId?.ToString();

if (!VerifyResponseMessage(amqpMessage))
{
_onDesiredPropertyReceived.Invoke(null, correlationId, null);
}
int status = GetStatus(amqpMessage);

Twin twin = null;
TwinCollection twinProperties = null;

if (correlationId != null)
if (status >= 400)
{
if (amqpMessage.BodyStream != null)
// Handle failures
_onTwinMessageReceived.Invoke(null, correlationId, null);
if (correlationId.StartsWith(AmqpTwinMessageType.Get.ToString(), StringComparison.OrdinalIgnoreCase))
{
// This a result of a GET TWIN so return (set) the full twin
using (StreamReader reader = new StreamReader(amqpMessage.BodyStream, System.Text.Encoding.UTF8))
string error = null;
using (var reader = new StreamReader(amqpMessage.BodyStream, System.Text.Encoding.UTF8))
{
string body = reader.ReadToEnd();
var properties = JsonConvert.DeserializeObject<TwinProperties>(body);
twin = new Twin(properties);
}
}
else
{
// This is a desired property ack from the service
twin = new Twin();
error = reader.ReadToEnd();
};
// Retry for Http status code request timeout, Too many requests and server errors
throw new IotHubException(error, status >= 500 || status == 429 || status == 408);
}
}
else
{
// No correlationId, this is a PATCH sent by the sevice so return (set) the TwinCollection

using (StreamReader reader = new StreamReader(amqpMessage.BodyStream, System.Text.Encoding.UTF8))
if (correlationId == null)
{
// Here we are getting desired property update notifications and want to handle it first
using var reader = new StreamReader(amqpMessage.BodyStream, System.Text.Encoding.UTF8);
string patch = reader.ReadToEnd();
twinProperties = JsonConvert.DeserializeObject<TwinCollection>(patch);
}
else if (correlationId.StartsWith(AmqpTwinMessageType.Get.ToString(), StringComparison.OrdinalIgnoreCase))
{
// This a response of a GET TWIN so return (set) the full twin
using var reader = new StreamReader(amqpMessage.BodyStream, System.Text.Encoding.UTF8);
string body = reader.ReadToEnd();
var properties = JsonConvert.DeserializeObject<TwinProperties>(body);
twin = new Twin(properties);
}
else if (correlationId.StartsWith(AmqpTwinMessageType.Patch.ToString(), StringComparison.OrdinalIgnoreCase))
{
// This can be used to coorelate success response with updating reported properties
// However currently we do not have it as request response style implementation
Logging.Info("Updated twin reported properties successfully", nameof(OnTwinChangesReceived));
}
else if (correlationId.StartsWith(AmqpTwinMessageType.Put.ToString(), StringComparison.OrdinalIgnoreCase))
{
// This is an acknowledgement received from service for subscribing to desired property updates
Logging.Info("Subscribed for twin successfully", nameof(OnTwinChangesReceived));
}
else
{
// This shouldn't happen
Logging.Info("Received a correlation Id for Twin operation that does not match Get, Patch or Put request", nameof(OnTwinChangesReceived));
}
_onTwinMessageReceived.Invoke(twin, correlationId, twinProperties);
}
_onDesiredPropertyReceived.Invoke(twin, correlationId, twinProperties);
}
finally
{
if (Logging.IsEnabled)
{
Logging.Exit(this, amqpMessage, $"{nameof(OnDesiredPropertyReceived)}");
Logging.Exit(this, amqpMessage, $"{nameof(OnTwinChangesReceived)}");
}
}
}

#endregion Twin handling

internal static bool VerifyResponseMessage(AmqpMessage response)
internal static int GetStatus(AmqpMessage response)
{
bool retVal = true;
if (response != null)
{
if (response.MessageAnnotations.Map.TryGetValue(AmqpIoTConstants.ResponseStatusName, out int status))
{
if (status >= 400)
{
retVal = false;
}
return status;
}
}
else
{
retVal = false;
}
return retVal;
return -1;
}
}
}

0 comments on commit 5fcd166

Please sign in to comment.