From 8231d62f9f54f96715d880fb82f6a711223f8694 Mon Sep 17 00:00:00 2001
From: Christian <6939810+chkr1011@users.noreply.github.com>
Date: Mon, 27 Jun 2022 20:39:53 +0200
Subject: [PATCH] Squashed commit of the following:

commit 9f289152736a682a8c31fbe7720dade37dbb1cb5
Author: Christian <6939810+chkr1011@users.noreply.github.com>
Date:   Sun Jun 26 13:38:35 2022 +0200

    Fix message storage.
---
 .github/workflows/ReleaseNotes.md                |  3 ++-
 Source/MQTTnet/Implementations/MqttTcpChannel.cs | 13 +++++++++----
 Source/MQTTnet/Server/Internal/MqttSession.cs    |  5 +++--
 3 files changed, 14 insertions(+), 7 deletions(-)

diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md
index 19de4389..49bf5684 100644
--- a/.github/workflows/ReleaseNotes.md
+++ b/.github/workflows/ReleaseNotes.md
@@ -1 +1,2 @@
-* [Server] Fixed reporting of _MaximumQoS_ in _ConnAck_ packet (MQTTv5 only) (#1442).
\ No newline at end of file
+* [Server] Fixed reporting of _MaximumQoS_ in _ConnAck_ packet (MQTTv5 only) (#1442).
+* [Server] Fix cross thread issue in session message storage for QoS 1 and 2.
\ No newline at end of file
diff --git a/Source/MQTTnet/Implementations/MqttTcpChannel.cs b/Source/MQTTnet/Implementations/MqttTcpChannel.cs
index bc85bcde..5dbfd8dd 100644
--- a/Source/MQTTnet/Implementations/MqttTcpChannel.cs
+++ b/Source/MQTTnet/Implementations/MqttTcpChannel.cs
@@ -77,7 +77,7 @@ namespace MQTTnet.Implementations
                 {
                     socket.LingerState = _tcpOptions.LingerState;
                 }
-                
+
                 if (_tcpOptions.DualMode.HasValue)
                 {
                     // It is important to avoid setting the flag if no specific value is set by the user
@@ -103,15 +103,20 @@ namespace MQTTnet.Implementations
                             ApplicationProtocols = _tcpOptions.TlsOptions.ApplicationProtocols,
                             ClientCertificates = LoadCertificates(),
                             EnabledSslProtocols = _tcpOptions.TlsOptions.SslProtocol,
-                            CertificateRevocationCheckMode = _tcpOptions.TlsOptions.IgnoreCertificateRevocationErrors ? X509RevocationMode.NoCheck : _tcpOptions.TlsOptions.RevocationMode,
+                            CertificateRevocationCheckMode =
+ _tcpOptions.TlsOptions.IgnoreCertificateRevocationErrors ? X509RevocationMode.NoCheck : _tcpOptions.TlsOptions.RevocationMode,
                             TargetHost = _tcpOptions.Server,
                             CipherSuitesPolicy = _tcpOptions.TlsOptions.CipherSuitesPolicy
                         };
 
                         await sslStream.AuthenticateAsClientAsync(sslOptions, cancellationToken).ConfigureAwait(false);
 #else
-                        await sslStream.AuthenticateAsClientAsync(_tcpOptions.Server, LoadCertificates(), _tcpOptions.TlsOptions.SslProtocol,
-                            !_tcpOptions.TlsOptions.IgnoreCertificateRevocationErrors).ConfigureAwait(false);
+                        await sslStream.AuthenticateAsClientAsync(
+                                _tcpOptions.Server,
+                                LoadCertificates(),
+                                _tcpOptions.TlsOptions.SslProtocol,
+                                !_tcpOptions.TlsOptions.IgnoreCertificateRevocationErrors)
+                            .ConfigureAwait(false);
 #endif
                     }
                     catch
diff --git a/Source/MQTTnet/Server/Internal/MqttSession.cs b/Source/MQTTnet/Server/Internal/MqttSession.cs
index 557a4dba..e152106a 100644
--- a/Source/MQTTnet/Server/Internal/MqttSession.cs
+++ b/Source/MQTTnet/Server/Internal/MqttSession.cs
@@ -4,6 +4,7 @@
 
 using System;
 using System.Collections;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading;
@@ -23,7 +24,7 @@ namespace MQTTnet.Server
 
         readonly MqttServerOptions _serverOptions;
 
-        readonly Dictionary<ushort, MqttPublishPacket> _unacknowledgedPublishPackets = new Dictionary<ushort, MqttPublishPacket>();
+        readonly ConcurrentDictionary<ushort, MqttPublishPacket> _unacknowledgedPublishPackets = new ConcurrentDictionary<ushort, MqttPublishPacket>();
 
         // Bookkeeping to know if this is a subscribing client; lazy initialize later.
         HashSet<string> _subscribedTopics;
@@ -72,7 +73,7 @@ namespace MQTTnet.Server
 
         public void AcknowledgePublishPacket(ushort packetIdentifier)
         {
-            _unacknowledgedPublishPackets.Remove(packetIdentifier);
+            _unacknowledgedPublishPackets.TryRemove(packetIdentifier, out _);
         }
 
         public void AddSubscribedTopic(string topic)