Skip to content

fix(iot-dev): Preserve response callback context #1847

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Arnab Nandy (arnabnandy7) wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix(iot-dev): Preserve response callback context #1847

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 27 additions & 25 deletions ...ce-client/src/main/java/com/microsoft/azure/sdk/iot/device/transport/IotHubTransport.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -322,6 +322,16 @@ public void onMessageSent(Message message, String deviceId, TransportException e
@Override
public void onMessageReceived(IotHubTransportMessage message, TransportException e)
{
CorrelationCallbackContext callbackContext = null;
if (message != null)
{
String correlationId = message.getCorrelationId();
if (!correlationId.isEmpty())
{
callbackContext = correlationCallbacks.get(correlationId);
}
}

if (message != null && e != null)
{
log.error("Exception encountered while receiving a message from service {}", message, e);
Expand All @@ -338,36 +348,28 @@ else if (message != null)

try
{
if (message != null)
if (callbackContext != null && callbackContext.getCallback() != null)
{
String correlationId = message.getCorrelationId();
if (!correlationId.isEmpty())
IotHubClientException clientException = null;
if (e != null)
{
CorrelationCallbackContext callbackContext = correlationCallbacks.get(correlationId);
if (callbackContext != null && callbackContext.getCallback() != null)
// This case indicates that the transport layer failed to construct a valid message out of
// a message delivered by the service
clientException = e.toIotHubClientException();
}
else
{
// This case indicates that the transport layer constructed a valid message out of a message
// delivered by the service, but that message may contain an unsuccessful status code in cases
// such as if an operation was rejected because it was badly formatted.
IotHubStatusCode statusCode = IotHubStatusCode.getIotHubStatusCode(Integer.parseInt(message.getStatus()));
if (!IotHubStatusCode.isSuccessful(statusCode))
{
IotHubClientException clientException = null;
if (e != null)
{
// This case indicates that the transport layer failed to construct a valid message out of
// a message delivered by the service
clientException = e.toIotHubClientException();
}
else
{
// This case indicates that the transport layer constructed a valid message out of a message
// delivered by the service, but that message may contain an unsuccessful status code in cases
// such as if an operation was rejected because it was badly formatted.
IotHubStatusCode statusCode = IotHubStatusCode.getIotHubStatusCode(Integer.parseInt(message.getStatus()));
if (!IotHubStatusCode.isSuccessful(statusCode))
{
clientException = new IotHubClientException(statusCode, "Received an unsuccessful operation error code from the service: " + statusCode);
}
}

callbackContext.getCallback().onResponseReceived(message, callbackContext.getUserContext(), clientException);
clientException = new IotHubClientException(statusCode, "Received an unsuccessful operation error code from the service: " + statusCode);
}
}

callbackContext.getCallback().onResponseReceived(message, callbackContext.getUserContext(), clientException);
}
}
catch (Exception ex)
Expand Down
48 changes: 48 additions & 0 deletions ...lient/src/test/java/com/microsoft/azure/sdk/iot/device/transport/IotHubTransportTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -225,6 +225,54 @@ public void onMessageReceivedWithMessageAndNoExceptionAddsToQueue()
assertEquals(mockedTransportMessage, receivedMessagesQueue.poll());
}

//Tests_SRS_IOTHUBTRANSPORT_34_010: [If a correlation callback exists for a received message, this function shall call onResponseReceived even if the saved correlation callback is removed while queueing the received message.]
@Test
public void onMessageReceivedCallsResponseCallbackEvenIfCallbackIsRemovedWhileAddingMessageToQueue(@Mocked final CorrelatingMessageCallback mockedCorrelationCallback)
{
//arrange
final String correlationId = "1234";
final Object callbackContext = new Object();
final Map<String, CorrelationCallbackContext> correlationCallbacks = new ConcurrentHashMap<>();
correlationCallbacks.put(correlationId, new CorrelationCallbackContext(mockedCorrelationCallback, callbackContext, System.currentTimeMillis()));

new Expectations()
{
{
mockedConfig.getDeviceId();
result = "someDeviceId";

mockedTransportMessage.getCorrelationId();
result = correlationId;

mockedTransportMessage.getStatus();
result = "200";
}
};

new MockUp<IotHubTransport>()
{
@Mock void addToReceivedMessagesQueue(IotHubTransportMessage message)
{
correlationCallbacks.remove(correlationId);
}
};

IotHubTransport transport = new IotHubTransport(mockedConfig, mockedIotHubConnectionStatusChangeCallback, false);
Deencapsulation.setField(transport, "correlationCallbacks", correlationCallbacks);

//act
transport.onMessageReceived(mockedTransportMessage, null);

//assert
new Verifications()
{
{
mockedCorrelationCallback.onResponseReceived(mockedTransportMessage, callbackContext, null);
times = 1;
}
};
}

//Tests_SRS_IOTHUBTRANSPORT_34_014: [If the provided connectionId is associated with the current connection, This function shall invoke updateStatus with status CONNECTED, change reason CONNECTION_OK and a null throwable.]
@Test
public void onConnectionEstablishedCallsUpdateStatus()
Expand Down