From b28b51570abbfcf5c42ce02840a2f6ea3b93ae53 Mon Sep 17 00:00:00 2001 From: KneShell Date: Sat, 19 Aug 2023 13:56:58 +0900 Subject: [PATCH] Update MQTTnet nuget package version 3.1.1 to 4.2.1.781 (#2272) (#2273) Change code to be compatible with MQTTnet 4.2.781 version. And check if all test passed. (It is) --- Libraries/Opc.Ua.PubSub/Opc.Ua.PubSub.csproj | 2 +- .../Transport/MqttClientCreator.cs | 38 ++++++++----------- .../Transport/MqttPubSubConnection.cs | 21 +++++----- 3 files changed, 28 insertions(+), 33 deletions(-) diff --git a/Libraries/Opc.Ua.PubSub/Opc.Ua.PubSub.csproj b/Libraries/Opc.Ua.PubSub/Opc.Ua.PubSub.csproj index f6fc47dc86..a9f12be749 100644 --- a/Libraries/Opc.Ua.PubSub/Opc.Ua.PubSub.csproj +++ b/Libraries/Opc.Ua.PubSub/Opc.Ua.PubSub.csproj @@ -36,7 +36,7 @@ - + diff --git a/Libraries/Opc.Ua.PubSub/Transport/MqttClientCreator.cs b/Libraries/Opc.Ua.PubSub/Transport/MqttClientCreator.cs index 8a9fe1ac90..64a13db41c 100644 --- a/Libraries/Opc.Ua.PubSub/Transport/MqttClientCreator.cs +++ b/Libraries/Opc.Ua.PubSub/Transport/MqttClientCreator.cs @@ -33,8 +33,6 @@ using System.Threading; using System.Threading.Tasks; using MQTTnet; using MQTTnet.Client; -using MQTTnet.Client.Connecting; -using MQTTnet.Client.Options; namespace Opc.Ua.PubSub.Transport { @@ -53,8 +51,8 @@ namespace Opc.Ua.PubSub.Transport /// The topics to which to subscribe /// internal static async Task GetMqttClientAsync(int reconnectInterval, - IMqttClientOptions mqttClientOptions, - Action receiveMessageHandler, + MqttClientOptions mqttClientOptions, + Func receiveMessageHandler, StringCollection topicFilter = null) { IMqttClient mqttClient = mqttClientFactory.Value.CreateMqttClient(); @@ -62,30 +60,26 @@ namespace Opc.Ua.PubSub.Transport // Hook the receiveMessageHandler in case we deal with a subscriber if ((receiveMessageHandler != null) && (topicFilter != null)) { - HashSet uniqueTopics = new HashSet(topicFilter.ToArray()); - List topics = new List(); - foreach (var topic in uniqueTopics) - { - MqttTopicFilter tf = new MqttTopicFilter(); - tf.Topic = topic; - topics.Add(tf); - } - mqttClient.UseApplicationMessageReceivedHandler(receiveMessageHandler); - mqttClient.UseConnectedHandler(async e => { + mqttClient.ApplicationMessageReceivedAsync += receiveMessageHandler; + mqttClient.ConnectedAsync += async e => { Utils.Trace("{0} Connected to MQTTBroker", mqttClient?.Options?.ClientId); try { - // subscribe to provided topics, messages are also filtered on the receiveMessageHandler - await mqttClient.SubscribeAsync(topics.ToArray()).ConfigureAwait(false); - Utils.Trace("{0} Subscribed to topics: {1}", mqttClient?.Options?.ClientId, string.Join(",", topics)); + foreach (string topic in topicFilter) + { + // subscribe to provided topics, messages are also filtered on the receiveMessageHandler + await mqttClient.SubscribeAsync(topic).ConfigureAwait(false); + } + + Utils.Trace("{0} Subscribed to topics: {1}", mqttClient?.Options?.ClientId, string.Join(",", topicFilter)); } catch (Exception exception) { - Utils.Trace(exception, "{0} could not subscribe to topics: {1}", mqttClient?.Options?.ClientId, string.Join(",", topics)); + Utils.Trace(exception, "{0} could not subscribe to topics: {1}", mqttClient?.Options?.ClientId, string.Join(",", topicFilter)); } - }); + }; } else { @@ -100,7 +94,7 @@ namespace Opc.Ua.PubSub.Transport } // Setup reconnect handler - mqttClient.UseDisconnectedHandler(async e => { + mqttClient.DisconnectedAsync += async e => { await Task.Delay(TimeSpan.FromSeconds(reconnectInterval)).ConfigureAwait(false); try { @@ -114,7 +108,7 @@ namespace Opc.Ua.PubSub.Transport { Utils.Trace("{0} Failed to reconnect after disconnect occurred: {1}", mqttClient?.Options?.ClientId, excOnDisconnect.Message); } - }); + }; await Connect(reconnectInterval, mqttClientOptions, mqttClient).ConfigureAwait(false); @@ -127,7 +121,7 @@ namespace Opc.Ua.PubSub.Transport /// /// /// - private static async Task Connect(int reconnectInterval, IMqttClientOptions mqttClientOptions, IMqttClient mqttClient) + private static async Task Connect(int reconnectInterval, MqttClientOptions mqttClientOptions, IMqttClient mqttClient) { try { diff --git a/Libraries/Opc.Ua.PubSub/Transport/MqttPubSubConnection.cs b/Libraries/Opc.Ua.PubSub/Transport/MqttPubSubConnection.cs index 889d78603d..845a517876 100644 --- a/Libraries/Opc.Ua.PubSub/Transport/MqttPubSubConnection.cs +++ b/Libraries/Opc.Ua.PubSub/Transport/MqttPubSubConnection.cs @@ -33,7 +33,6 @@ using System.Security.Cryptography.X509Certificates; using System.Threading.Tasks; using MQTTnet; using MQTTnet.Client; -using MQTTnet.Client.Options; using MQTTnet.Formatter; using MQTTnet.Protocol; using Opc.Ua.PubSub.Encoding; @@ -246,7 +245,7 @@ namespace Opc.Ua.PubSub.Transport { var message = new MqttApplicationMessage { Topic = queueName, - Payload = bytes, + PayloadSegment = new ArraySegment(bytes), QualityOfServiceLevel = GetMqttQualityOfServiceLevel(qos), Retain = networkMessage.IsMetaDataMessage }; @@ -360,7 +359,7 @@ namespace Opc.Ua.PubSub.Transport MqttClient publisherClient = null; MqttClient subscriberClient = null; - IMqttClientOptions mqttOptions = GetMqttClientOptions(); + MqttClientOptions mqttOptions = GetMqttClientOptions(); int nrOfPublishers = Publishers.Count; int nrOfSubscribers = GetAllDataSetReaders().Count; @@ -508,7 +507,7 @@ namespace Opc.Ua.PubSub.Transport /// Processes a message from the MQTT broker. /// /// - private void ProcessMqttMessage(MqttApplicationMessageReceivedEventArgs eventArgs) + private Task ProcessMqttMessage(MqttApplicationMessageReceivedEventArgs eventArgs) { string topic = eventArgs.ApplicationMessage.Topic; @@ -549,7 +548,7 @@ namespace Opc.Ua.PubSub.Transport { // raise RawData received event RawDataReceivedEventArgs rawDataReceivedEventArgs = new RawDataReceivedEventArgs() { - Message = eventArgs.ApplicationMessage.Payload, + Message = eventArgs.ApplicationMessage.PayloadSegment.Array, Source = topic, TransportProtocol = this.TransportProtocol, MessageMapping = m_messageMapping, @@ -563,7 +562,7 @@ namespace Opc.Ua.PubSub.Transport if (rawDataReceivedEventArgs.Handled) { Utils.Trace("MqttConnection message from topic={0} is marked as handled and will not be decoded.", topic); - return; + return Task.CompletedTask; } // initialize the expected NetworkMessage @@ -572,7 +571,7 @@ namespace Opc.Ua.PubSub.Transport // trigger message decoding if (networkMessage != null) { - networkMessage.Decode(MessageContext, eventArgs.ApplicationMessage.Payload, dataSetReaders); + networkMessage.Decode(MessageContext, eventArgs.ApplicationMessage.PayloadSegment.Array, dataSetReaders); // Handle the decoded message and raise the necessary event on UaPubSubApplication ProcessDecodedNetworkMessage(networkMessage, topic); @@ -582,6 +581,8 @@ namespace Opc.Ua.PubSub.Transport { Utils.Trace("MqttConnection - ProcessMqttMessage() No DataSetReader is registered for topic={0}.", topic); } + + return Task.CompletedTask; } /// @@ -608,9 +609,9 @@ namespace Opc.Ua.PubSub.Transport /// Get appropriate IMqttClientOptions with which to connect to the MQTTBroker /// /// - private IMqttClientOptions GetMqttClientOptions() + private MqttClientOptions GetMqttClientOptions() { - IMqttClientOptions mqttOptions = null; + MqttClientOptions mqttOptions = null; TimeSpan mqttKeepAlive = TimeSpan.FromSeconds(GetWriterGroupsMaxKeepAlive() + MaxKeepAliveIncrement); NetworkAddressUrlDataType networkAddressUrlState = @@ -762,7 +763,7 @@ namespace Opc.Ua.PubSub.Transport /// Validates the broker certificate. /// /// The context of the validation - private bool ValidateBrokerCertificate(MqttClientCertificateValidationCallbackContext context) + private bool ValidateBrokerCertificate(MqttClientCertificateValidationEventArgs context) { X509Certificate2 brokerCertificate = new X509Certificate2(context.Certificate.GetRawCertData());