Avoid using obsolete code

* implement intercept publish message event on ManagedMqttClient

* rename ShouldPublish to ACceptPublish per PR feedback

.github/workflows/ReleaseNotes.md

@@ -1,3 +1,5 @@
+* [Client] Exposed the _EndPoint_ type to support other endpoint types (like Unix Domain Sockets) in client options (#1919).
+* [ManagedClient] Added support for intercepting (and cancelling) publish messages (#1915, thanks to @hannasm).
 * [Server] Added new events for delivered and dropped messages (#1866, thanks to @kallayj).
 * [Server] The server will no longer treat a client which is receiving a large payload as alive. The packet must be received completely within the keep alive boundaries (BREAKING CHANGE!, #1883).
 * [Server] Fixed "service not registered" exception in ASP.NET integration (#1889).

MQTTnet.sln.DotSettings

@@ -230,6 +230,7 @@
 	<s:String x:Key="/Default/CodeStyle/FileHeader/FileHeaderText/@EntryValue">Licensed to the .NET Foundation under one or more agreements.
 The .NET Foundation licenses this file to you under the MIT license.
 See the LICENSE file in the project root for more information.</s:String>
+	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EFeature_002EServices_002ECodeCleanup_002EFileHeader_002EFileHeaderSettingsMigrate/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpAttributeForSingleLineMethodUpgrade/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpKeepExistingMigration/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpPlaceEmbeddedOnSameLineMigration/@EntryIndexedValue">True</s:Boolean>

Source/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs

@@ -48,7 +48,6 @@ namespace MQTTnet.AspNetCore.Tests
         {
             var mockup = new ConnectionHandlerMockup();
 
-
             using (var host = new WebHostBuilder().UseKestrel(kestrel => kestrel.ListenLocalhost(1883, listener => listener.Use((ctx, next) => mockup.OnConnectedAsync(ctx))))
                        .UseStartup<Startup>()
                        .ConfigureServices(

Source/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs

@@ -6,7 +6,6 @@ using MQTTnet.Adapter;
 using MQTTnet.AspNetCore.Client.Tcp;
 using MQTTnet.Formatter;
 using System;
-using System.Net;
 using MQTTnet.Client;
 using MQTTnet.Diagnostics;
 
@@ -22,8 +21,7 @@ namespace MQTTnet.AspNetCore.Client
             {
                 case MqttClientTcpOptions tcpOptions:
                     {
-                        var endpoint = new DnsEndPoint(tcpOptions.Server, tcpOptions.GetPort());
+                        var tcpConnection = new SocketConnection(tcpOptions.RemoteEndpoint);
-                        var tcpConnection = new TcpConnection(endpoint);
                         
                         var formatter = new MqttPacketFormatterAdapter(options.ProtocolVersion, new MqttBufferWriter(4096, 65535));
                         return new MqttConnectionContext(formatter, tcpConnection);

Source/MQTTnet.AspnetCore/Client/Tcp/SocketConnection.cs

@@ -15,15 +15,15 @@ using System.Threading.Tasks;
 
 namespace MQTTnet.AspNetCore.Client.Tcp
 {
-    public class TcpConnection : ConnectionContext
+    public sealed class SocketConnection : ConnectionContext
     {
-        private volatile bool _aborted;
+        volatile bool _aborted;
-        private readonly EndPoint _endPoint;
+        readonly EndPoint _endPoint;
-        private SocketSender _sender;
+        SocketSender _sender;
-        private SocketReceiver _receiver;
+        SocketReceiver _receiver;
 
-        private Socket _socket;
+        Socket _socket;
-        private IDuplexPipe _application;
+        IDuplexPipe _application;
 
         public bool IsConnected { get; private set; }
         public override string ConnectionId { get; set; }
@@ -31,12 +31,12 @@ namespace MQTTnet.AspNetCore.Client.Tcp
         public override IDictionary<object, object> Items { get; set; }
         public override IDuplexPipe Transport { get; set; }
 
-        public TcpConnection(EndPoint endPoint)
+        public SocketConnection(EndPoint endPoint)
         {
             _endPoint = endPoint;
         }
 
-        public TcpConnection(Socket socket)
+        public SocketConnection(Socket socket)
         {
             _socket = socket;
             _endPoint = socket.RemoteEndPoint;
@@ -87,7 +87,7 @@ namespace MQTTnet.AspNetCore.Client.Tcp
             IsConnected = true;
         }
 
-        private async Task ExecuteAsync()
+        async Task ExecuteAsync()
         {
             Exception sendError = null;
             try
@@ -114,7 +114,7 @@ namespace MQTTnet.AspNetCore.Client.Tcp
             }
             catch (Exception ex)
             {
-                Console.WriteLine($"Unexpected exception in {nameof(TcpConnection)}.{nameof(StartAsync)}: " + ex);
+                Console.WriteLine($"Unexpected exception in {nameof(SocketConnection)}.{nameof(StartAsync)}: " + ex);
             }
             finally
             {
@@ -122,7 +122,8 @@ namespace MQTTnet.AspNetCore.Client.Tcp
                 _application.Input.Complete(sendError);
             }
         }
-        private async Task DoReceive()
+
+        async Task DoReceive()
         {
             Exception error = null;
 
@@ -171,7 +172,7 @@ namespace MQTTnet.AspNetCore.Client.Tcp
             }
         }
 
-        private async Task ProcessReceives()
+        async Task ProcessReceives()
         {
             while (true)
             {
@@ -204,12 +205,12 @@ namespace MQTTnet.AspNetCore.Client.Tcp
             }
         }
 
-        private Exception ConnectionAborted()
+        Exception ConnectionAborted()
         {
             return new MqttCommunicationException("Connection Aborted");
         }
 
-        private async Task<Exception> DoSend()
+        async Task<Exception> DoSend()
         {
             Exception error = null;
 
@@ -240,7 +241,7 @@ namespace MQTTnet.AspNetCore.Client.Tcp
             return error;
         }
 
-        private async Task ProcessSends()
+        async Task ProcessSends()
         {
             while (true)
             {

Source/MQTTnet.AspnetCore/MqttConnectionContext.cs

@@ -33,7 +33,7 @@ namespace MQTTnet.AspNetCore
             PacketFormatterAdapter = packetFormatterAdapter ?? throw new ArgumentNullException(nameof(packetFormatterAdapter));
             _connection = connection ?? throw new ArgumentNullException(nameof(connection));
 
-            if (!(_connection is TcpConnection tcp) || tcp.IsConnected)
+            if (!(_connection is SocketConnection tcp) || tcp.IsConnected)
             {
                 _input = connection.Transport.Input;
                 _output = connection.Transport.Output;
@@ -109,7 +109,7 @@ namespace MQTTnet.AspNetCore
 
         public async Task ConnectAsync(CancellationToken cancellationToken)
         {
-            if (_connection is TcpConnection tcp && !tcp.IsConnected)
+            if (_connection is SocketConnection tcp && !tcp.IsConnected)
             {
                 await tcp.StartAsync().ConfigureAwait(false);
             }

Source/MQTTnet.Benchmarks/TcpPipesBenchmark.cs

@@ -29,12 +29,12 @@ namespace MQTTnet.Benchmarks
 
             var task = Task.Run(() => server.AcceptSocket());
 
-            var clientConnection = new TcpConnection(new IPEndPoint(IPAddress.Loopback, 1883));
+            var clientConnection = new SocketConnection(new IPEndPoint(IPAddress.Loopback, 1883));
 
             clientConnection.StartAsync().GetAwaiter().GetResult();
             _client = clientConnection.Transport;
 
-            var serverConnection = new TcpConnection(task.GetAwaiter().GetResult());
+            var serverConnection = new SocketConnection(task.GetAwaiter().GetResult());
             serverConnection.StartAsync().GetAwaiter().GetResult();
             _server = serverConnection.Transport;
         }

Source/MQTTnet.Extensions.ManagedClient/InterceptingPublishMessageEventArgs.cs

@@ -0,0 +1,20 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+
+namespace MQTTnet.Extensions.ManagedClient
+{
+    public sealed class InterceptingPublishMessageEventArgs : EventArgs
+    {
+        public InterceptingPublishMessageEventArgs(ManagedMqttApplicationMessage applicationMessage)
+        {
+            ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage));
+        }
+
+        public ManagedMqttApplicationMessage ApplicationMessage { get; }
+
+        public bool AcceptPublish { get; set; } = true;
+    }
+}

Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs

@@ -19,6 +19,7 @@ namespace MQTTnet.Extensions.ManagedClient
 {
     public sealed class ManagedMqttClient : Disposable, IManagedMqttClient
     {
+        readonly AsyncEvent<InterceptingPublishMessageEventArgs> _interceptingPublishMessageEvent = new AsyncEvent<InterceptingPublishMessageEventArgs>();
         readonly AsyncEvent<ApplicationMessageProcessedEventArgs> _applicationMessageProcessedEvent = new AsyncEvent<ApplicationMessageProcessedEventArgs>();
         readonly AsyncEvent<ApplicationMessageSkippedEventArgs> _applicationMessageSkippedEvent = new AsyncEvent<ApplicationMessageSkippedEventArgs>();
         readonly AsyncEvent<ConnectingFailedEventArgs> _connectingFailedEvent = new AsyncEvent<ConnectingFailedEventArgs>();
@@ -88,6 +89,14 @@ namespace MQTTnet.Extensions.ManagedClient
             remove => _applicationMessageProcessedEvent.RemoveHandler(value);
         }
 
+
+        public event Func<InterceptingPublishMessageEventArgs, Task> InterceptPublishMessageAsync
+        {
+            add => _interceptingPublishMessageEvent.AddHandler(value);
+            remove => _interceptingPublishMessageEvent.RemoveHandler(value);
+        }
+
+
         public event Func<MqttApplicationMessageReceivedEventArgs, Task> ApplicationMessageReceivedAsync
         {
             add => InternalClient.ApplicationMessageReceivedAsync += value;
@@ -700,9 +709,20 @@ namespace MQTTnet.Extensions.ManagedClient
         async Task TryPublishQueuedMessageAsync(ManagedMqttApplicationMessage message)
         {
             Exception transmitException = null;
+            bool acceptPublish = true;
             try
             {
-                await InternalClient.PublishAsync(message.ApplicationMessage).ConfigureAwait(false);
+                if (_interceptingPublishMessageEvent.HasHandlers)
+                {
+                    var interceptEventArgs = new InterceptingPublishMessageEventArgs(message);
+                    await _interceptingPublishMessageEvent.InvokeAsync(interceptEventArgs).ConfigureAwait(false);
+                    acceptPublish = interceptEventArgs.AcceptPublish;
+                }
+
+                if (acceptPublish)
+                {
+                    await InternalClient.PublishAsync(message.ApplicationMessage).ConfigureAwait(false);
+                }
 
                 using (await _messageQueueLock.EnterAsync().ConfigureAwait(false)) //lock to avoid conflict with this.PublishAsync
                 {

Source/MQTTnet.TestApp/ClientTest.cs

@@ -4,6 +4,7 @@
 
 using MQTTnet.Client;
 using System;
+using System.Net;
 using System.Text;
 using System.Threading.Tasks;
 using MQTTnet.Diagnostics;
@@ -27,7 +28,7 @@ namespace MQTTnet.TestApp
                 {
                     ChannelOptions = new MqttClientTcpOptions
                     {
-                        Server = "127.0.0.1"
+                        RemoteEndpoint = new DnsEndPoint("127.0.0.1", 0)
                     }
                 };
 

Source/MQTTnet.TestApp/MQTTnet.TestApp.csproj

@@ -10,6 +10,7 @@
         <EnforceCodeStyleInBuild>false</EnforceCodeStyleInBuild>
         <DisableImplicitAspNetCoreAnalyzers>true</DisableImplicitAspNetCoreAnalyzers>
         <NoWarn>1591;NETSDK1138</NoWarn>
+        <LangVersion>7.3</LangVersion>
     </PropertyGroup>
 
     <ItemGroup>

Source/MQTTnet.TestApp/ManagedClientTest.cs

@@ -7,6 +7,7 @@ using System.Threading.Tasks;
 using System.IO;
 using Newtonsoft.Json;
 using System.Collections.Generic;
+using System.Net;
 using MQTTnet.Client;
 using MQTTnet.Extensions.ManagedClient;
 using MQTTnet.Internal;
@@ -28,7 +29,7 @@ namespace MQTTnet.TestApp
                     Credentials = new RandomPassword(),
                     ChannelOptions = new MqttClientTcpOptions
                     {
-                        Server = "broker.hivemq.com"
+                        RemoteEndpoint = new DnsEndPoint("broker.hivemq.com", 0)
                     }
                 },
 

Source/MQTTnet.TestApp/PerformanceTest.cs

@@ -5,13 +5,13 @@
 using System;
 using System.Diagnostics;
 using System.Linq;
+using System.Net;
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Client;
 using MQTTnet.Protocol;
 using MQTTnet.Server;
-using MqttClient = MQTTnet.Client.MqttClient;
 
 namespace MQTTnet.TestApp
 {
@@ -25,7 +25,7 @@ namespace MQTTnet.TestApp
                 {
                     ChannelOptions = new MqttClientTcpOptions
                     {
-                        Server = "127.0.0.1"
+                        RemoteEndpoint = new DnsEndPoint("127.0.0.1", 0)
                     },
                     CleanSession = true
                 };
@@ -71,7 +71,7 @@ namespace MQTTnet.TestApp
                 {
                     ChannelOptions = new MqttClientTcpOptions
                     {
-                        Server = "127.0.0.1"
+                        RemoteEndpoint = new DnsEndPoint("127.0.0.1", 0)
                     }
                 };
 
@@ -114,7 +114,10 @@ namespace MQTTnet.TestApp
             {
                 var options = new MqttClientOptions
                 {
-                    ChannelOptions = new MqttClientTcpOptions { Server = "localhost" },
+                    ChannelOptions = new MqttClientTcpOptions
+                    {
+                        RemoteEndpoint = new DnsEndPoint("localhost", 0)
+                    },
                     ClientId = "Client1",
                     CleanSession = true
                 };
@@ -218,7 +221,7 @@ namespace MQTTnet.TestApp
                 {
                     ChannelOptions = new MqttClientTcpOptions
                     {
-                        Server = "127.0.0.1"
+                        RemoteEndpoint = new DnsEndPoint("127.0.0.1", 0)
                     },
                     CleanSession = true
                 };
@@ -269,7 +272,7 @@ namespace MQTTnet.TestApp
                 {
                     ChannelOptions = new MqttClientTcpOptions
                     {
-                        Server = "127.0.0.1"
+                        RemoteEndpoint = new DnsEndPoint("127.0.0.1", 0)
                     },
                     CleanSession = true
                 };
@@ -320,7 +323,7 @@ namespace MQTTnet.TestApp
                 {
                     ChannelOptions = new MqttClientTcpOptions
                     {
-                        Server = "127.0.0.1"
+                        RemoteEndpoint = new DnsEndPoint("127.0.0.1", 0)
                     },
                     CleanSession = true
                 };

Source/MQTTnet.TestApp/PublicBrokerTest.cs

@@ -42,45 +42,45 @@ namespace MQTTnet.TestApp
             };
 
             // mqtt.eclipseprojects.io
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "mqtt.eclipseprojects.io TCP",
                 new MqttClientOptionsBuilder().WithTcpServer("mqtt.eclipseprojects.io", 1883).WithProtocolVersion(MqttProtocolVersion.V311).Build());
 
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "mqtt.eclipseprojects.io WS",
                 new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("mqtt.eclipseprojects.io:80/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).Build());
 
 #if NET5_0_OR_GREATER
-            await ExecuteTestAsync("mqtt.eclipseprojects.io WS TLS13",
+            await ExecuteTestsAsync("mqtt.eclipseprojects.io WS TLS13",
                 new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("mqtt.eclipseprojects.io:443/mqtt"))
                     .WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls13).Build());
             
-            await ExecuteTestAsync("mqtt.eclipseprojects.io WS TLS13 (WebSocket4Net)",
+            await ExecuteTestsAsync("mqtt.eclipseprojects.io WS TLS13 (WebSocket4Net)",
                 new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("mqtt.eclipseprojects.io:443/mqtt"))
                     .WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls13).Build(),
                     true);
 #endif
 
             // test.mosquitto.org
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "test.mosquitto.org TCP",
                 new MqttClientOptionsBuilder().WithTcpServer("test.mosquitto.org", 1883).WithProtocolVersion(MqttProtocolVersion.V311).Build());
 
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "test.mosquitto.org TCP - Authenticated",
                 new MqttClientOptionsBuilder().WithTcpServer("test.mosquitto.org", 1884).WithCredentials("rw", "readwrite").WithProtocolVersion(MqttProtocolVersion.V311).Build());
 
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "test.mosquitto.org TCP TLS12",
                 new MqttClientOptionsBuilder().WithTcpServer("test.mosquitto.org", 8883).WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls12).Build());
 
 #if NET5_0_OR_GREATER
-            await ExecuteTestAsync("test.mosquitto.org TCP TLS13",
+            await ExecuteTestsAsync("test.mosquitto.org TCP TLS13",
                 new MqttClientOptionsBuilder().WithTcpServer("test.mosquitto.org", 8883)
                     .WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls13).Build());
 #endif
 
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "test.mosquitto.org TCP TLS12 - Authenticated",
                 new MqttClientOptionsBuilder().WithTcpServer("test.mosquitto.org", 8885)
                     .WithCredentials("rw", "readwrite")
@@ -88,67 +88,67 @@ namespace MQTTnet.TestApp
                     .WithTlsOptions(unsafeTls12)
                     .Build());
 
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "test.mosquitto.org WS",
                 new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("test.mosquitto.org:8080/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).Build());
 
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "test.mosquitto.org WS (WebSocket4Net)",
                 new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("test.mosquitto.org:8080/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).Build(),
                 true);
 
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "test.mosquitto.org WS TLS12",
                 new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("test.mosquitto.org:8081/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls12).Build());
 
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "test.mosquitto.org WS TLS12 (WebSocket4Net)",
                 new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("test.mosquitto.org:8081/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls12).Build(),
                 true);
 
             // broker.emqx.io
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "broker.emqx.io TCP",
                 new MqttClientOptionsBuilder().WithTcpServer("broker.emqx.io", 1883).WithProtocolVersion(MqttProtocolVersion.V311).Build());
 
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "broker.emqx.io TCP TLS12",
                 new MqttClientOptionsBuilder().WithTcpServer("broker.emqx.io", 8883).WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls12).Build());
 
 #if NET5_0_OR_GREATER
-            await ExecuteTestAsync("broker.emqx.io TCP TLS13",
+            await ExecuteTestsAsync("broker.emqx.io TCP TLS13",
                 new MqttClientOptionsBuilder().WithTcpServer("broker.emqx.io", 8883)
                     .WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls13).Build());
 #endif
 
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "broker.emqx.io WS",
                 new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("broker.emqx.io:8083/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).Build());
 
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "broker.emqx.io WS (WebSocket4Net)",
                 new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("broker.emqx.io:8084/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).Build(),
                 true);
 
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "broker.emqx.io WS TLS12",
                 new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("broker.emqx.io:8084/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls12).Build());
 
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "broker.emqx.io WS TLS12 (WebSocket4Net)",
                 new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("broker.emqx.io:8084/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).WithTlsOptions(unsafeTls12).Build(),
                 true);
 
             // broker.hivemq.com
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "broker.hivemq.com TCP",
                 new MqttClientOptionsBuilder().WithTcpServer("broker.hivemq.com", 1883).WithProtocolVersion(MqttProtocolVersion.V311).Build());
 
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "broker.hivemq.com WS",
                 new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("broker.hivemq.com:8000/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).Build());
 
-            await ExecuteTestAsync(
+            await ExecuteTestsAsync(
                 "broker.hivemq.com WS (WebSocket4Net)",
                 new MqttClientOptionsBuilder().WithWebSocketServer(o => o.WithUri("broker.hivemq.com:8000/mqtt")).WithProtocolVersion(MqttProtocolVersion.V311).Build(),
                 true);
@@ -160,6 +160,15 @@ namespace MQTTnet.TestApp
             Console.ReadLine();
         }
 
+        static async Task ExecuteTestsAsync(string name, MqttClientOptions options, bool useWebSocket4Net = false)
+        {
+            options.ProtocolVersion = MqttProtocolVersion.V311;
+            await ExecuteTestAsync(name + " V3.1.1", options, useWebSocket4Net);
+
+            options.ProtocolVersion = MqttProtocolVersion.V500;
+            await ExecuteTestAsync(name + " V5.0.0", options, useWebSocket4Net);
+        }
+        
         static async Task ExecuteTestAsync(string name, MqttClientOptions options, bool useWebSocket4Net = false)
         {
             try

Source/MQTTnet.Tests/Clients/ManagedMqttClient/ManagedMqttClient_Tests.cs

@@ -437,6 +437,109 @@ namespace MQTTnet.Tests.Clients.ManagedMqttClient
             }
         }
 
+
+        [TestMethod]
+        public async Task ManagedClients_CanInterceptPublishedMessage_PreventingPublish()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                await testEnvironment.StartServer();
+
+                var receivedMessagesCount = 0;
+                var interceptedMessagesCount = 0;
+
+                var clientOptions = new MqttClientOptionsBuilder()
+                    .WithTcpServer("127.0.0.1", testEnvironment.ServerPort)
+                    .Build();
+
+                var sendingClient = testEnvironment.CreateClient();
+                var sendingManagedClient = new MQTTnet.Extensions.ManagedClient.ManagedMqttClient(sendingClient, testEnvironment.ClientLogger);
+
+                await sendingManagedClient.StartAsync(new ManagedMqttClientOptionsBuilder().WithClientOptions(clientOptions).Build());
+
+                // Wait until the managed client is fully set up and running.
+                await Task.Delay(1000);
+
+                var receivingClient = await testEnvironment.ConnectClient();
+
+                receivingClient.ApplicationMessageReceivedAsync += e =>
+                {
+                    Interlocked.Increment(ref receivedMessagesCount);
+                    return CompletedTask.Instance;
+                };
+                sendingManagedClient.InterceptPublishMessageAsync += e =>
+                {
+                    Interlocked.Increment(ref interceptedMessagesCount);
+                    e.AcceptPublish = false;
+                    return CompletedTask.Instance;
+                };
+
+                await receivingClient.SubscribeAsync("My/last/will");
+
+                // Wait for arrival of the will message at the receiver.
+                await Task.Delay(5000);
+
+                await sendingManagedClient.EnqueueAsync("My/last/will", "hello world");
+
+                // Wait for arrival of the will message at the receiver.
+                await Task.Delay(5000);
+
+                Assert.AreEqual(0, receivedMessagesCount);
+                Assert.AreEqual(1, interceptedMessagesCount);
+            }
+        }
+
+        [TestMethod]
+        public async Task ManagedClients_CanInterceptPublishedMessage_AllowingPublish()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                await testEnvironment.StartServer();
+
+                var receivedMessagesCount = 0;
+                var interceptedMessagesCount = 0;
+
+                var clientOptions = new MqttClientOptionsBuilder()
+                    .WithTcpServer("127.0.0.1", testEnvironment.ServerPort)
+                    .Build();
+
+                var sendingClient = testEnvironment.CreateClient();
+                var sendingManagedClient = new MQTTnet.Extensions.ManagedClient.ManagedMqttClient(sendingClient, testEnvironment.ClientLogger);
+
+                await sendingManagedClient.StartAsync(new ManagedMqttClientOptionsBuilder().WithClientOptions(clientOptions).Build());
+
+                // Wait until the managed client is fully set up and running.
+                await Task.Delay(1000);
+
+                var receivingClient = await testEnvironment.ConnectClient();
+
+                receivingClient.ApplicationMessageReceivedAsync += e =>
+                {
+                    Interlocked.Increment(ref receivedMessagesCount);
+                    return CompletedTask.Instance;
+                };
+                sendingManagedClient.InterceptPublishMessageAsync += e =>
+                {
+                    Interlocked.Increment(ref interceptedMessagesCount);
+                    e.AcceptPublish = true;
+                    return CompletedTask.Instance;
+                };
+
+                await receivingClient.SubscribeAsync("My/last/will");
+
+                // Wait for arrival of the will message at the receiver.
+                await Task.Delay(5000);
+
+                await sendingManagedClient.EnqueueAsync("My/last/will", "hello world");
+
+                // Wait for arrival of the will message at the receiver.
+                await Task.Delay(5000);
+
+                Assert.AreEqual(1, receivedMessagesCount);
+                Assert.AreEqual(1, interceptedMessagesCount);
+            }
+        }
+
         // This case also serves as a regression test for the previous behavior which re-published
         // each and every existing subscriptions with every new subscription that was made
         // (causing performance problems and having the visible symptom of retained messages being received again)

Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs

@@ -7,6 +7,7 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Net;
 using System.Text;
 using MQTTnet.Formatter;
 using MQTTnet.Packets;
@@ -17,8 +18,10 @@ namespace MQTTnet.Client
     public sealed class MqttClientOptionsBuilder
     {
         readonly MqttClientOptions _options = new MqttClientOptions();
+        int? _port;
 
         [Obsolete] MqttClientWebSocketProxyOptions _proxyOptions;
+        EndPoint _remoteEndPoint;
 
         MqttClientTcpOptions _tcpOptions;
         MqttClientTlsOptions _tlsOptions;
@@ -63,6 +66,31 @@ namespace MQTTnet.Client
             if (_tcpOptions != null)
             {
                 _tcpOptions.TlsOptions = tlsOptions;
+
+                if (_remoteEndPoint == null)
+                {
+                    throw new ArgumentException("No endpoint is set.");
+                }
+
+                if (_remoteEndPoint is DnsEndPoint dns)
+                {
+                    if (dns.Port == 0)
+                    {
+                        if (_port.HasValue)
+                        {
+                            _remoteEndPoint = new DnsEndPoint(dns.Host, _port.Value);
+                        }
+                        else
+                        {
+                            _remoteEndPoint = new DnsEndPoint(dns.Host, tlsOptions?.UseTls == false ? MqttPorts.Default : MqttPorts.Secure);
+                        }
+                    }
+                }
+
+                if (_tcpOptions.RemoteEndpoint == null)
+                {
+                    _tcpOptions.RemoteEndpoint = _remoteEndPoint;
+                }
             }
             else if (_webSocketOptions != null)
             {
@@ -117,7 +145,6 @@ namespace MQTTnet.Client
             return this;
         }
 
-        [Obsolete("Use WithTcpServer(... configure) or WithWebSocketServer(... configure) instead.")]
         public MqttClientOptionsBuilder WithConnectionUri(Uri uri)
         {
             if (uri == null)
@@ -130,24 +157,33 @@ namespace MQTTnet.Client
             {
                 case "tcp":
                 case "mqtt":
+                {
                     WithTcpServer(uri.Host, port);
                     break;
+                }
 
                 case "mqtts":
+                {
                     WithTcpServer(uri.Host, port)
                         .WithTlsOptions(
                             o =>
                             {
+                                o.UseTls();
                             });
                     break;
+                }
 
                 case "ws":
                 case "wss":
+                {
                     WithWebSocketServer(o => o.WithUri(uri.ToString()));
                     break;
+                }
 
                 default:
-                    throw new ArgumentException("Unexpected scheme in uri.");
+                {
+                    throw new ArgumentException("Unexpected scheme in URI.");
+                }
             }
 
             if (!string.IsNullOrEmpty(uri.UserInfo))
@@ -161,7 +197,6 @@ namespace MQTTnet.Client
             return this;
         }
 
-        [Obsolete("Use WithTcpServer(... configure) or WithWebSocketServer(... configure) instead.")]
         public MqttClientOptionsBuilder WithConnectionUri(string uri)
         {
             return WithConnectionUri(new Uri(uri, UriKind.Absolute));
@@ -190,6 +225,14 @@ namespace MQTTnet.Client
             return this;
         }
 
+        public MqttClientOptionsBuilder WithEndPoint(EndPoint endPoint)
+        {
+            _remoteEndPoint = endPoint ?? throw new ArgumentNullException(nameof(endPoint));
+            _tcpOptions = new MqttClientTcpOptions();
+
+            return this;
+        }
+
         public MqttClientOptionsBuilder WithExtendedAuthenticationExchangeHandler(IMqttExtendedAuthenticationExchangeHandler handler)
         {
             _options.ExtendedAuthenticationExchangeHandler = handler;
@@ -307,11 +350,12 @@ namespace MQTTnet.Client
 
         public MqttClientOptionsBuilder WithTcpServer(string server, int? port = null)
         {
-            _tcpOptions = new MqttClientTcpOptions
+            _tcpOptions = new MqttClientTcpOptions();
-            {
+
-                Server = server,
+            // The value 0 will be updated when building the options.
-                Port = port
+            // This a backward compatibility feature.
-            };
+            _remoteEndPoint = new DnsEndPoint(server, port ?? 0);
+            _port = port;
 
             return this;
         }

Source/MQTTnet/Client/Options/MqttClientTcpOptions.cs

@@ -11,34 +11,32 @@ namespace MQTTnet.Client
     {
         public AddressFamily AddressFamily { get; set; } = AddressFamily.Unspecified;
 
-        /// <summary>
-        /// Gets the local endpoint (network card) which is used by the client.
-        /// Set it to _null_ to let the OS select the network card.
-        /// </summary>
-        public EndPoint LocalEndpoint { get; set; }
-        
         public int BufferSize { get; set; } = 8192;
 
         /// <summary>
-        /// Gets or sets whether the underlying socket should run in dual mode.
+        ///     Gets or sets whether the underlying socket should run in dual mode.
-        /// Leaving this _null_ will avoid setting this value at socket level.
+        ///     Leaving this _null_ will avoid setting this value at socket level.
-        /// Setting this a value other than _null_ will throw an exception when only IPv4 is supported on the machine.
+        ///     Setting this a value other than _null_ will throw an exception when only IPv4 is supported on the machine.
         /// </summary>
         public bool? DualMode { get; set; }
 
         public LingerOption LingerState { get; set; } = new LingerOption(true, 0);
 
+        /// <summary>
+        ///     Gets the local endpoint (network card) which is used by the client.
+        ///     Set it to _null_ to let the OS select the network card.
+        /// </summary>
+        public EndPoint LocalEndpoint { get; set; }
+
         public bool NoDelay { get; set; } = true;
 
-        public int? Port { get; set; }
+        public EndPoint RemoteEndpoint { get; set; }
-        
-        public string Server { get; set; }
 
         public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions();
 
         public override string ToString()
         {
-            return Server + ":" + this.GetPort();
+            return RemoteEndpoint?.ToString() ?? string.Empty;
         }
     }
 }

Source/MQTTnet/Client/Options/MqttClientTcpOptionsExtensions.cs

@@ -1,23 +0,0 @@
-// Licensed to the .NET Foundation under one or more agreements.
-// The .NET Foundation licenses this file to you under the MIT license.
-// See the LICENSE file in the project root for more information.
-
-using System;
-
-namespace MQTTnet.Client
-{
-    public static class MqttClientTcpOptionsExtensions
-    {
-        public static int GetPort(this MqttClientTcpOptions options)
-        {
-            if (options == null) throw new ArgumentNullException(nameof(options));
-
-            if (options.Port.HasValue)
-            {
-                return options.Port.Value;
-            }
-
-            return !(options.TlsOptions?.UseTls ?? false) ? 1883 : 8883;
-        }
-    }
-}

Source/MQTTnet/Implementations/CrossPlatformSocket.cs

@@ -170,11 +170,16 @@ namespace MQTTnet.Implementations
             _socket.Bind(localEndPoint);
         }
 
-        public async Task ConnectAsync(string host, int port, CancellationToken cancellationToken)
+        public Task ConnectAsync(string host, int port, CancellationToken cancellationToken)
         {
-            if (host is null)
+            return ConnectAsync(new DnsEndPoint(host, port), cancellationToken);
+        }
+
+        public async Task ConnectAsync(EndPoint endPoint, CancellationToken cancellationToken)
+        {
+            if (endPoint is null)
             {
-                throw new ArgumentNullException(nameof(host));
+                throw new ArgumentNullException(nameof(endPoint));
             }
 
             cancellationToken.ThrowIfCancellationRequested();
@@ -191,15 +196,15 @@ namespace MQTTnet.Implementations
 #endif
 
 #if NET5_0_OR_GREATER
-                await _socket.ConnectAsync(host, port, cancellationToken).ConfigureAwait(false);
+                await _socket.ConnectAsync(endPoint, cancellationToken).ConfigureAwait(false);
 #else
                 // Workaround for: https://github.com/dotnet/corefx/issues/24430
                 using (cancellationToken.Register(_socketDisposeAction))
                 {
 #if NET452 || NET461
-                    await Task.Factory.FromAsync(_socket.BeginConnect, _socket.EndConnect, host, port, null).ConfigureAwait(false);
+                    await Task.Factory.FromAsync(_socket.BeginConnect, _socket.EndConnect, endPoint, null).ConfigureAwait(false);
 #else
-                    await _socket.ConnectAsync(host, port).ConfigureAwait(false);
+                    await _socket.ConnectAsync(endPoint).ConfigureAwait(false);
 #endif
                 }
 #endif
@@ -217,7 +222,7 @@ namespace MQTTnet.Implementations
                     throw new MqttCommunicationTimedOutException();
                 }
 
-                throw new MqttCommunicationException($"Error while connecting with host '{host}:{port}'.", socketException);
+                throw new MqttCommunicationException($"Error while connecting host '{endPoint}'.", socketException);
             }
             catch (ObjectDisposedException)
             {

Source/MQTTnet/Implementations/MqttTcpChannel.Uwp.cs

@@ -17,6 +17,7 @@ using MQTTnet.Channel;
 using MQTTnet.Client;
 using MQTTnet.Server;
 using System.Runtime.InteropServices.WindowsRuntime;
+using System.Net;
 
 namespace MQTTnet.Implementations
 {
@@ -65,9 +66,22 @@ namespace MQTTnet.Implementations
                 _socket.Control.KeepAlive = true;
             }
 
+            string hostName;
+            string serviceName;
+
+            if (_options.RemoteEndpoint is DnsEndPoint dns)
+            {
+                hostName = dns.Host;
+                serviceName = dns.Port.ToString();
+            }
+            else
+            {
+                throw new NotSupportedException("UWP only supports a DNS endpoint.");
+            }
+
             if (_options.TlsOptions?.UseTls != true)
             {
-                await _socket.ConnectAsync(new HostName(_options.Server), _options.GetPort().ToString()).AsTask().ConfigureAwait(false);
+                await _socket.ConnectAsync(new HostName(hostName), serviceName).AsTask().ConfigureAwait(false);
             }
             else
             {
@@ -88,7 +102,7 @@ namespace MQTTnet.Implementations
                     socketProtectionLevel = SocketProtectionLevel.Tls10;
                 }
 
-                await _socket.ConnectAsync(new HostName(_options.Server), _options.GetPort().ToString(), socketProtectionLevel).AsTask().ConfigureAwait(false);
+                await _socket.ConnectAsync(new HostName(hostName), serviceName, socketProtectionLevel).AsTask().ConfigureAwait(false);
             }
 
             Endpoint = _socket.Information.RemoteAddress + ":" + _socket.Information.RemotePort;

Source/MQTTnet/Implementations/MqttTcpChannel.cs

@@ -5,6 +5,7 @@
 #if !WINDOWS_UWP
 using System;
 using System.IO;
+using System.Net;
 using System.Net.Security;
 using System.Net.Sockets;
 using System.Runtime.ExceptionServices;
@@ -92,7 +93,7 @@ namespace MQTTnet.Implementations
                     socket.DualMode = _tcpOptions.DualMode.Value;
                 }
 
-                await socket.ConnectAsync(_tcpOptions.Server, _tcpOptions.GetPort(), cancellationToken).ConfigureAwait(false);
+                await socket.ConnectAsync(_tcpOptions.RemoteEndpoint, cancellationToken).ConfigureAwait(false);
 
                 cancellationToken.ThrowIfCancellationRequested();
 
@@ -103,9 +104,11 @@ namespace MQTTnet.Implementations
                     var targetHost = _tcpOptions.TlsOptions.TargetHost;
                     if (string.IsNullOrEmpty(targetHost))
                     {
-                        targetHost = _tcpOptions.Server;
+                        if (_tcpOptions.RemoteEndpoint is DnsEndPoint dns)
+                        {
+                            targetHost = dns.Host;    
+                        }
                     }
-
                     var sslStream = new SslStream(networkStream, false, InternalUserCertificateValidationCallback);
                     try
                     {

Source/MQTTnet/Protocol/MqttPorts.cs

@@ -0,0 +1,13 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+namespace MQTTnet.Protocol
+{
+    public static class MqttPorts
+    {
+        public const int Default = 1883;
+
+        public const int Secure = 8883;
+    }
+}

Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs

@@ -57,7 +57,7 @@ namespace MQTTnet.Server
                 while (!cancellationToken.IsCancellationRequested)
                 {
                     TryProcessClients();
-                    Sleep(_options.KeepAliveMonitorInterval);
+                    Sleep(_options.KeepAliveOptions.MonitorInterval);
                 }
             }
             catch (OperationCanceledException)