GithubStar/UA-.NETStandard
1
0
Fork 0
mirror of https://github.com/OPCFoundation/UA-.NETStandard.git synced 2026-10-29 07:07:24 +08:00
Code Issues Packages Projects Releases Wiki Activity

Update MQTTnet nuget package version 3.1.1 to 4.2.1.781 (#2272) (#2273)

Browse Source 
Change code to be compatible with MQTTnet 4.2.781 version.
And check if all test passed. (It is)
This commit is contained in:
KneShell
2023-08-19 06:56:58 +02:00
committed by GitHub
parent 7f864c1ef6
commit b28b51570a
3 changed files with 28 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Libraries/Opc.Ua.PubSub/Opc.Ua.PubSub.csproj
+1 -1
View File
@@ -36,7 +36,7 @@
</ItemGroup>
<ItemGroup>
<PackageReference Include="MQTTnet" Version="3.1.1" />
<PackageReference Include="MQTTnet" Version="4.2.1.781" />
<PackageReference Include="System.Net.NetworkInformation" Version="4.3.0" />
</ItemGroup>
Libraries/Opc.Ua.PubSub/Transport/MqttClientCreator.cs
+16 -22
View File
@@ -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
/// <param name="topicFilter">The topics to which to subscribe</param>
/// <returns></returns>
internal static async Task<IMqttClient> GetMqttClientAsync(int reconnectInterval,
IMqttClientOptions mqttClientOptions,
Action<MqttApplicationMessageReceivedEventArgs> receiveMessageHandler,
MqttClientOptions mqttClientOptions,
Func<MqttApplicationMessageReceivedEventArgs, Task> 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<string> uniqueTopics = new HashSet<string>(topicFilter.ToArray());
List<MqttTopicFilter> topics = new List<MqttTopicFilter>();
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
/// <param name="reconnectInterval"></param>
/// <param name="mqttClientOptions"></param>
/// <param name="mqttClient"></param>
private static async Task Connect(int reconnectInterval, IMqttClientOptions mqttClientOptions, IMqttClient mqttClient)
private static async Task Connect(int reconnectInterval, MqttClientOptions mqttClientOptions, IMqttClient mqttClient)
{
try
{
Libraries/Opc.Ua.PubSub/Transport/MqttPubSubConnection.cs
+11 -10
View File
@@ -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<byte>(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.
/// </summary>
/// <param name="eventArgs"></param>
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;
}
/// <summary>
@@ -608,9 +609,9 @@ namespace Opc.Ua.PubSub.Transport
/// Get appropriate IMqttClientOptions with which to connect to the MQTTBroker
/// </summary>
/// <returns></returns>
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.
/// </summary>
/// <param name="context">The context of the validation</param>
private bool ValidateBrokerCertificate(MqttClientCertificateValidationCallbackContext context)
private bool ValidateBrokerCertificate(MqttClientCertificateValidationEventArgs context)
{
X509Certificate2 brokerCertificate = new X509Certificate2(context.Certificate.GetRawCertData());