-
Notifications
You must be signed in to change notification settings - Fork 456
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
fixed subscription restore from client state when mqtt client reconne…
…cts (#5048) Cherry picked (#4189) When an MQTT client used persistent session, the previous subscriptions did not get restored. If the same client connected to IoTHub directly, its subscriptions (e.g. for receiving twin responses) would have been restored. In case of c# sdk it would not be a problem, because the sdk automatically resubscribes when it is able to restore the connection to edgeHub. In case of python sdk however this does not happen. As a result, if a device connects using the python sdk and edgeHub gets restarted, the client will not be receiving m2m messages, twin patches, etc anymore. The fix seems big but the main code in "MessagingServiceClient.cs", the rest is just the result of a changed interface and some plumbing.
- Loading branch information
Showing
12 changed files
with
360 additions
and
128 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
17 changes: 17 additions & 0 deletions
17
edge-hub/src/Microsoft.Azure.Devices.Edge.Hub.Mqtt/AuthenticatedIdentity.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,17 @@ | ||
| // Copyright (c) Microsoft. All rights reserved. | ||
| namespace Microsoft.Azure.Devices.Edge.Hub.Mqtt | ||
| { | ||
| using Microsoft.Azure.Devices.Edge.Util; | ||
| using Microsoft.Azure.Devices.ProtocolGateway.Identity; | ||
|
|
||
| public class AuthenticatedIdentity : IDeviceIdentity | ||
| { | ||
| public AuthenticatedIdentity(string id) | ||
| { | ||
| this.Id = Preconditions.CheckNotNull(id, nameof(id)); | ||
| } | ||
|
|
||
| public bool IsAuthenticated => true; | ||
| public string Id { get; } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
83 changes: 83 additions & 0 deletions
83
edge-hub/src/Microsoft.Azure.Devices.Edge.Hub.Mqtt/SessionStateParser.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,83 @@ | ||
| // Copyright (c) Microsoft. All rights reserved. | ||
| namespace Microsoft.Azure.Devices.Edge.Hub.Mqtt | ||
| { | ||
| using System.Collections.Generic; | ||
| using System.Linq; | ||
| using System.Text.RegularExpressions; | ||
| using Microsoft.Azure.Devices.Edge.Hub.Core.Device; | ||
| using Microsoft.Azure.Devices.Edge.Util; | ||
| using Microsoft.Extensions.Logging; | ||
|
|
||
| using static System.FormattableString; | ||
|
|
||
| public class SessionStateParser | ||
| { | ||
| internal const string C2DSubscriptionTopicPrefix = @"messages/devicebound/#"; | ||
| internal const string MethodSubscriptionTopicPrefix = @"$iothub/methods/POST/"; | ||
| internal const string TwinSubscriptionTopicPrefix = @"$iothub/twin/PATCH/properties/desired/"; | ||
| internal const string TwinResponseTopicFilter = "$iothub/twin/res/#"; | ||
| internal static readonly Regex ModuleMessageTopicRegex = new Regex("^devices/.+/modules/.+/#$"); | ||
|
|
||
| public static IEnumerable<(DeviceSubscription, bool)> GetDeviceSubscriptions(IReadOnlyDictionary<string, bool> topics, string id) | ||
| { | ||
| return topics.Select( | ||
| subscriptionRegistration => | ||
| { | ||
| string topicName = subscriptionRegistration.Key; | ||
| bool addSubscription = subscriptionRegistration.Value; | ||
| DeviceSubscription deviceSubscription = GetDeviceSubscription(topicName); | ||
| if (deviceSubscription == DeviceSubscription.Unknown) | ||
| { | ||
| Events.UnknownTopicSubscription(topicName, id); | ||
| } | ||
| return (deviceSubscription, addSubscription); | ||
| }); | ||
| } | ||
|
|
||
| public static DeviceSubscription GetDeviceSubscription(string topicName) | ||
| { | ||
| Preconditions.CheckNonWhiteSpace(topicName, nameof(topicName)); | ||
| if (topicName.StartsWith(MethodSubscriptionTopicPrefix)) | ||
| { | ||
| return DeviceSubscription.Methods; | ||
| } | ||
| else if (topicName.StartsWith(TwinSubscriptionTopicPrefix)) | ||
| { | ||
| return DeviceSubscription.DesiredPropertyUpdates; | ||
| } | ||
| else if (topicName.EndsWith(C2DSubscriptionTopicPrefix)) | ||
| { | ||
| return DeviceSubscription.C2D; | ||
| } | ||
| else if (topicName.Equals(TwinResponseTopicFilter)) | ||
| { | ||
| return DeviceSubscription.TwinResponse; | ||
| } | ||
| else if (ModuleMessageTopicRegex.IsMatch(topicName)) | ||
| { | ||
| return DeviceSubscription.ModuleMessages; | ||
| } | ||
| else | ||
| { | ||
| return DeviceSubscription.Unknown; | ||
| } | ||
| } | ||
|
|
||
| static class Events | ||
| { | ||
| const int IdStart = MqttEventIds.SessionStateParser; | ||
| static readonly ILogger Log = Logger.Factory.CreateLogger<SessionStateParser>(); | ||
|
|
||
| enum EventIds | ||
| { | ||
| UnknownSubscription = IdStart, | ||
| } | ||
|
|
||
| public static void UnknownTopicSubscription(string topicName, string id) | ||
| { | ||
| Log.LogInformation((int)EventIds.UnknownSubscription, Invariant($"Ignoring unknown subscription to topic {topicName} for client {id}.")); | ||
| } | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.