iotmodels · ridomin · Apr 3, 2023 · Apr 3, 2023
diff --git a/samples/iothub-sample/Device.cs b/samples/iothub-sample/Device.cs
@@ -34,15 +34,15 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
         var twin = await client.GetTwinAsync(stoppingToken);
         Console.WriteLine(twin);
 
-        client.OnCommandReceived = m =>
+        client.OnCommandReceived = async m =>
         {
             Console.WriteLine(m.CommandName);
             Console.WriteLine(m.CommandPayload);
-            return new GenericCommandResponse()
+            return await Task.FromResult(new GenericCommandResponse()
             {
                 Status = 200,
                 ReponsePayload = JsonSerializer.Serialize(new { myResponse = "whatever" })
-            };
+            });
         };
 
         client.OnPropertyUpdateReceived = m =>

diff --git a/src/MQTTnet.Extensions.MultiCloud.AzureIoTClient/Dps/MqttDpsClient.cs b/src/MQTTnet.Extensions.MultiCloud.AzureIoTClient/Dps/MqttDpsClient.cs
@@ -56,7 +56,7 @@ public async Task<DpsStatus> ProvisionDeviceIdentity()
         {
             var putTopic = $"$dps/registrations/PUT/iotdps-register/?$rid={rid++}";
             var registrationId = mqttClient.Options.ClientId;
-            var bytes = new UTF8JsonSerializer().ToBytes(new { registrationId, payload = new { modelId } });
+            var bytes = new Utf8JsonSerializer().ToBytes(new { registrationId, payload = new { modelId } });
             var puback = await mqttClient.PublishBinaryAsync(putTopic, bytes);
             if (puback.ReasonCode != MqttClientPublishReasonCode.Success)
             {

diff --git a/src/MQTTnet.Extensions.MultiCloud.AzureIoTClient/HubMqttClient.cs b/src/MQTTnet.Extensions.MultiCloud.AzureIoTClient/HubMqttClient.cs
@@ -30,7 +30,7 @@ public HubMqttClient(IMqttClient c)
             genericDesiredUpdateProperty = new GenericDesiredUpdatePropertyBinder(c, updateTwinBinder!);
         }
 
-        public Func<GenericCommandRequest, GenericCommandResponse> OnCommandReceived
+        public Func<GenericCommandRequest, Task<GenericCommandResponse>> OnCommandReceived
         {
             get => command.OnCmdDelegate!;
             set => command.OnCmdDelegate = value;
@@ -62,7 +62,7 @@ public async Task<MqttClientPublishResult> SendTelemetryAsync(object payload, Ca
                 clientSegment = clientSegment.Replace("/", "/modules/");
             }
             return await Connection.PublishBinaryAsync($"devices/{clientSegment}/messages/events/",
-                new UTF8JsonSerializer().ToBytes(payload),
+                new Utf8JsonSerializer().ToBytes(payload),
                 Protocol.MqttQualityOfServiceLevel.AtLeastOnce,
                 false, t);
         }

diff --git a/src/MQTTnet.Extensions.MultiCloud.AzureIoTClient/IHubMqttClient.cs b/src/MQTTnet.Extensions.MultiCloud.AzureIoTClient/IHubMqttClient.cs
@@ -7,7 +7,7 @@ namespace MQTTnet.Extensions.MultiCloud.AzureIoTClient
     public interface IHubMqttClient
     {
         IMqttClient Connection { get; set; }
-        Func<GenericCommandRequest, GenericCommandResponse> OnCommandReceived { get; set; }
+        Func<GenericCommandRequest, Task<GenericCommandResponse>> OnCommandReceived { get; set; }
         Func<JsonNode, GenericPropertyAck> OnPropertyUpdateReceived { get; set; }
 
         Task<string> GetTwinAsync(CancellationToken cancellationToken = default);

diff --git a/src/MQTTnet.Extensions.MultiCloud.AzureIoTClient/Untyped/GenericCommandBinder.cs b/src/MQTTnet.Extensions.MultiCloud.AzureIoTClient/Untyped/GenericCommandBinder.cs
@@ -1,13 +1,12 @@
 using MQTTnet.Client;
 using MQTTnet.Extensions.MultiCloud.Binders;
-using System.Text;
 
 namespace MQTTnet.Extensions.MultiCloud.AzureIoTClient.Untyped
 {
     public class GenericCommand
     {
         private readonly IMqttClient connection;
-        public Func<GenericCommandRequest, GenericCommandResponse>? OnCmdDelegate { get; set; }
+        public Func<GenericCommandRequest, Task<GenericCommandResponse>>? OnCmdDelegate { get; set; }
 
         public GenericCommand(IMqttClient c)
         {
@@ -20,7 +19,7 @@ public GenericCommand(IMqttClient c)
                 {
                     var segments = topic.Split('/');
                     var cmdName = segments[3];
-                    string msg = Encoding.UTF8.GetString(m.ApplicationMessage.Payload);
+                    string msg = m.ApplicationMessage.ConvertPayloadToString();
                     GenericCommandRequest req = new()
                     {
                         CommandName = cmdName,
@@ -29,7 +28,7 @@ public GenericCommand(IMqttClient c)
                     if (OnCmdDelegate != null && req != null)
                     {
                         var tp = TopicParser.ParseTopic(topic);
-                        GenericCommandResponse response = OnCmdDelegate.Invoke(req);
+                        GenericCommandResponse response = await OnCmdDelegate.Invoke(req);
                         _ = connection.PublishStringAsync($"$iothub/methods/res/{response.Status}/?$rid={tp.Rid}", response.ReponsePayload);
                     }
                 }

diff --git a/src/MQTTnet.Extensions.MultiCloud.BrokerIoTClient/BrokerClientFactory.cs b/src/MQTTnet.Extensions.MultiCloud.BrokerIoTClient/BrokerClientFactory.cs
@@ -17,6 +17,7 @@ public static async Task<IMqttClient> CreateFromConnectionSettingsAsync(Connecti
             MqttClient? mqtt = new MqttFactory().CreateMqttClient(MqttNetTraceLogger.CreateTraceLogger()) as MqttClient;
             var connAck = await mqtt!.ConnectAsync(new MqttClientOptionsBuilder()
                 .WithConnectionSettings(cs, withBirth)
+                //.WithProtocolVersion(Formatter.MqttProtocolVersion.V500)
                 .Build(), cancellationToken);
             ComputedSettings = cs;
             if (connAck.ResultCode != MqttClientConnectResultCode.Success)
@@ -25,7 +26,7 @@ public static async Task<IMqttClient> CreateFromConnectionSettingsAsync(Connecti
             }
             if (withBirth)
             {
-                var birthPayload = new UTF8JsonSerializer().ToBytes(
+                var birthPayload = new Utf8JsonSerializer().ToBytes(
                        new BirthConvention.BirthMessage(BirthConvention.ConnectionStatus.online)
                        {
                            ModelId = cs.ModelId
@@ -34,11 +35,11 @@ public static async Task<IMqttClient> CreateFromConnectionSettingsAsync(Connecti
                 var pubAck = await mqtt.PublishBinaryAsync(
                    BirthConvention.BirthTopic(mqtt.Options.ClientId),
                    birthPayload,
-                   Protocol.MqttQualityOfServiceLevel.AtMostOnce, true, cancellationToken); 
-                //if (pubAck.ReasonCode != MqttClientPublishReasonCode.Success)
-                //{
-                //    throw new ApplicationException($"Error publishing Birth {cs}");
-                //}
+                   Protocol.MqttQualityOfServiceLevel.AtLeastOnce, true, cancellationToken); //hack to disable retained in registry
+                if (pubAck.ReasonCode != MqttClientPublishReasonCode.Success)
+                {
+                    throw new ApplicationException($"Error publishing Birth {cs}");
+                }
             }
             return mqtt;
         }

diff --git a/src/MQTTnet.Extensions.MultiCloud.BrokerIoTClient/PropertyClient.cs b/src/MQTTnet.Extensions.MultiCloud.BrokerIoTClient/PropertyClient.cs
@@ -16,7 +16,7 @@ public class PropertyClient<T>
 
     public Action<string,T>? OnPropertyUpdated { get; set; } = null;
 
-    public PropertyClient(IMqttClient client, string name) : this(client, name, new UTF8JsonSerializer()) { }
+    public PropertyClient(IMqttClient client, string name) : this(client, name, new Utf8JsonSerializer()) { }
 
     public PropertyClient(IMqttClient client, string name, IMessageSerializer messageSerializer)
     {

diff --git a/src/MQTTnet.Extensions.MultiCloud.BrokerIoTClient/ReadOnlyProperty.cs b/src/MQTTnet.Extensions.MultiCloud.BrokerIoTClient/ReadOnlyProperty.cs
@@ -29,7 +29,7 @@ public ReadOnlyProperty(IMqttClient mqttClient, string name)
         {
             if (m.ApplicationMessage.Topic == _topic)
             {
-                var ser = new UTF8JsonSerializer();
+                var ser = new Utf8JsonSerializer();
                 if (ser.TryReadFromBytes(m.ApplicationMessage.Payload, _name, out T propVal))
                 {
                     Value = propVal;

diff --git a/src/MQTTnet.Extensions.MultiCloud.BrokerIoTClient/TelemetryClient.cs b/src/MQTTnet.Extensions.MultiCloud.BrokerIoTClient/TelemetryClient.cs
@@ -13,7 +13,7 @@ public class TelemetryClient<T>
         public Action<string,T>? OnTelemetry { get; set; }
 
         public TelemetryClient(IMqttClient client, string name) 
-            : this(client, name, new UTF8JsonSerializer())
+            : this(client, name, new Utf8JsonSerializer())
         {
 
         }

diff --git a/src/MQTTnet.Extensions.MultiCloud.BrokerIoTClient/Untyped/GenericCommandBinder.cs b/src/MQTTnet.Extensions.MultiCloud.BrokerIoTClient/Untyped/GenericCommandBinder.cs
@@ -12,7 +12,7 @@ public class GenericCommand
 
     public GenericCommand(IMqttClient c)
     {
-        _serializer = new UTF8JsonSerializer();
+        _serializer = new Utf8JsonSerializer();
         connection = c;
         _ = connection.SubscribeWithReplyAsync($"device/{c.Options.ClientId}/commands/+");
         connection.ApplicationMessageReceivedAsync += async m =>
@@ -23,7 +23,7 @@ public GenericCommand(IMqttClient c)
                 var segments = topic.Split('/');
                 var cmdName = segments[3];
 
-                if (_serializer.TryReadFromBytes(m.ApplicationMessage.Payload, string.Empty, out object reqPayload))
+                if (_serializer.TryReadFromBytes(m.ApplicationMessage.Payload, string.Empty, out string reqPayload))
                 {
                     var responseTopic = m.ApplicationMessage.ResponseTopic ?? $"{topic}/resp";
 

diff --git a/src/MQTTnet.Extensions.MultiCloud.BrokerIoTClient/Untyped/GenericCommandClient.cs b/src/MQTTnet.Extensions.MultiCloud.BrokerIoTClient/Untyped/GenericCommandClient.cs
@@ -23,7 +23,7 @@ public GenericCommandClient(IMqttClient client)
     {
         _mqttClient = client;
         _remoteClientId = string.Empty;
-        _serializer = new UTF8JsonSerializer();
+        _serializer = new Utf8JsonSerializer();
 
         _mqttClient.ApplicationMessageReceivedAsync += async m =>
         {
@@ -40,7 +40,7 @@ public GenericCommandClient(IMqttClient client)
                 var up = m.ApplicationMessage.UserProperties.FirstOrDefault(p => p.Name.Equals("status"));
                 int status = up != null ? int.Parse(up.Value) : 500;
 
-                if (_serializer.TryReadFromBytes(m.ApplicationMessage.Payload, string.Empty, out object respPayload))
+                if (_serializer.TryReadFromBytes(m.ApplicationMessage.Payload, string.Empty, out string respPayload))
                 {
                     GenericCommandResponse resp = new()
                     {
@@ -67,20 +67,16 @@ public async Task<GenericCommandResponse> InvokeAsync(string clientId, GenericCo
 
         string commandTopic = requestTopicPattern.Replace("{clientId}", _remoteClientId).Replace("{commandName}", _commandName);
         var responseTopic = responseTopicSub.Replace("{clientId}", _remoteClientId).Replace("{commandName}", _commandName);
-        await _mqttClient.SubscribeAsync(responseTopic, Protocol.MqttQualityOfServiceLevel.AtMostOnce, ct);
+        _ =_mqttClient.SubscribeAsync(responseTopic, Protocol.MqttQualityOfServiceLevel.AtMostOnce, ct);
 
-        var pubAck = await _mqttClient.PublishAsync(
+        _ = _mqttClient.PublishAsync(
             new MqttApplicationMessageBuilder()
                 .WithTopic(commandTopic)
                 .WithPayload(_serializer.ToBytes(request.RequestPayload))
                 .WithResponseTopic(responseTopicSuccess.Replace("{clientId}", _remoteClientId).Replace("{commandName}", _commandName))
                 .WithCorrelationData(corr.ToByteArray())
                 .Build());
-
-        if (!pubAck.IsSuccess)
-        {
-            throw new ApplicationException("Error publishing Request Message");
-        }
+
         return await _tcs.Task.TimeoutAfter(TimeSpan.FromSeconds(5));
 
     }

diff --git a/src/MQTTnet.Extensions.MultiCloud/Binders/CloudToDeviceBinder.cs b/src/MQTTnet.Extensions.MultiCloud/Binders/CloudToDeviceBinder.cs
@@ -20,7 +20,7 @@ public abstract class CloudToDeviceBinder<T, TResp> : ICloudToDevice<T, TResp>
     protected Action<TopicParameters>? PreProcessMessage;
 
     public CloudToDeviceBinder(IMqttClient connection, string name)
-        : this(connection, name, new UTF8JsonSerializer()) { }
+        : this(connection, name, new Utf8JsonSerializer()) { }
 
     public CloudToDeviceBinder(IMqttClient connection, string name, IMessageSerializer serializer)
     {

diff --git a/src/MQTTnet.Extensions.MultiCloud/Binders/DeviceToCloudBinder.cs b/src/MQTTnet.Extensions.MultiCloud/Binders/DeviceToCloudBinder.cs
@@ -15,7 +15,7 @@ public abstract class DeviceToCloudBinder<T> : IDeviceToCloud<T>
     public bool WrapMessage = false;
     public bool Retain = false;
 
-    public DeviceToCloudBinder(IMqttClient mqttClient, string name) : this(mqttClient, name, new UTF8JsonSerializer()) { }
+    public DeviceToCloudBinder(IMqttClient mqttClient, string name) : this(mqttClient, name, new Utf8JsonSerializer()) { }
 
     public DeviceToCloudBinder(IMqttClient mqttClient, string name, IMessageSerializer ser)
     {

diff --git a/src/MQTTnet.Extensions.MultiCloud/Binders/RequestResponseBinder.cs b/src/MQTTnet.Extensions.MultiCloud/Binders/RequestResponseBinder.cs
@@ -23,7 +23,7 @@ public class RequestResponseBinder<T, TResp>
     readonly IMessageSerializer _serializer;
 
     public RequestResponseBinder(IMqttClient client, string name, bool unwrap)
-        : this(client, name, unwrap, new UTF8JsonSerializer())
+        : this(client, name, unwrap, new Utf8JsonSerializer())
     {
 
     }

diff --git a/src/MQTTnet.Extensions.MultiCloud/Connections/BirthConvention.cs b/src/MQTTnet.Extensions.MultiCloud/Connections/BirthConvention.cs
@@ -27,7 +27,7 @@ public BirthMessage(ConnectionStatus connectionStatus)
     }
 
     public static byte[] LastWillPayload() =>
-        new UTF8JsonSerializer().ToBytes(new BirthMessage(ConnectionStatus.offline));
+        new Utf8JsonSerializer().ToBytes(new BirthMessage(ConnectionStatus.offline));
     public static byte[] LastWillPayload(string modelId) =>
-        new UTF8JsonSerializer().ToBytes(new BirthMessage(ConnectionStatus.offline) { ModelId = modelId });
+        new Utf8JsonSerializer().ToBytes(new BirthMessage(ConnectionStatus.offline) { ModelId = modelId });
 }
diff --git a/src/MQTTnet.Extensions.MultiCloud/Serializers/UTF8JsonSerializer.cs b/src/MQTTnet.Extensions.MultiCloud/Serializers/UTF8JsonSerializer.cs
@@ -4,7 +4,7 @@
 
 namespace MQTTnet.Extensions.MultiCloud.Serializers;
 
-public class UTF8JsonSerializer : IMessageSerializer
+public class Utf8JsonSerializer : IMessageSerializer
 {
     private static class Json
     {
@@ -44,6 +44,7 @@ public byte[] ToBytes<T>(T payload, string name = "")
             {
                 return Encoding.UTF8.GetBytes(Json.Stringify(payload!));
             }
+
         }
         else
         {
@@ -67,7 +68,7 @@ public bool TryReadFromBytes<T>(byte[] payload, string name, out T result)
             if (typeof(T) == typeof(string))
             {
                 result = (T)Convert.ChangeType(Encoding.UTF8.GetString(payload), typeof(T));
-            } 
+            }
             else if (typeof(T) == typeof(object))
             {
                 result = (T)Convert.ChangeType(Encoding.UTF8.GetString(payload), typeof(T))!;
@@ -76,6 +77,7 @@ public bool TryReadFromBytes<T>(byte[] payload, string name, out T result)
             {
                 result = Json.FromString<T>(Encoding.UTF8.GetString(payload))!;
             }
+
         }
         else
         {

diff --git a/src/MQTTnet.Extensions.MultiCloud/Serializers/Utf8StringSerializer.cs b/src/MQTTnet.Extensions.MultiCloud/Serializers/Utf8StringSerializer.cs
@@ -0,0 +1,20 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace MQTTnet.Extensions.MultiCloud.Serializers
+{
+    public class Utf8StringSerializer : IMessageSerializer
+    {
+        public byte[] ToBytes<T>(T payload, string name = "") => Encoding.UTF8.GetBytes((payload as string)!);
+
+
+        public bool TryReadFromBytes<T>(byte[] payload, string name, out T result) 
+        {
+            result = (T)Convert.ChangeType(Encoding.UTF8.GetString(payload), typeof(T));
+            return true;
+        }
+    }
+}
diff --git a/tests/MQTTnet.Extensions.MultiCloud.IntegrationTests/e2e/HubEndToEndFixture.cs b/tests/MQTTnet.Extensions.MultiCloud.IntegrationTests/e2e/HubEndToEndFixture.cs
@@ -227,6 +227,38 @@ internal async Task CommandsGetCalled()
             await rm.RemoveDeviceAsync(deviceId);
 
         }
+        [Fact]
+        internal async Task InvokeGenericCommand()
+        {
+            var deviceId = "gencmd" + new Random().Next(100);
+            var device = await GetOrCreateDeviceAsync(deviceId);
+            bool commandInvoked = false;
+            string commandName = string.Empty;
+            string commandRequestPayload = string.Empty;
+            var td = new HubMqttClient(await HubDpsFactory.CreateFromConnectionSettingsAsync($"HostName={hubName};DeviceId={deviceId};SharedAccessKey={device.Authentication.SymmetricKey.PrimaryKey}"));
+            td.OnCommandReceived += async cmd =>
+            {
+                commandName = cmd.CommandName!;
+                commandRequestPayload = cmd.CommandPayload!.ToString();
+                commandInvoked = true;
+                return await Task.FromResult(new AzureIoTClient.Untyped.GenericCommandResponse 
+                    { 
+                    Status = 200, 
+                    ReponsePayload = JsonSerializer.Serialize(new { myCommandResponse = "adios"})
+                    });
+            };
+            await Task.Delay(200);
+            var sc = ServiceClient.CreateFromConnectionString(hubConnectionString);
+            CloudToDeviceMethod c2dMethod = new("aCommand");
+            string requestPayload = JsonSerializer.Serialize(new { myComandRequest = "hello" });
+            c2dMethod.SetPayloadJson(requestPayload); ;
+            var dmRes = await sc.InvokeDeviceMethodAsync(deviceId, c2dMethod);
+            Assert.True(commandInvoked);
+            Assert.Equal("aCommand", commandName);
+            Assert.Equal(requestPayload, commandRequestPayload);
+            string expectedJson = Json.Stringify(new { myCommandResponse = "adios" });
+            Assert.Equal(expectedJson, dmRes.GetPayloadAsJson());
+        }
 
         private async Task<Device> GetOrCreateDeviceAsync(string deviceId, bool x509 = false)
         {

diff --git a/tests/MQTTnet.Extensions.MultiCloud.UnitTests/BrokerJsonBindings/CommandBinderFixture.cs b/tests/MQTTnet.Extensions.MultiCloud.UnitTests/BrokerJsonBindings/CommandBinderFixture.cs
@@ -1,9 +1,5 @@
 using MQTTnet.Extensions.MultiCloud.BrokerIoTClient;
 using MQTTnet.Extensions.MultiCloud.Serializers;
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
 using System.Threading.Tasks;
 using Xunit;
 
@@ -20,12 +16,12 @@ public void CommandWithReqResp()
             cmd.OnMessage = async req =>
             {
                 cmdReceived = true;
-                return await Task.FromResult(req.ToString());
+                return await Task.FromResult($"received {req}");
             };
-            mockMqtt.SimulateNewBinaryMessage("device/mock/commands/aCmdReqResp",new UTF8JsonSerializer().ToBytes(2));
+            mockMqtt.SimulateNewBinaryMessage("device/mock/commands/aCmdReqResp",new Utf8JsonSerializer().ToBytes(2));
             Assert.True(cmdReceived);
             Assert.Equal("device/mock/commands/aCmdReqResp/resp", mockMqtt.topicRecceived);
-            Assert.Equal("2", mockMqtt.payloadReceived);
+            Assert.Equal("received 2", mockMqtt.payloadReceived);
         }
 
         [Fact]
@@ -39,7 +35,7 @@ public void CommandWithReq()
                 cmdReceived = true;
                 return await Task.FromResult(string.Empty);
             };
-            mockMqtt.SimulateNewBinaryMessage("device/mock/commands/aCmdReq", new UTF8JsonSerializer().ToBytes(2));
+            mockMqtt.SimulateNewBinaryMessage("device/mock/commands/aCmdReq", new Utf8JsonSerializer().ToBytes(2));
             Assert.True(cmdReceived);
             Assert.Equal("device/mock/commands/aCmdReq/resp", mockMqtt.topicRecceived);
             Assert.Empty(mockMqtt.payloadReceived);
@@ -56,7 +52,7 @@ public void CommandWithRes()
                 cmdReceived = true;
                 return await Task.FromResult(1);
             };
-            mockMqtt.SimulateNewBinaryMessage("device/mock/commands/aCmdRes", new UTF8JsonSerializer().ToBytes(""));
+            mockMqtt.SimulateNewBinaryMessage("device/mock/commands/aCmdRes", new Utf8JsonSerializer().ToBytes(""));
             Assert.True(cmdReceived);
             Assert.Equal("device/mock/commands/aCmdRes/resp", mockMqtt.topicRecceived);
             Assert.Equal("1", mockMqtt.payloadReceived);
@@ -73,7 +69,7 @@ public void CommandEmpty()
                 cmdReceived = true;
                 return await Task.FromResult(string.Empty);
             };
-            mockMqtt.SimulateNewBinaryMessage("device/mock/commands/aCmd", new UTF8JsonSerializer().ToBytes(""));
+            mockMqtt.SimulateNewBinaryMessage("device/mock/commands/aCmd", new Utf8JsonSerializer().ToBytes(""));
             Assert.True(cmdReceived);
             Assert.Equal("device/mock/commands/aCmd/resp", mockMqtt.topicRecceived);
             Assert.Empty(mockMqtt.payloadReceived);