增加扩展

BPA.MQTTnet.Extensions/BPA - Backup.MQTTnet.Extensions.csproj

@@ -0,0 +1,18 @@
+<Project Sdk="Microsoft.NET.Sdk">
+  <PropertyGroup>
+    <TargetFramework>netcoreapp3.1</TargetFramework>
+    <VersionPrefix>1.0.2</VersionPrefix>
+    <PackageReleaseNotes />
+  </PropertyGroup>
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Extensions.Configuration.Binder" Version="6.0.0" />
+    <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="3.1.0" />
+    <PackageReference Include="Nuget.Tools.V2" Version="1.1.6">
+      <PrivateAssets>all</PrivateAssets>
+      <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+    </PackageReference>
+  </ItemGroup>
+  <ItemGroup>
+    <ProjectReference Include="..\Source\MQTTnet\MQTTnet.csproj" />
+  </ItemGroup>
+</Project>

BPA.MQTTnet.Extensions/BPA.MQTTnet.Extensions.csproj

@@ -0,0 +1,16 @@
+<Project Sdk="Microsoft.NET.Sdk">
+  <PropertyGroup>
+    <TargetFramework>netcoreapp3.1</TargetFramework>
+    <VersionPrefix>1.0.3</VersionPrefix>
+    <PackageReleaseNotes />
+  </PropertyGroup>
+  <ItemGroup>
+    <PackageReference Include="BPA.MQTTnet" Version="1.0.3" />
+    <PackageReference Include="Microsoft.Extensions.Configuration.Binder" Version="6.0.0" />
+    <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="3.1.0" />
+    <PackageReference Include="Nuget.Tools.V2" Version="1.1.6">
+      <PrivateAssets>all</PrivateAssets>
+      <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+    </PackageReference>
+  </ItemGroup>
+</Project>

BPA.MQTTnet.Extensions/Extensions/ServiceCollectionExtension.cs

@@ -0,0 +1,104 @@
+
+using BPA.MQTTnet.Extensions.Settings;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.DependencyInjection;
+using MQTTnet;
+using MQTTnet.Client;
+using MQTTnet.Client.Connecting;
+using MQTTnet.Client.Disconnecting;
+using MQTTnet.Client.Options;
+using System;
+
+using System.Threading.Tasks;
+
+namespace BPA.MQTTnet.Extensions.Extensions
+{
+    public static class ServiceCollectionExtension
+    {
+
+        public static IServiceCollection AddMqttClientHostedService(this IServiceCollection services, IConfiguration configuration)
+        {
+            ClientSettings clientSettings = new ClientSettings();
+            configuration.GetSection(nameof(ClientSettings)).Bind(clientSettings);
+            services.AddSingleton(clientSettings);
+
+            BrokerHostSettings brokerHostSettings = new BrokerHostSettings();
+            configuration.GetSection(nameof(BrokerHostSettings)).Bind(brokerHostSettings);
+            services.AddSingleton(brokerHostSettings);
+            services.AddMqttClientServiceWithConfig();
+            //services.AddMqttClientServiceWithConfig(aspOptionBuilder =>
+            //{
+               
+            //    aspOptionBuilder
+            //    .WithCredentials(clientSettings.UserName, clientSettings.Password)
+            //    .WithClientId(clientSettings.Id)
+            //    .WithTcpServer(brokerHostSettings.Host, brokerHostSettings.Port);
+            //});
+            return services;
+        }
+
+        private static IServiceCollection AddMqttClientServiceWithConfig(this IServiceCollection services)
+        {
+            var clientSettinigs = services.BuildServiceProvider().GetService<ClientSettings>();
+            var brokerHostSettings = services.BuildServiceProvider().GetService<BrokerHostSettings>();
+            //services.AddSingleton(serviceProvider =>
+            //{
+            //    var optionBuilder = new AspCoreMqttClientOptionBuilder(serviceProvider);
+            //    configure(new MqttClientOptionsBuilder());
+            //    return optionBuilder.Build();
+            //});
+            services.AddSingleton(serviceProvider =>
+            {
+                var clientOptions = new MqttClientOptionsBuilder()
+                 .WithTcpServer(brokerHostSettings.Host, brokerHostSettings.Port)
+                 .WithCredentials(clientSettinigs.UserName, clientSettinigs.Password)
+                 .WithClientId(clientSettinigs.Id)
+                 .Build();
+                return clientOptions;
+            });
+            services.AddSingleton(serviceProvider =>
+            {
+                var factory = new MqttFactory();
+                var client = factory.CreateMqttClient();
+                client.DisconnectedHandler = new MqttClientDisconnectedHandlerDelegate(async e =>
+                {
+                    Console.WriteLine("### MQTT 从服务端断开连接 ###");
+                    await Task.Delay(TimeSpan.FromSeconds(5));
+
+                    try
+                    {
+                        await client.ConnectAsync(serviceProvider.GetService<IMqttClientOptions>());
+                    }
+                    catch
+                    {
+                        Console.WriteLine("### MQTT 重连失败 ###");
+                    }
+                });
+                client.ConnectedHandler = new MqttClientConnectedHandlerDelegate(async e =>
+                {
+                    Console.WriteLine("###  MQTT 连接服务端  ###");
+
+                    await client.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build());
+
+                    Console.WriteLine("### MQTT 连接成功 ###");
+                });
+                Task.WaitAny(client.ConnectAsync(serviceProvider.GetService<IMqttClientOptions>()));
+                return client;
+            });
+            //service
+            //
+            //s.AddSingleton<MqttClientService>();
+            //services.AddSingleton<IHostedService>(serviceProvider =>
+            //{
+            //    return serviceProvider.GetService<MqttClientService>();
+            //});
+            //services.AddSingleton<MqttClientServiceProvider>(serviceProvider =>
+            //{
+            //    var mqttClientService = serviceProvider.GetService<MqttClientService>();
+            //    var mqttClientServiceProvider = new MqttClientServiceProvider(mqttClientService);
+            //    return mqttClientServiceProvider;
+            //});
+            return services;
+        }
+    }
+}

BPA.MQTTnet.Extensions/Settings/BrokerHostSettings.cs

@@ -0,0 +1,13 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+
+namespace BPA.MQTTnet.Extensions.Settings
+{
+    public class BrokerHostSettings
+    {
+        public string Host { set; get; }
+        public int Port { set; get; }
+    }
+}

BPA.MQTTnet.Extensions/Settings/ClientSettings.cs

@@ -0,0 +1,14 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+
+namespace BPA.MQTTnet.Extensions.Settings
+{
+    public class ClientSettings
+    {
+        public string Id { set; get; }
+        public string UserName { set; get; }
+        public string Password { set; get; }
+    }
+}

MQTTnet.client.Test/Extensions/ServiceCollectionExtension.cs

@@ -0,0 +1,83 @@
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.DependencyInjection;
+using MQTTnet.client.Test.Options;
+using MQTTnet.client.Test.Settings;
+using MQTTnet.Client;
+using MQTTnet.Client.Connecting;
+using MQTTnet.Client.Disconnecting;
+using MQTTnet.Client.Options;
+using System;
+using System.Threading.Tasks;
+
+namespace MQTTnet.client.Test.Extensions
+{
+    public static class ServiceCollectionExtension
+    {
+
+        public static IServiceCollection AddMqttClientHostedService(this IServiceCollection services, IConfiguration configuration)
+        {
+            ClientSettings clientSettings = new ClientSettings();
+            configuration.GetSection(nameof(ClientSettings)).Bind(clientSettings);
+            services.AddSingleton(clientSettings);
+
+            BrokerHostSettings brokerHostSettings = new BrokerHostSettings();
+            configuration.GetSection(nameof(BrokerHostSettings)).Bind(brokerHostSettings);
+            services.AddSingleton(brokerHostSettings);
+            services.AddMqttClientServiceWithConfig( );
+            return services;
+        }
+
+        private static IServiceCollection AddMqttClientServiceWithConfig(this IServiceCollection services)
+        {
+
+            var clientSettinigs = services.BuildServiceProvider().GetService<ClientSettings>();
+            var brokerHostSettings = services.BuildServiceProvider().GetService<BrokerHostSettings>();
+            //services.AddSingleton(serviceProvider =>
+            //{
+            //    var optionBuilder = new AspCoreMqttClientOptionBuilder(serviceProvider);
+            //    configure(new MqttClientOptionsBuilder());
+            //    return optionBuilder.Build();
+            //});
+            services.AddSingleton(serviceProvider =>
+            {
+                var clientOptions = new MqttClientOptionsBuilder()
+                 .WithTcpServer(brokerHostSettings.Host, brokerHostSettings.Port)
+                 .WithCredentials(clientSettinigs.UserName, clientSettinigs.Password)
+                 .WithClientId(clientSettinigs.Id)
+                 .Build();
+                return clientOptions;
+            });
+            services.AddSingleton(serviceProvider =>
+            {
+                var factory = new MqttFactory();
+                var client = factory.CreateMqttClient();
+                client.DisconnectedHandler = new MqttClientDisconnectedHandlerDelegate(async e =>
+                {
+                    Console.WriteLine("### MQTT 从服务端断开连接 ###");
+                    await Task.Delay(TimeSpan.FromSeconds(5));
+
+                    try
+                    {
+                        await client.ConnectAsync(serviceProvider.GetService<IMqttClientOptions>());
+                    }
+                    catch
+                    {
+                        Console.WriteLine("### MQTT 重连失败 ###");
+                    }
+                });
+                client.ConnectedHandler = new MqttClientConnectedHandlerDelegate(async e =>
+                {
+                    Console.WriteLine("###  MQTT 连接服务端  ###");
+
+                    await client.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build());
+
+                    Console.WriteLine("### MQTT 连接成功 ###");
+                });
+                Task.WaitAny(client.ConnectAsync(serviceProvider.GetService<IMqttClientOptions>()));
+                return client;
+            });
+           
+            return services;
+        }
+    }
+}

MQTTnet.client.Test/MQTTnet.client.Test.csproj

@@ -0,0 +1,11 @@
+<Project Sdk="Microsoft.NET.Sdk.Web">
+
+  <PropertyGroup>
+    <TargetFramework>netcoreapp3.1</TargetFramework>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\Source\MQTTnet\MQTTnet.csproj" />
+  </ItemGroup>
+
+</Project>

MQTTnet.client.Test/MqttClient.cs

@@ -0,0 +1,433 @@
+using MQTTnet;
+using MQTTnet.Client;
+using MQTTnet.Client.Disconnecting;
+using MQTTnet.Client.Options;
+using MQTTnet.Protocol;
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace MQTT
+{
+    public class MqttClient
+    {
+        #region Field Area
+
+        /// <summary>
+        /// Mqtt factory
+        /// </summary>
+        private MqttFactory _factory;
+        /// <summary>
+        /// Mqtt client
+        /// </summary>
+        private IMqttClient _mqttClient;
+        /// <summary>
+        /// Mqtt 配置信息
+        /// </summary>
+        private MqttClientConfig _mqttClientConfig;
+        /// <summary>
+        /// Mqtt options
+        /// </summary>
+        private IMqttClientOptions _options;
+
+        #endregion
+
+        #region CTOR
+
+        /// <summary>
+        /// 默认启动IP为127.0.0.1 端口为1883
+        /// </summary>
+        public MqttClient()
+        {
+            _mqttClientConfig = new MqttClientConfig
+            {
+                ServerIp = "10.2.1.21",
+                Port = 1883
+            };
+            Init();
+        }
+        /// <summary>
+        /// 调用示例
+        /// 
+        /// //var client = new MqttClient(s =>
+        ///    {
+        ///        s.Port = 1883;
+        ///        s.ServerIp = "127.0.0.1";
+        ///        s.UserName = "mqtt-test";
+        ///        s.Password = "mqtt-test";
+        ///        s.ReciveMsgCallback = s =>
+        ///        {
+        ///            Console.WriteLine(s.Payload_UTF8);
+        ///        };
+        ///    }, true);
+        ///    client.Subscribe("/TopicName/");
+        /// </summary>
+        /// <param name="config">客户端配置信息</param>
+        /// <param name="autoStart">直接启动</param>
+        public MqttClient(Action<MqttClientConfig> config, bool autoStart = false)
+        {
+            _mqttClientConfig = new MqttClientConfig();
+            config(_mqttClientConfig);
+            Init();
+            if (autoStart)
+            {
+                Task.WaitAll( Start());
+            }
+        }
+
+        #endregion
+
+        /// <summary>
+        /// 获取MqttClient实例
+        /// 
+        /// 调用示例
+        ///    //var client = MqttClient.Instance(s =>
+        ///    {
+        ///        s.Port = 1883;
+        ///        s.ServerIp = "127.0.0.1";
+        ///        s.UserName = "mqtt-test";
+        ///        s.Password = "mqtt-test";
+        ///        s.ReciveMsgCallback = s =>
+        ///        {
+        ///            Console.WriteLine(s.Payload_UTF8);
+        ///        };
+        ///    }, true);
+        ///    client.Subscribe("/TopicName/");
+        /// </summary>
+        /// <param name="config">客户端配置信息</param>
+        /// <param name="autoStart">直接启动</param>
+        /// <returns></returns>
+        public static MqttClient Instance(Action<MqttClientConfig> config, bool autoStart = false)
+            => new MqttClient(config, autoStart);
+
+        /// <summary>
+        /// 初始化注册
+        /// </summary>
+        private void Init()
+        {
+            try
+            {
+                _factory = new MqttFactory();
+
+                _mqttClient = _factory.CreateMqttClient();
+
+                _options = new MqttClientOptionsBuilder()
+                    .WithTcpServer(_mqttClientConfig.ServerIp, _mqttClientConfig.Port)
+                    .WithCredentials(_mqttClientConfig.UserName, _mqttClientConfig.Password)
+                    .WithClientId(_mqttClientConfig.ClientId)
+                    .Build();
+
+                //消息回调
+                _mqttClient.UseApplicationMessageReceivedHandler(ReciveMsg);
+            }
+            catch (Exception exp)
+            {
+                if (_mqttClientConfig.Exception is null)
+                {
+                    throw exp;
+                }
+                _mqttClientConfig.Exception(exp);
+            }
+        }
+
+        #region 内部事件转换处理
+
+        /// <summary>
+        /// 消息接收回调
+        /// </summary>
+        /// <param name="e"></param>
+        /// <returns></returns>
+        private void ReciveMsg(MqttApplicationMessageReceivedEventArgs e)
+        {
+            if (_mqttClientConfig.ReciveMsgCallback != null)
+            {
+                _mqttClientConfig.ReciveMsgCallback(new MqttClientReciveMsg
+                {
+                    Topic = e.ApplicationMessage.Topic,
+                    Payload_UTF8 = Encoding.UTF8.GetString(e.ApplicationMessage.Payload),
+                    Payload = e.ApplicationMessage.Payload,
+                    Qos = e.ApplicationMessage.QualityOfServiceLevel,
+                    Retain = e.ApplicationMessage.Retain,
+                });
+            }
+        }
+
+        /// <summary>
+        /// 订阅
+        /// </summary>
+        /// <param name="topicName"></param>
+        public async void Subscribe(string topicName)
+        {
+            topicName = topicName.Trim();
+            if (string.IsNullOrEmpty(topicName))
+            {
+                throw new Exception("订阅主题不能为空！");
+            }
+
+            if (!_mqttClient.IsConnected)
+            {
+                throw new Exception("MQTT客户端尚未连接！请先启动连接");
+            }
+            await _mqttClient.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic(topicName).Build());
+        }
+
+        /// <summary>
+        /// 取消订阅
+        /// </summary>
+        /// <param name="topicName"></param>
+        public async void Unsubscribe(string topicName)
+        {
+            topicName = topicName.Trim();
+            if (string.IsNullOrEmpty(topicName))
+            {
+                throw new Exception("订阅主题不能为空！");
+            }
+
+            if (!_mqttClient.IsConnected)
+            {
+                throw new Exception("MQTT客户端尚未连接！请先启动连接");
+            }
+            await _mqttClient.UnsubscribeAsync(topicName);
+        }
+
+        /// <summary>
+        /// 重连机制
+        /// </summary>
+        private void ReConnected()
+        {
+            if (_mqttClient.IsConnected)
+            {
+                return;
+            }
+            for (int i = 0; i < 10; i++)
+            {
+                //重连机制
+                _mqttClient.UseDisconnectedHandler(async e =>
+                {
+                    try
+                    {
+                        await Task.Delay(TimeSpan.FromSeconds(_mqttClientConfig.ReconneTime));
+                        await _mqttClient.ConnectAsync(_options);
+                        return;
+                    }
+                    catch (Exception exp)
+                    {
+                        if (_mqttClientConfig.Exception is null)
+                        {
+                            throw exp;
+                        }
+                        _mqttClientConfig.Exception(exp);
+                    }
+                });
+            }
+        }
+
+        #endregion
+
+        #region 消息发送
+
+        /// <summary>
+        /// 发送消息,含有重连机制，如掉线会自动重连
+        /// </summary>
+        /// <param name="message"></param>
+        private async Task PublishAsync(string topicName, MqttApplicationMessageBuilder MessageBuilder, PublicQos qos = 0)
+        {
+            string topic = topicName.Trim();
+            if (string.IsNullOrEmpty(topic))
+            {
+                throw new Exception("主题不能为空！");
+            }
+            ReConnected();
+
+            MessageBuilder.WithTopic(topic).WithRetainFlag();
+            if (qos == PublicQos.Qos_0)
+            {
+                MessageBuilder.WithAtLeastOnceQoS();
+            }
+            else if (qos == PublicQos.Qos_1)
+            {
+                MessageBuilder.WithAtMostOnceQoS();
+            }
+            else
+            {
+                MessageBuilder.WithExactlyOnceQoS();
+            }
+            var Message = MessageBuilder.Build();
+            try
+            {
+                await _mqttClient.PublishAsync(Message);
+            }
+            catch (Exception e)
+            {
+                if (_mqttClientConfig.Exception is null)
+                {
+                    throw e;
+                }
+                _mqttClientConfig.Exception(e);
+            }
+
+        }
+
+        /// <summary>
+        /// 发送消息,含有重连机制，如掉线会自动重连
+        /// </summary>
+        /// <param name="message">文字消息</param>
+        public async Task Publish(string topicName, string message, PublicQos qos = 0)
+        {
+            await PublishAsync(topicName, new MqttApplicationMessageBuilder()
+                .WithPayload(message), qos);
+        }
+        /// <summary>
+        /// 发送消息,含有重连机制，如掉线会自动重连
+        /// </summary>
+        /// <param name="message">消息流</param>
+        public async void Publish(string topicName, Stream message, PublicQos qos = 0)
+            => await PublishAsync(topicName, new MqttApplicationMessageBuilder()
+                .WithPayload(message), qos);
+
+        /// <summary>
+        /// 发送消息,含有重连机制，如掉线会自动重连
+        /// </summary>
+        /// <param name="message">Byte消息</param>
+        public async void Publish(string topicName, IEnumerable<byte> message, PublicQos qos = 0)
+            => await PublishAsync(topicName, new MqttApplicationMessageBuilder()
+                .WithPayload(message), qos);
+
+        /// <summary>
+        /// 发送消息,含有重连机制，如掉线会自动重连
+        /// </summary>
+        /// <param name="message">Byte消息</param>
+        public async void Publish(string topicName, byte[] message, PublicQos qos = 0)
+            => await PublishAsync(topicName, new MqttApplicationMessageBuilder()
+                .WithPayload(message), qos);
+
+        #endregion
+
+        /// <summary>
+        /// 启动服务
+        /// </summary>
+        /// <returns></returns>
+        public async Task Start()
+            => await _mqttClient.ConnectAsync(_options);
+
+        /// <summary>
+        /// 停止服务
+        /// </summary>
+        /// <returns></returns>
+        public async Task Stop()
+            => await _mqttClient.DisconnectAsync(new MqttClientDisconnectOptions { ReasonCode = MqttClientDisconnectReason.NormalDisconnection }, CancellationToken.None);
+
+    }
+    public class MqttClientConfig
+    {
+        private string _serverIp;
+        /// <summary>
+        /// 服务器IP
+        /// </summary>
+        public string ServerIp
+        {
+            get => _serverIp;
+            set
+            {
+                if (string.IsNullOrEmpty(value.Trim()))
+                {
+                    throw new ArgumentException("ServerIp can't be null or empty!");
+                }
+                _serverIp = value;
+            }
+        }
+        private int _port;
+        /// <summary>
+        /// 服务器端口
+        /// </summary>
+        public int Port
+        {
+            get => _port;
+            set
+            {
+                if (value <= 0)
+                {
+                    throw new ArgumentException("Port can't below the zero!");
+                }
+                _port = value;
+            }
+        }
+
+        /// <summary>
+        /// 用户名
+        /// </summary>
+        public string UserName { get; set; }
+        /// <summary>
+        /// 密码
+        /// </summary>
+        public string Password { get; set; }
+
+        private string _clientId;
+        /// <summary>
+        /// 唯一用户ID，默认使用Guid
+        /// </summary>
+        public string ClientId
+        {
+            get
+            {
+                _clientId = _clientId ?? Guid.NewGuid().ToString();
+                return _clientId;
+            }
+            set => _clientId = value;
+        }
+        /// <summary>
+        /// 客户端掉线重连时间，单位/s，默认5s
+        /// </summary>
+        public double ReconneTime { get; set; } = 5;
+        /// <summary>
+        /// 异常回调，默认为空，为空抛异常
+        /// </summary>
+        public Action<Exception> Exception = null;
+
+        /// <summary>
+        /// 接收消息回调，默认不接收
+        /// </summary>
+        public Action<MqttClientReciveMsg> ReciveMsgCallback = null;
+    }
+    public class MqttClientReciveMsg
+    {
+        /// <summary>
+        /// 主题
+        /// </summary>
+        public string Topic { get; set; }
+        /// <summary>
+        /// UTF-8格式下的 负载/消息
+        /// </summary>
+        public string Payload_UTF8 { get; set; }
+        /// <summary>
+        /// 原始 负载/消息
+        /// </summary>
+        public byte[] Payload { get; set; }
+        /// <summary>
+        /// Qos
+        /// </summary>
+        public MqttQualityOfServiceLevel Qos { get; set; }
+        /// <summary>
+        /// 保留
+        /// </summary>
+        public bool Retain { get; set; }
+    }
+    public enum PublicQos
+    {
+        /// <summary>
+        /// //At most once，至多一次
+        /// </summary>
+        Qos_0,
+        /// <summary>
+        /// //At least once，至少一次
+        /// </summary>
+        Qos_1,
+        /// <summary>
+        /// //QoS2，Exactly once
+        /// </summary>
+        Qos_2,
+    }
+}

MQTTnet.client.Test/Options/AspCoreMqttClientOptionBuilder.cs

@@ -0,0 +1,18 @@
+using MQTTnet.Client.Options;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+
+namespace MQTTnet.client.Test.Options
+{
+    public class AspCoreMqttClientOptionBuilder : MqttClientOptionsBuilder
+    {
+        public IServiceProvider ServiceProvider { get; }
+
+        public AspCoreMqttClientOptionBuilder(IServiceProvider serviceProvider)
+        {
+            ServiceProvider = serviceProvider;
+        }
+    }
+}

MQTTnet.client.Test/Pages/Error.cshtml

@@ -0,0 +1,26 @@
+@page
+@model ErrorModel
+@{
+    ViewData["Title"] = "Error";
+}
+
+<h1 class="text-danger">Error.</h1>
+<h2 class="text-danger">An error occurred while processing your request.</h2>
+
+@if (Model.ShowRequestId)
+{
+    <p>
+        <strong>Request ID:</strong> <code>@Model.RequestId</code>
+    </p>
+}
+
+<h3>Development Mode</h3>
+<p>
+    Swapping to the <strong>Development</strong> environment displays detailed information about the error that occurred.
+</p>
+<p>
+    <strong>The Development environment shouldn't be enabled for deployed applications.</strong>
+    It can result in displaying sensitive information from exceptions to end users.
+    For local debugging, enable the <strong>Development</strong> environment by setting the <strong>ASPNETCORE_ENVIRONMENT</strong> environment variable to <strong>Development</strong>
+    and restarting the app.
+</p>

MQTTnet.client.Test/Pages/Error.cshtml.cs

@@ -0,0 +1,31 @@
+using Microsoft.AspNetCore.Mvc;
+using Microsoft.AspNetCore.Mvc.RazorPages;
+using Microsoft.Extensions.Logging;
+using System;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.Linq;
+using System.Threading.Tasks;
+
+namespace MQTTnet.client.Test.Pages
+{
+    [ResponseCache(Duration = 0, Location = ResponseCacheLocation.None, NoStore = true)]
+    public class ErrorModel : PageModel
+    {
+        public string RequestId { get; set; }
+
+        public bool ShowRequestId => !string.IsNullOrEmpty(RequestId);
+
+        private readonly ILogger<ErrorModel> _logger;
+
+        public ErrorModel(ILogger<ErrorModel> logger)
+        {
+            _logger = logger;
+        }
+
+        public void OnGet()
+        {
+            RequestId = Activity.Current?.Id ?? HttpContext.TraceIdentifier;
+        }
+    }
+}

MQTTnet.client.Test/Pages/Index.cshtml

@@ -0,0 +1,10 @@
+@page
+@model IndexModel
+@{
+    ViewData["Title"] = "Home page";
+}
+
+<div class="text-center">
+    <h1 class="display-4">Welcome</h1>
+    <p>Learn about <a href="https://docs.microsoft.com/aspnet/core">building Web apps with ASP.NET Core</a>.</p>
+</div>

MQTTnet.client.Test/Pages/Index.cshtml.cs

@@ -0,0 +1,78 @@
+using Microsoft.AspNetCore.Mvc;
+using Microsoft.AspNetCore.Mvc.RazorPages;
+using Microsoft.Extensions.Logging;
+using MQTTnet.Client;
+using MQTTnet.Client.Connecting;
+using MQTTnet.Client.Disconnecting;
+using MQTTnet.Client.Options;
+using MQTTnet.Client.Receiving;
+using MQTTnet.Protocol;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace MQTTnet.client.Test.Pages
+{
+    public class IndexModel : PageModel
+    {
+        private readonly ILogger<IndexModel> _logger;
+        private IMqttClient _mqttClient;
+
+        private IMqttClientOptions _mqttClientOptions;
+        public IndexModel(ILogger<IndexModel> logger)
+        {
+            //, IMqttClient mqttClient, IMqttClientOptions mqttClientOptions
+            //_mqttClient =mqttClient;
+            //_mqttClientOptions=mqttClientOptions;
+           // Task.Run(RunAsync);
+            Task.Run(PublicAsync);
+        }
+
+        public async Task PublicAsync()
+        {
+            var applicationMessage = new MqttApplicationMessageBuilder()
+                .WithTopic("test")
+                .WithPayload("Hello World")
+                .WithAtLeastOnceQoS()
+                .Build();
+
+            await _mqttClient.PublishAsync(applicationMessage);
+        }
+
+        public  async Task RunAsync()
+        {
+            try
+            {
+             
+                IMqttClient client = _mqttClient;
+                client.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(e =>
+                {
+                    Console.WriteLine("### RECEIVED APPLICATION MESSAGE ###");
+                    Console.WriteLine($"+ Topic = {e.ApplicationMessage.Topic}");
+                    Console.WriteLine($"+ Payload = {Encoding.UTF8.GetString(e.ApplicationMessage.Payload)}");
+                    Console.WriteLine($"+ QoS = {e.ApplicationMessage.QualityOfServiceLevel}");
+                    Console.WriteLine($"+ Retain = {e.ApplicationMessage.Retain}");
+                   // Console.WriteLine();
+                });
+
+               
+                await client.SubscribeAsync(new MqttTopicFilter { Topic = "test", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
+                await client.SubscribeAsync(new MqttTopicFilter { Topic = "6666", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
+               
+            }
+            catch (Exception exception)
+            {
+                Console.WriteLine(exception);
+            }
+        }
+
+
+
+        public void OnGet()
+        {
+
+        }
+    }
+}

MQTTnet.client.Test/Pages/Privacy.cshtml

@@ -0,0 +1,8 @@
+@page
+@model PrivacyModel
+@{
+    ViewData["Title"] = "Privacy Policy";
+}
+<h1>@ViewData["Title"]</h1>
+
+<p>Use this page to detail your site's privacy policy.</p>

MQTTnet.client.Test/Pages/Privacy.cshtml.cs

@@ -0,0 +1,24 @@
+using Microsoft.AspNetCore.Mvc;
+using Microsoft.AspNetCore.Mvc.RazorPages;
+using Microsoft.Extensions.Logging;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+
+namespace MQTTnet.client.Test.Pages
+{
+    public class PrivacyModel : PageModel
+    {
+        private readonly ILogger<PrivacyModel> _logger;
+
+        public PrivacyModel(ILogger<PrivacyModel> logger)
+        {
+            _logger = logger;
+        }
+
+        public void OnGet()
+        {
+        }
+    }
+}

MQTTnet.client.Test/Pages/Shared/_Layout.cshtml

@@ -0,0 +1,50 @@
+<!DOCTYPE html>
+<html lang="en">
+<head>
+    <meta charset="utf-8" />
+    <meta name="viewport" content="width=device-width, initial-scale=1.0" />
+    <title>@ViewData["Title"] - MQTTnet.client.Test</title>
+    <link rel="stylesheet" href="~/lib/bootstrap/dist/css/bootstrap.min.css" />
+    <link rel="stylesheet" href="~/css/site.css" />
+</head>
+<body>
+    <header>
+        <nav class="navbar navbar-expand-sm navbar-toggleable-sm navbar-light bg-white border-bottom box-shadow mb-3">
+            <div class="container">
+                <a class="navbar-brand" asp-area="" asp-page="/Index">MQTTnet.client.Test</a>
+                <button class="navbar-toggler" type="button" data-toggle="collapse" data-target=".navbar-collapse" aria-controls="navbarSupportedContent"
+                        aria-expanded="false" aria-label="Toggle navigation">
+                    <span class="navbar-toggler-icon"></span>
+                </button>
+                <div class="navbar-collapse collapse d-sm-inline-flex flex-sm-row-reverse">
+                    <ul class="navbar-nav flex-grow-1">
+                        <li class="nav-item">
+                            <a class="nav-link text-dark" asp-area="" asp-page="/Index">Home</a>
+                        </li>
+                        <li class="nav-item">
+                            <a class="nav-link text-dark" asp-area="" asp-page="/Privacy">Privacy</a>
+                        </li>
+                    </ul>
+                </div>
+            </div>
+        </nav>
+    </header>
+    <div class="container">
+        <main role="main" class="pb-3">
+            @RenderBody()
+        </main>
+    </div>
+
+    <footer class="border-top footer text-muted">
+        <div class="container">
+            &copy; 2022 - MQTTnet.client.Test - <a asp-area="" asp-page="/Privacy">Privacy</a>
+        </div>
+    </footer>
+
+    <script src="~/lib/jquery/dist/jquery.min.js"></script>
+    <script src="~/lib/bootstrap/dist/js/bootstrap.bundle.min.js"></script>
+    <script src="~/js/site.js" asp-append-version="true"></script>
+
+    @RenderSection("Scripts", required: false)
+</body>
+</html>

MQTTnet.client.Test/Pages/Shared/_ValidationScriptsPartial.cshtml

@@ -0,0 +1,2 @@
+<script src="~/lib/jquery-validation/dist/jquery.validate.min.js"></script>
+<script src="~/lib/jquery-validation-unobtrusive/jquery.validate.unobtrusive.min.js"></script>

MQTTnet.client.Test/Pages/_ViewImports.cshtml

@@ -0,0 +1,3 @@
+@using MQTTnet.client.Test
+@namespace MQTTnet.client.Test.Pages
+@addTagHelper *, Microsoft.AspNetCore.Mvc.TagHelpers

MQTTnet.client.Test/Pages/_ViewStart.cshtml

@@ -0,0 +1,3 @@
+@{
+    Layout = "_Layout";
+}

MQTTnet.client.Test/Program.cs

@@ -0,0 +1,26 @@
+using Microsoft.AspNetCore.Hosting;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+
+namespace MQTTnet.client.Test
+{
+    public class Program
+    {
+        public static void Main(string[] args)
+        {
+            CreateHostBuilder(args).Build().Run();
+        }
+
+        public static IHostBuilder CreateHostBuilder(string[] args) =>
+            Host.CreateDefaultBuilder(args)
+                .ConfigureWebHostDefaults(webBuilder =>
+                {
+                    webBuilder.UseStartup<Startup>();
+                });
+    }
+}

MQTTnet.client.Test/Settings/AppSettingsProvider.cs

@@ -0,0 +1,13 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+
+namespace MQTTnet.client.Test.Settings
+{
+    public class AppSettingsProvider
+    {
+        public static BrokerHostSettings BrokerHostSettings;
+        public static ClientSettings ClientSettings;
+    }
+}

MQTTnet.client.Test/Settings/BrokerHostSettings.cs

@@ -0,0 +1,13 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+
+namespace MQTTnet.client.Test.Settings
+{
+    public class BrokerHostSettings
+    {
+        public string Host { set; get; }
+        public int Port { set; get; }
+    }
+}

MQTTnet.client.Test/Settings/ClientSettings.cs

@@ -0,0 +1,14 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+
+namespace MQTTnet.client.Test.Settings
+{
+    public class ClientSettings
+    {
+        public string Id { set; get; }
+        public string UserName { set; get; }
+        public string Password { set; get; }
+    }
+}

MQTTnet.client.Test/Startup.cs

@@ -0,0 +1,70 @@
+using Microsoft.AspNetCore.Builder;
+using Microsoft.AspNetCore.Hosting;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using MQTTnet.client.Test.Extensions;
+using MQTTnet.client.Test.Settings;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+
+namespace MQTTnet.client.Test
+{
+    public class Startup
+    {
+        public Startup(IConfiguration configuration)
+        {
+            Configuration = configuration;
+            MapBrokerHostSettings();
+            MapClientSettings();
+        }
+
+        public IConfiguration Configuration { get; }
+
+        // This method gets called by the runtime. Use this method to add services to the container.
+        public void ConfigureServices(IServiceCollection services)
+        {
+            services.AddRazorPages();
+            services.AddMqttClientHostedService(Configuration);
+        }
+        private void MapBrokerHostSettings()
+        {
+            BrokerHostSettings brokerHostSettings = new BrokerHostSettings();
+            Configuration.GetSection(nameof(BrokerHostSettings)).Bind(brokerHostSettings);
+            AppSettingsProvider.BrokerHostSettings = brokerHostSettings;
+        }
+
+        private void MapClientSettings()
+        {
+            ClientSettings clientSettings = new ClientSettings();
+            Configuration.GetSection(nameof(ClientSettings)).Bind(clientSettings);
+            AppSettingsProvider.ClientSettings = clientSettings;
+        }
+
+        // This method gets called by the runtime. Use this method to configure the HTTP request pipeline.
+        public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
+        {
+            if (env.IsDevelopment())
+            {
+                app.UseDeveloperExceptionPage();
+            }
+            else
+            {
+                app.UseExceptionHandler("/Error");
+            }
+
+            app.UseStaticFiles();
+
+            app.UseRouting();
+
+            app.UseAuthorization();
+
+            app.UseEndpoints(endpoints =>
+            {
+                endpoints.MapRazorPages();
+            });
+        }
+    }
+}

MQTTnet.client.Test/appsettings.Development.json

@@ -0,0 +1,9 @@
+{
+  "Logging": {
+    "LogLevel": {
+      "Default": "Information",
+      "Microsoft": "Warning",
+      "Microsoft.Hosting.Lifetime": "Information"
+    }
+  }
+}

MQTTnet.client.Test/appsettings.json

@@ -0,0 +1,20 @@
+{
+  "Logging": {
+    "LogLevel": {
+      "Default": "Information",
+      "Microsoft": "Warning",
+      "Microsoft.Hosting.Lifetime": "Information"
+    }
+  },
+  "AllowedHosts": "*",
+  "BrokerHostSettings": {
+    "Host": "10.2.1.21",
+    "Port": 1883
+  },
+
+  "ClientSettings": {
+    "Id": "5eb020f043ba8930506acbdd",
+    "UserName": "rafiul",
+    "Password": "12345678"
+  }
+}

MQTTnet.client.Test/wwwroot/css/site.css

@@ -0,0 +1,71 @@
+/* Please see documentation at https://docs.microsoft.com/aspnet/core/client-side/bundling-and-minification
+for details on configuring this project to bundle and minify static web assets. */
+
+a.navbar-brand {
+  white-space: normal;
+  text-align: center;
+  word-break: break-all;
+}
+
+/* Provide sufficient contrast against white background */
+a {
+  color: #0366d6;
+}
+
+.btn-primary {
+  color: #fff;
+  background-color: #1b6ec2;
+  border-color: #1861ac;
+}
+
+.nav-pills .nav-link.active, .nav-pills .show > .nav-link {
+  color: #fff;
+  background-color: #1b6ec2;
+  border-color: #1861ac;
+}
+
+/* Sticky footer styles
+-------------------------------------------------- */
+html {
+  font-size: 14px;
+}
+@media (min-width: 768px) {
+  html {
+    font-size: 16px;
+  }
+}
+
+.border-top {
+  border-top: 1px solid #e5e5e5;
+}
+.border-bottom {
+  border-bottom: 1px solid #e5e5e5;
+}
+
+.box-shadow {
+  box-shadow: 0 .25rem .75rem rgba(0, 0, 0, .05);
+}
+
+button.accept-policy {
+  font-size: 1rem;
+  line-height: inherit;
+}
+
+/* Sticky footer styles
+-------------------------------------------------- */
+html {
+  position: relative;
+  min-height: 100%;
+}
+
+body {
+  /* Margin bottom by footer height */
+  margin-bottom: 60px;
+}
+.footer {
+  position: absolute;
+  bottom: 0;
+  width: 100%;
+  white-space: nowrap;
+  line-height: 60px; /* Vertically center the text there */
+}

MQTTnet.client.Test/wwwroot/lib/bootstrap/LICENSE

@@ -0,0 +1,22 @@
+The MIT License (MIT)
+
+Copyright (c) 2011-2018 Twitter, Inc.
+Copyright (c) 2011-2018 The Bootstrap Authors
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.

MQTTnet.client.Test/wwwroot/lib/bootstrap/dist/css/bootstrap-reboot.css

@@ -0,0 +1,331 @@
+/*!
+ * Bootstrap Reboot v4.3.1 (https://getbootstrap.com/)
+ * Copyright 2011-2019 The Bootstrap Authors
+ * Copyright 2011-2019 Twitter, Inc.
+ * Licensed under MIT (https://github.com/twbs/bootstrap/blob/master/LICENSE)
+ * Forked from Normalize.css, licensed MIT (https://github.com/necolas/normalize.css/blob/master/LICENSE.md)
+ */
+*,
+*::before,
+*::after {
+  box-sizing: border-box;
+}
+
+html {
+  font-family: sans-serif;
+  line-height: 1.15;
+  -webkit-text-size-adjust: 100%;
+  -webkit-tap-highlight-color: rgba(0, 0, 0, 0);
+}
+
+article, aside, figcaption, figure, footer, header, hgroup, main, nav, section {
+  display: block;
+}
+
+body {
+  margin: 0;
+  font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, "Helvetica Neue", Arial, "Noto Sans", sans-serif, "Apple Color Emoji", "Segoe UI Emoji", "Segoe UI Symbol", "Noto Color Emoji";
+  font-size: 1rem;
+  font-weight: 400;
+  line-height: 1.5;
+  color: #212529;
+  text-align: left;
+  background-color: #fff;
+}
+
+[tabindex="-1"]:focus {
+  outline: 0 !important;
+}
+
+hr {
+  box-sizing: content-box;
+  height: 0;
+  overflow: visible;
+}
+
+h1, h2, h3, h4, h5, h6 {
+  margin-top: 0;
+  margin-bottom: 0.5rem;
+}
+
+p {
+  margin-top: 0;
+  margin-bottom: 1rem;
+}
+
+abbr[title],
+abbr[data-original-title] {
+  text-decoration: underline;
+  -webkit-text-decoration: underline dotted;
+  text-decoration: underline dotted;
+  cursor: help;
+  border-bottom: 0;
+  -webkit-text-decoration-skip-ink: none;
+  text-decoration-skip-ink: none;
+}
+
+address {
+  margin-bottom: 1rem;
+  font-style: normal;
+  line-height: inherit;
+}
+
+ol,
+ul,
+dl {
+  margin-top: 0;
+  margin-bottom: 1rem;
+}
+
+ol ol,
+ul ul,
+ol ul,
+ul ol {
+  margin-bottom: 0;
+}
+
+dt {
+  font-weight: 700;
+}
+
+dd {
+  margin-bottom: .5rem;
+  margin-left: 0;
+}
+
+blockquote {
+  margin: 0 0 1rem;
+}
+
+b,
+strong {
+  font-weight: bolder;
+}
+
+small {
+  font-size: 80%;
+}
+
+sub,
+sup {
+  position: relative;
+  font-size: 75%;
+  line-height: 0;
+  vertical-align: baseline;
+}
+
+sub {
+  bottom: -.25em;
+}
+
+sup {
+  top: -.5em;
+}
+
+a {
+  color: #007bff;
+  text-decoration: none;
+  background-color: transparent;
+}
+
+a:hover {
+  color: #0056b3;
+  text-decoration: underline;
+}
+
+a:not([href]):not([tabindex]) {
+  color: inherit;
+  text-decoration: none;
+}
+
+a:not([href]):not([tabindex]):hover, a:not([href]):not([tabindex]):focus {
+  color: inherit;
+  text-decoration: none;
+}
+
+a:not([href]):not([tabindex]):focus {
+  outline: 0;
+}
+
+pre,
+code,
+kbd,
+samp {
+  font-family: SFMono-Regular, Menlo, Monaco, Consolas, "Liberation Mono", "Courier New", monospace;
+  font-size: 1em;
+}
+
+pre {
+  margin-top: 0;
+  margin-bottom: 1rem;
+  overflow: auto;
+}
+
+figure {
+  margin: 0 0 1rem;
+}
+
+img {
+  vertical-align: middle;
+  border-style: none;
+}
+
+svg {
+  overflow: hidden;
+  vertical-align: middle;
+}
+
+table {
+  border-collapse: collapse;
+}
+
+caption {
+  padding-top: 0.75rem;
+  padding-bottom: 0.75rem;
+  color: #6c757d;
+  text-align: left;
+  caption-side: bottom;
+}
+
+th {
+  text-align: inherit;
+}
+
+label {
+  display: inline-block;
+  margin-bottom: 0.5rem;
+}
+
+button {
+  border-radius: 0;
+}
+
+button:focus {
+  outline: 1px dotted;
+  outline: 5px auto -webkit-focus-ring-color;
+}
+
+input,
+button,
+select,
+optgroup,
+textarea {
+  margin: 0;
+  font-family: inherit;
+  font-size: inherit;
+  line-height: inherit;
+}
+
+button,
+input {
+  overflow: visible;
+}
+
+button,
+select {
+  text-transform: none;
+}
+
+select {
+  word-wrap: normal;
+}
+
+button,
+[type="button"],
+[type="reset"],
+[type="submit"] {
+  -webkit-appearance: button;
+}
+
+button:not(:disabled),
+[type="button"]:not(:disabled),
+[type="reset"]:not(:disabled),
+[type="submit"]:not(:disabled) {
+  cursor: pointer;
+}
+
+button::-moz-focus-inner,
+[type="button"]::-moz-focus-inner,
+[type="reset"]::-moz-focus-inner,
+[type="submit"]::-moz-focus-inner {
+  padding: 0;
+  border-style: none;
+}
+
+input[type="radio"],
+input[type="checkbox"] {
+  box-sizing: border-box;
+  padding: 0;
+}
+
+input[type="date"],
+input[type="time"],
+input[type="datetime-local"],
+input[type="month"] {
+  -webkit-appearance: listbox;
+}
+
+textarea {
+  overflow: auto;
+  resize: vertical;
+}
+
+fieldset {
+  min-width: 0;
+  padding: 0;
+  margin: 0;
+  border: 0;
+}
+
+legend {
+  display: block;
+  width: 100%;
+  max-width: 100%;
+  padding: 0;
+  margin-bottom: .5rem;
+  font-size: 1.5rem;
+  line-height: inherit;
+  color: inherit;
+  white-space: normal;
+}
+
+progress {
+  vertical-align: baseline;
+}
+
+[type="number"]::-webkit-inner-spin-button,
+[type="number"]::-webkit-outer-spin-button {
+  height: auto;
+}
+
+[type="search"] {
+  outline-offset: -2px;
+  -webkit-appearance: none;
+}
+
+[type="search"]::-webkit-search-decoration {
+  -webkit-appearance: none;
+}
+
+::-webkit-file-upload-button {
+  font: inherit;
+  -webkit-appearance: button;
+}
+
+output {
+  display: inline-block;
+}
+
+summary {
+  display: list-item;
+  cursor: pointer;
+}
+
+template {
+  display: none;
+}
+
+[hidden] {
+  display: none !important;
+}
+/*# sourceMappingURL=bootstrap-reboot.css.map */

MQTTnet.client.Test/wwwroot/lib/bootstrap/dist/css/bootstrap-reboot.min.css

@@ -0,0 +1,8 @@
+/*!
+ * Bootstrap Reboot v4.3.1 (https://getbootstrap.com/)
+ * Copyright 2011-2019 The Bootstrap Authors
+ * Copyright 2011-2019 Twitter, Inc.
+ * Licensed under MIT (https://github.com/twbs/bootstrap/blob/master/LICENSE)
+ * Forked from Normalize.css, licensed MIT (https://github.com/necolas/normalize.css/blob/master/LICENSE.md)
+ */*,::after,::before{box-sizing:border-box}html{font-family:sans-serif;line-height:1.15;-webkit-text-size-adjust:100%;-webkit-tap-highlight-color:transparent}article,aside,figcaption,figure,footer,header,hgroup,main,nav,section{display:block}body{margin:0;font-family:-apple-system,BlinkMacSystemFont,"Segoe UI",Roboto,"Helvetica Neue",Arial,"Noto Sans",sans-serif,"Apple Color Emoji","Segoe UI Emoji","Segoe UI Symbol","Noto Color Emoji";font-size:1rem;font-weight:400;line-height:1.5;color:#212529;text-align:left;background-color:#fff}[tabindex="-1"]:focus{outline:0!important}hr{box-sizing:content-box;height:0;overflow:visible}h1,h2,h3,h4,h5,h6{margin-top:0;margin-bottom:.5rem}p{margin-top:0;margin-bottom:1rem}abbr[data-original-title],abbr[title]{text-decoration:underline;-webkit-text-decoration:underline dotted;text-decoration:underline dotted;cursor:help;border-bottom:0;-webkit-text-decoration-skip-ink:none;text-decoration-skip-ink:none}address{margin-bottom:1rem;font-style:normal;line-height:inherit}dl,ol,ul{margin-top:0;margin-bottom:1rem}ol ol,ol ul,ul ol,ul ul{margin-bottom:0}dt{font-weight:700}dd{margin-bottom:.5rem;margin-left:0}blockquote{margin:0 0 1rem}b,strong{font-weight:bolder}small{font-size:80%}sub,sup{position:relative;font-size:75%;line-height:0;vertical-align:baseline}sub{bottom:-.25em}sup{top:-.5em}a{color:#007bff;text-decoration:none;background-color:transparent}a:hover{color:#0056b3;text-decoration:underline}a:not([href]):not([tabindex]){color:inherit;text-decoration:none}a:not([href]):not([tabindex]):focus,a:not([href]):not([tabindex]):hover{color:inherit;text-decoration:none}a:not([href]):not([tabindex]):focus{outline:0}code,kbd,pre,samp{font-family:SFMono-Regular,Menlo,Monaco,Consolas,"Liberation Mono","Courier New",monospace;font-size:1em}pre{margin-top:0;margin-bottom:1rem;overflow:auto}figure{margin:0 0 1rem}img{vertical-align:middle;border-style:none}svg{overflow:hidden;vertical-align:middle}table{border-collapse:collapse}caption{padding-top:.75rem;padding-bottom:.75rem;color:#6c757d;text-align:left;caption-side:bottom}th{text-align:inherit}label{display:inline-block;margin-bottom:.5rem}button{border-radius:0}button:focus{outline:1px dotted;outline:5px auto -webkit-focus-ring-color}button,input,optgroup,select,textarea{margin:0;font-family:inherit;font-size:inherit;line-height:inherit}button,input{overflow:visible}button,select{text-transform:none}select{word-wrap:normal}[type=button],[type=reset],[type=submit],button{-webkit-appearance:button}[type=button]:not(:disabled),[type=reset]:not(:disabled),[type=submit]:not(:disabled),button:not(:disabled){cursor:pointer}[type=button]::-moz-focus-inner,[type=reset]::-moz-focus-inner,[type=submit]::-moz-focus-inner,button::-moz-focus-inner{padding:0;border-style:none}input[type=checkbox],input[type=radio]{box-sizing:border-box;padding:0}input[type=date],input[type=datetime-local],input[type=month],input[type=time]{-webkit-appearance:listbox}textarea{overflow:auto;resize:vertical}fieldset{min-width:0;padding:0;margin:0;border:0}legend{display:block;width:100%;max-width:100%;padding:0;margin-bottom:.5rem;font-size:1.5rem;line-height:inherit;color:inherit;white-space:normal}progress{vertical-align:baseline}[type=number]::-webkit-inner-spin-button,[type=number]::-webkit-outer-spin-button{height:auto}[type=search]{outline-offset:-2px;-webkit-appearance:none}[type=search]::-webkit-search-decoration{-webkit-appearance:none}::-webkit-file-upload-button{font:inherit;-webkit-appearance:button}output{display:inline-block}summary{display:list-item;cursor:pointer}template{display:none}[hidden]{display:none!important}
+/*# sourceMappingURL=bootstrap-reboot.min.css.map */

MQTTnet.client.Test/wwwroot/lib/jquery-validation-unobtrusive/LICENSE.txt

@@ -0,0 +1,12 @@
+Copyright (c) .NET Foundation. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License"); you may not use
+these files except in compliance with the License. You may obtain a copy of the
+License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software distributed
+under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
+CONDITIONS OF ANY KIND, either express or implied. See the License for the
+specific language governing permissions and limitations under the License.

MQTTnet.client.Test/wwwroot/lib/jquery-validation/LICENSE.md

@@ -0,0 +1,22 @@
+The MIT License (MIT)
+=====================
+
+Copyright Jörn Zaefferer
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.

MQTTnet.client.Test/wwwroot/lib/jquery/LICENSE.txt

@@ -0,0 +1,36 @@
+Copyright JS Foundation and other contributors, https://js.foundation/
+
+This software consists of voluntary contributions made by many
+individuals. For exact contribution history, see the revision history
+available at https://github.com/jquery/jquery
+
+The following license applies to all parts of this software except as
+documented below:
+
+====
+
+Permission is hereby granted, free of charge, to any person obtaining
+a copy of this software and associated documentation files (the
+"Software"), to deal in the Software without restriction, including
+without limitation the rights to use, copy, modify, merge, publish,
+distribute, sublicense, and/or sell copies of the Software, and to
+permit persons to whom the Software is furnished to do so, subject to
+the following conditions:
+
+The above copyright notice and this permission notice shall be
+included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
+LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+====
+
+All files located in the node_modules and external directories are
+externally maintained libraries used by this software which have their
+own licenses; we recommend you read them, as their terms may differ from
+the terms above.

MQTTnet.sln

@@ -57,6 +57,12 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Test.BlazorApp.Clie
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Test.BlazorApp.Shared", "Tests\MQTTnet.Test.BlazorApp\Shared\MQTTnet.Test.BlazorApp.Shared.csproj", "{DDB069BA-6E1A-48C7-B374-5D903B553CAD}"
 EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.client.Test", "MQTTnet.client.Test\MQTTnet.client.Test.csproj", "{DEA26B21-C87D-4426-8F13-3BE5D3F61673}"
+EndProject
+Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "BPA", "BPA", "{9D386218-14AA-4340-96B5-540A60B28B87}"
+EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "BPA.MQTTnet.Extensions", "BPA.MQTTnet.Extensions\BPA.MQTTnet.Extensions.csproj", "{B860A392-D4EC-4034-B7A2-5E49C884C28B}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -277,6 +283,38 @@ Global
 		{DDB069BA-6E1A-48C7-B374-5D903B553CAD}.Release|x64.Build.0 = Release|Any CPU
 		{DDB069BA-6E1A-48C7-B374-5D903B553CAD}.Release|x86.ActiveCfg = Release|Any CPU
 		{DDB069BA-6E1A-48C7-B374-5D903B553CAD}.Release|x86.Build.0 = Release|Any CPU
+		{DEA26B21-C87D-4426-8F13-3BE5D3F61673}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{DEA26B21-C87D-4426-8F13-3BE5D3F61673}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{DEA26B21-C87D-4426-8F13-3BE5D3F61673}.Debug|ARM.ActiveCfg = Debug|Any CPU
+		{DEA26B21-C87D-4426-8F13-3BE5D3F61673}.Debug|ARM.Build.0 = Debug|Any CPU
+		{DEA26B21-C87D-4426-8F13-3BE5D3F61673}.Debug|x64.ActiveCfg = Debug|Any CPU
+		{DEA26B21-C87D-4426-8F13-3BE5D3F61673}.Debug|x64.Build.0 = Debug|Any CPU
+		{DEA26B21-C87D-4426-8F13-3BE5D3F61673}.Debug|x86.ActiveCfg = Debug|Any CPU
+		{DEA26B21-C87D-4426-8F13-3BE5D3F61673}.Debug|x86.Build.0 = Debug|Any CPU
+		{DEA26B21-C87D-4426-8F13-3BE5D3F61673}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{DEA26B21-C87D-4426-8F13-3BE5D3F61673}.Release|Any CPU.Build.0 = Release|Any CPU
+		{DEA26B21-C87D-4426-8F13-3BE5D3F61673}.Release|ARM.ActiveCfg = Release|Any CPU
+		{DEA26B21-C87D-4426-8F13-3BE5D3F61673}.Release|ARM.Build.0 = Release|Any CPU
+		{DEA26B21-C87D-4426-8F13-3BE5D3F61673}.Release|x64.ActiveCfg = Release|Any CPU
+		{DEA26B21-C87D-4426-8F13-3BE5D3F61673}.Release|x64.Build.0 = Release|Any CPU
+		{DEA26B21-C87D-4426-8F13-3BE5D3F61673}.Release|x86.ActiveCfg = Release|Any CPU
+		{DEA26B21-C87D-4426-8F13-3BE5D3F61673}.Release|x86.Build.0 = Release|Any CPU
+		{B860A392-D4EC-4034-B7A2-5E49C884C28B}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{B860A392-D4EC-4034-B7A2-5E49C884C28B}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{B860A392-D4EC-4034-B7A2-5E49C884C28B}.Debug|ARM.ActiveCfg = Debug|Any CPU
+		{B860A392-D4EC-4034-B7A2-5E49C884C28B}.Debug|ARM.Build.0 = Debug|Any CPU
+		{B860A392-D4EC-4034-B7A2-5E49C884C28B}.Debug|x64.ActiveCfg = Debug|Any CPU
+		{B860A392-D4EC-4034-B7A2-5E49C884C28B}.Debug|x64.Build.0 = Debug|Any CPU
+		{B860A392-D4EC-4034-B7A2-5E49C884C28B}.Debug|x86.ActiveCfg = Debug|Any CPU
+		{B860A392-D4EC-4034-B7A2-5E49C884C28B}.Debug|x86.Build.0 = Debug|Any CPU
+		{B860A392-D4EC-4034-B7A2-5E49C884C28B}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{B860A392-D4EC-4034-B7A2-5E49C884C28B}.Release|Any CPU.Build.0 = Release|Any CPU
+		{B860A392-D4EC-4034-B7A2-5E49C884C28B}.Release|ARM.ActiveCfg = Release|Any CPU
+		{B860A392-D4EC-4034-B7A2-5E49C884C28B}.Release|ARM.Build.0 = Release|Any CPU
+		{B860A392-D4EC-4034-B7A2-5E49C884C28B}.Release|x64.ActiveCfg = Release|Any CPU
+		{B860A392-D4EC-4034-B7A2-5E49C884C28B}.Release|x64.Build.0 = Release|Any CPU
+		{B860A392-D4EC-4034-B7A2-5E49C884C28B}.Release|x86.ActiveCfg = Release|Any CPU
+		{B860A392-D4EC-4034-B7A2-5E49C884C28B}.Release|x86.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -296,6 +334,8 @@ Global
 		{A9662AF3-3520-4BF8-9DFF-C55C0C33D08F} = {61B165A0-5AA8-4E04-A53D-A22A84AA6EB7}
 		{D260D63D-7902-4C55-9665-84C5CACBBB24} = {61B165A0-5AA8-4E04-A53D-A22A84AA6EB7}
 		{DDB069BA-6E1A-48C7-B374-5D903B553CAD} = {61B165A0-5AA8-4E04-A53D-A22A84AA6EB7}
+		{DEA26B21-C87D-4426-8F13-3BE5D3F61673} = {9248C2E1-B9D6-40BF-81EC-86004D7765B4}
+		{B860A392-D4EC-4034-B7A2-5E49C884C28B} = {9D386218-14AA-4340-96B5-540A60B28B87}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {07536672-5CBC-4BE3-ACE0-708A431A7894}

Source/MQTTnet/MQTTnet.csproj

@@ -1,5 +1,4 @@
-<Project Sdk="Microsoft.NET.Sdk">
-
+<Project Sdk="Microsoft.NET.Sdk">
   <PropertyGroup>
     <TargetFrameworks>netstandard1.3;netstandard2.0;netstandard2.1;netcoreapp3.1;net5.0</TargetFrameworks>
     <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">$(TargetFrameworks);net452;net461</TargetFrameworks>
@@ -7,18 +6,19 @@
     <AssemblyName>MQTTnet</AssemblyName>
     <RootNamespace>MQTTnet</RootNamespace>
     <GeneratePackageOnBuild>False</GeneratePackageOnBuild>
-    <Company />
-    <Product />
-    <Description />
-    <Authors />
-    <PackageId />
+    <Company>黑菠萝</Company>
+    <Product>BPA.MQTTnet</Product>
+    <Description>黑菠萝-BPA.MQTTnet</Description>
+    <Authors>BPA.MQTTnet</Authors>
+    <PackageId>BPA.MQTTnet</PackageId>
     <SignAssembly>false</SignAssembly>
     <DelaySign>false</DelaySign>
     <PublishRepositoryUrl>true</PublishRepositoryUrl>
     <IncludeSymbols>true</IncludeSymbols>
     <SymbolPackageFormat>snupkg</SymbolPackageFormat>
+    <VersionPrefix>1.0.3</VersionPrefix>
+    <PackageReleaseNotes />
   </PropertyGroup>
-
   <PropertyGroup Condition="'$(TargetFramework)'=='uap10.0'">
     <CopyLocalLockFileAssemblies>false</CopyLocalLockFileAssemblies>
     <NugetTargetMoniker>UAP,Version=v10.0</NugetTargetMoniker>
@@ -31,23 +31,18 @@
     <DefaultLanguage>en</DefaultLanguage>
     <LanguageTargets>$(MSBuildExtensionsPath)\Microsoft\WindowsXaml\v$(VisualStudioVersion)\Microsoft.Windows.UI.Xaml.CSharp.targets</LanguageTargets>
   </PropertyGroup>
-
   <PropertyGroup Condition="'$(Configuration)'=='Debug'">
     <DebugType>Full</DebugType>
   </PropertyGroup>
-
   <ItemGroup>
     <PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.0.0" PrivateAssets="All" />
   </ItemGroup>
-
   <ItemGroup Condition="'$(TargetFramework)'=='netstandard1.3'">
     <PackageReference Include="System.Net.Security" Version="4.3.2" />
     <PackageReference Include="System.Net.WebSockets" Version="4.3.0" />
     <PackageReference Include="System.Net.WebSockets.Client" Version="4.3.2" />
   </ItemGroup>
-
   <ItemGroup Condition="'$(TargetFramework)'=='uap10.0'">
     <PackageReference Include="Microsoft.NETCore.UniversalWindowsPlatform" Version="6.2.10" />
   </ItemGroup>
-
 </Project>

Tests/MQTTnet.AspNetCore.Tests/MqttClient.cs

@@ -0,0 +1,433 @@
+using MQTTnet;
+using MQTTnet.Client;
+using MQTTnet.Client.Disconnecting;
+using MQTTnet.Client.Options;
+using MQTTnet.Protocol;
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace MQTT
+{
+    public class MqttClient
+    {
+        #region Field Area
+
+        /// <summary>
+        /// Mqtt factory
+        /// </summary>
+        private MqttFactory _factory;
+        /// <summary>
+        /// Mqtt client
+        /// </summary>
+        private IMqttClient _mqttClient;
+        /// <summary>
+        /// Mqtt 配置信息
+        /// </summary>
+        private MqttClientConfig _mqttClientConfig;
+        /// <summary>
+        /// Mqtt options
+        /// </summary>
+        private IMqttClientOptions _options;
+
+        #endregion
+
+        #region CTOR
+
+        /// <summary>
+        /// 默认启动IP为127.0.0.1 端口为1883
+        /// </summary>
+        public MqttClient()
+        {
+            _mqttClientConfig = new MqttClientConfig
+            {
+                ServerIp = "127.0.0.1",
+                Port = 1883
+            };
+            Init();
+        }
+        /// <summary>
+        /// 调用示例
+        /// 
+        /// //var client = new MqttClient(s =>
+        ///    {
+        ///        s.Port = 1883;
+        ///        s.ServerIp = "127.0.0.1";
+        ///        s.UserName = "mqtt-test";
+        ///        s.Password = "mqtt-test";
+        ///        s.ReciveMsgCallback = s =>
+        ///        {
+        ///            Console.WriteLine(s.Payload_UTF8);
+        ///        };
+        ///    }, true);
+        ///    client.Subscribe("/TopicName/");
+        /// </summary>
+        /// <param name="config">客户端配置信息</param>
+        /// <param name="autoStart">直接启动</param>
+        public MqttClient(Action<MqttClientConfig> config, bool autoStart = false)
+        {
+            _mqttClientConfig = new MqttClientConfig();
+            config(_mqttClientConfig);
+            Init();
+            if (autoStart)
+            {
+                Task.WaitAll( Start());
+            }
+        }
+
+        #endregion
+
+        /// <summary>
+        /// 获取MqttClient实例
+        /// 
+        /// 调用示例
+        ///    //var client = MqttClient.Instance(s =>
+        ///    {
+        ///        s.Port = 1883;
+        ///        s.ServerIp = "127.0.0.1";
+        ///        s.UserName = "mqtt-test";
+        ///        s.Password = "mqtt-test";
+        ///        s.ReciveMsgCallback = s =>
+        ///        {
+        ///            Console.WriteLine(s.Payload_UTF8);
+        ///        };
+        ///    }, true);
+        ///    client.Subscribe("/TopicName/");
+        /// </summary>
+        /// <param name="config">客户端配置信息</param>
+        /// <param name="autoStart">直接启动</param>
+        /// <returns></returns>
+        public static MqttClient Instance(Action<MqttClientConfig> config, bool autoStart = false)
+            => new MqttClient(config, autoStart);
+
+        /// <summary>
+        /// 初始化注册
+        /// </summary>
+        private void Init()
+        {
+            try
+            {
+                _factory = new MqttFactory();
+
+                _mqttClient = _factory.CreateMqttClient();
+
+                _options = new MqttClientOptionsBuilder()
+                    .WithTcpServer(_mqttClientConfig.ServerIp, _mqttClientConfig.Port)
+                    .WithCredentials(_mqttClientConfig.UserName, _mqttClientConfig.Password)
+                    .WithClientId(_mqttClientConfig.ClientId)
+                    .Build();
+
+                //消息回调
+                _mqttClient.UseApplicationMessageReceivedHandler(ReciveMsg);
+            }
+            catch (Exception exp)
+            {
+                if (_mqttClientConfig.Exception is null)
+                {
+                    throw exp;
+                }
+                _mqttClientConfig.Exception(exp);
+            }
+        }
+
+        #region 内部事件转换处理
+
+        /// <summary>
+        /// 消息接收回调
+        /// </summary>
+        /// <param name="e"></param>
+        /// <returns></returns>
+        private void ReciveMsg(MqttApplicationMessageReceivedEventArgs e)
+        {
+            if (_mqttClientConfig.ReciveMsgCallback != null)
+            {
+                _mqttClientConfig.ReciveMsgCallback(new MqttClientReciveMsg
+                {
+                    Topic = e.ApplicationMessage.Topic,
+                    Payload_UTF8 = Encoding.UTF8.GetString(e.ApplicationMessage.Payload),
+                    Payload = e.ApplicationMessage.Payload,
+                    Qos = e.ApplicationMessage.QualityOfServiceLevel,
+                    Retain = e.ApplicationMessage.Retain,
+                });
+            }
+        }
+
+        /// <summary>
+        /// 订阅
+        /// </summary>
+        /// <param name="topicName"></param>
+        public async void Subscribe(string topicName)
+        {
+            topicName = topicName.Trim();
+            if (string.IsNullOrEmpty(topicName))
+            {
+                throw new Exception("订阅主题不能为空！");
+            }
+
+            if (!_mqttClient.IsConnected)
+            {
+                throw new Exception("MQTT客户端尚未连接！请先启动连接");
+            }
+            await _mqttClient.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic(topicName).Build());
+        }
+
+        /// <summary>
+        /// 取消订阅
+        /// </summary>
+        /// <param name="topicName"></param>
+        public async void Unsubscribe(string topicName)
+        {
+            topicName = topicName.Trim();
+            if (string.IsNullOrEmpty(topicName))
+            {
+                throw new Exception("订阅主题不能为空！");
+            }
+
+            if (!_mqttClient.IsConnected)
+            {
+                throw new Exception("MQTT客户端尚未连接！请先启动连接");
+            }
+            await _mqttClient.UnsubscribeAsync(topicName);
+        }
+
+        /// <summary>
+        /// 重连机制
+        /// </summary>
+        private void ReConnected()
+        {
+            if (_mqttClient.IsConnected)
+            {
+                return;
+            }
+            for (int i = 0; i < 10; i++)
+            {
+                //重连机制
+                _mqttClient.UseDisconnectedHandler(async e =>
+                {
+                    try
+                    {
+                        await Task.Delay(TimeSpan.FromSeconds(_mqttClientConfig.ReconneTime));
+                        await _mqttClient.ConnectAsync(_options);
+                        return;
+                    }
+                    catch (Exception exp)
+                    {
+                        if (_mqttClientConfig.Exception is null)
+                        {
+                            throw exp;
+                        }
+                        _mqttClientConfig.Exception(exp);
+                    }
+                });
+            }
+        }
+
+        #endregion
+
+        #region 消息发送
+
+        /// <summary>
+        /// 发送消息,含有重连机制，如掉线会自动重连
+        /// </summary>
+        /// <param name="message"></param>
+        private async Task PublishAsync(string topicName, MqttApplicationMessageBuilder MessageBuilder, PublicQos qos = 0)
+        {
+            string topic = topicName.Trim();
+            if (string.IsNullOrEmpty(topic))
+            {
+                throw new Exception("主题不能为空！");
+            }
+            ReConnected();
+
+            MessageBuilder.WithTopic(topic).WithRetainFlag();
+            if (qos == PublicQos.Qos_0)
+            {
+                MessageBuilder.WithAtLeastOnceQoS();
+            }
+            else if (qos == PublicQos.Qos_1)
+            {
+                MessageBuilder.WithAtMostOnceQoS();
+            }
+            else
+            {
+                MessageBuilder.WithExactlyOnceQoS();
+            }
+            var Message = MessageBuilder.Build();
+            try
+            {
+                await _mqttClient.PublishAsync(Message);
+            }
+            catch (Exception e)
+            {
+                if (_mqttClientConfig.Exception is null)
+                {
+                    throw e;
+                }
+                _mqttClientConfig.Exception(e);
+            }
+
+        }
+
+        /// <summary>
+        /// 发送消息,含有重连机制，如掉线会自动重连
+        /// </summary>
+        /// <param name="message">文字消息</param>
+        public async Task Publish(string topicName, string message, PublicQos qos = 0)
+        {
+            await PublishAsync(topicName, new MqttApplicationMessageBuilder()
+                .WithPayload(message), qos);
+        }
+        /// <summary>
+        /// 发送消息,含有重连机制，如掉线会自动重连
+        /// </summary>
+        /// <param name="message">消息流</param>
+        public async void Publish(string topicName, Stream message, PublicQos qos = 0)
+            => await PublishAsync(topicName, new MqttApplicationMessageBuilder()
+                .WithPayload(message), qos);
+
+        /// <summary>
+        /// 发送消息,含有重连机制，如掉线会自动重连
+        /// </summary>
+        /// <param name="message">Byte消息</param>
+        public async void Publish(string topicName, IEnumerable<byte> message, PublicQos qos = 0)
+            => await PublishAsync(topicName, new MqttApplicationMessageBuilder()
+                .WithPayload(message), qos);
+
+        /// <summary>
+        /// 发送消息,含有重连机制，如掉线会自动重连
+        /// </summary>
+        /// <param name="message">Byte消息</param>
+        public async void Publish(string topicName, byte[] message, PublicQos qos = 0)
+            => await PublishAsync(topicName, new MqttApplicationMessageBuilder()
+                .WithPayload(message), qos);
+
+        #endregion
+
+        /// <summary>
+        /// 启动服务
+        /// </summary>
+        /// <returns></returns>
+        public async Task Start()
+            => await _mqttClient.ConnectAsync(_options);
+
+        /// <summary>
+        /// 停止服务
+        /// </summary>
+        /// <returns></returns>
+        public async Task Stop()
+            => await _mqttClient.DisconnectAsync(new MqttClientDisconnectOptions { ReasonCode = MqttClientDisconnectReason.NormalDisconnection }, CancellationToken.None);
+
+    }
+    public class MqttClientConfig
+    {
+        private string _serverIp;
+        /// <summary>
+        /// 服务器IP
+        /// </summary>
+        public string ServerIp
+        {
+            get => _serverIp;
+            set
+            {
+                if (string.IsNullOrEmpty(value.Trim()))
+                {
+                    throw new ArgumentException("ServerIp can't be null or empty!");
+                }
+                _serverIp = value;
+            }
+        }
+        private int _port;
+        /// <summary>
+        /// 服务器端口
+        /// </summary>
+        public int Port
+        {
+            get => _port;
+            set
+            {
+                if (value <= 0)
+                {
+                    throw new ArgumentException("Port can't below the zero!");
+                }
+                _port = value;
+            }
+        }
+
+        /// <summary>
+        /// 用户名
+        /// </summary>
+        public string UserName { get; set; }
+        /// <summary>
+        /// 密码
+        /// </summary>
+        public string Password { get; set; }
+
+        private string _clientId;
+        /// <summary>
+        /// 唯一用户ID，默认使用Guid
+        /// </summary>
+        public string ClientId
+        {
+            get
+            {
+                _clientId = _clientId ?? Guid.NewGuid().ToString();
+                return _clientId;
+            }
+            set => _clientId = value;
+        }
+        /// <summary>
+        /// 客户端掉线重连时间，单位/s，默认5s
+        /// </summary>
+        public double ReconneTime { get; set; } = 5;
+        /// <summary>
+        /// 异常回调，默认为空，为空抛异常
+        /// </summary>
+        public Action<Exception> Exception = null;
+
+        /// <summary>
+        /// 接收消息回调，默认不接收
+        /// </summary>
+        public Action<MqttClientReciveMsg> ReciveMsgCallback = null;
+    }
+    public class MqttClientReciveMsg
+    {
+        /// <summary>
+        /// 主题
+        /// </summary>
+        public string Topic { get; set; }
+        /// <summary>
+        /// UTF-8格式下的 负载/消息
+        /// </summary>
+        public string Payload_UTF8 { get; set; }
+        /// <summary>
+        /// 原始 负载/消息
+        /// </summary>
+        public byte[] Payload { get; set; }
+        /// <summary>
+        /// Qos
+        /// </summary>
+        public MqttQualityOfServiceLevel Qos { get; set; }
+        /// <summary>
+        /// 保留
+        /// </summary>
+        public bool Retain { get; set; }
+    }
+    public enum PublicQos
+    {
+        /// <summary>
+        /// //At most once，至多一次
+        /// </summary>
+        Qos_0,
+        /// <summary>
+        /// //At least once，至少一次
+        /// </summary>
+        Qos_1,
+        /// <summary>
+        /// //QoS2，Exactly once
+        /// </summary>
+        Qos_2,
+    }
+}

Tests/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs

@@ -1,4 +1,4 @@
-using Microsoft.AspNetCore.Builder;
+using Microsoft.AspNetCore.Builder;
 using Microsoft.AspNetCore.Connections;
 using Microsoft.AspNetCore.Hosting;
 using Microsoft.Extensions.DependencyInjection;
@@ -18,6 +18,7 @@ using System.Net;
 using MQTTnet.AspNetCore.Extensions;
 using MQTTnet.Protocol;
 using MQTTnet.Tests.Extensions;
+using MQTT;
 
 namespace MQTTnet.AspNetCore.Tests
 {
@@ -27,15 +28,42 @@ namespace MQTTnet.AspNetCore.Tests
         [TestMethod]
         public async Task TestReceivePacketAsyncThrowsWhenReaderCompleted()
         {
-            var serializer = new MqttPacketFormatterAdapter(MqttProtocolVersion.V311);
-            var pipe = new DuplexPipeMockup();
-            var connection = new DefaultConnectionContext();
-            connection.Transport = pipe;
-            var ctx = new MqttConnectionContext(serializer, connection);
-
-            pipe.Receive.Writer.Complete();
+          var client=  MqttClient.Instance(s =>
+            {
+                s.Port = 1883;
+                s.ServerIp = "10.2.1.21";
+                s.UserName = "mqtt-test";
+                s.Password = "mqtt-test";
+                s.ReciveMsgCallback = sqq =>
+                {
+                    Console.WriteLine(sqq.Payload_UTF8);
+                };
+            }, true);
+            //var client = new MqttClient(s =>
+            //  {
+            //      s.Port = 1883;
+            //      s.ServerIp = "10.2.1.21";
+            //      s.UserName = "mqtt-test";
+            //      s.Password = "mqtt-test";
+            //      s.ReciveMsgCallback = sqq =>
+            //      {
+            //          Console.WriteLine(sqq.Payload_UTF8);
+            //      };
+            //  }, true);
+
+            client.Subscribe("/TopicName/");
+
+
+            //var serializer = new MqttPacketFormatterAdapter(MqttProtocolVersion.V311);
+            //var pipe = new DuplexPipeMockup();
+            //var connection = new DefaultConnectionContext();
+            //connection.Transport = pipe;
+            //var ctx = new MqttConnectionContext(serializer, connection);
+
+            //pipe.Receive.Writer.Complete();
+
+            //await Assert.ThrowsExceptionAsync<MqttCommunicationException>(() => ctx.ReceivePacketAsync(CancellationToken.None));
 
-            await Assert.ThrowsExceptionAsync<MqttCommunicationException>(() => ctx.ReceivePacketAsync(CancellationToken.None));
         }
 
         [TestMethod]

Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs

@@ -1,4 +1,4 @@
-using Microsoft.VisualStudio.TestTools.UnitTesting;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
 using MQTTnet.Client;
 using MQTTnet.Client.Options;
 using MQTTnet.Server;
@@ -120,7 +120,7 @@ namespace MQTTnet.Tests.Mockups
         {
             if (options == null) throw new ArgumentNullException(nameof(options));
 
-            options = options.WithTcpServer("127.0.0.1", ServerPort);
+            options = options.WithTcpServer("10.2.1.21", 1883);
 
             var client = CreateClient();
             await client.ConnectAsync(options.Build()).ConfigureAwait(false);

Tests/MQTTnet.Core.Tests/Server/User_Properties_Tests.cs

@@ -1,4 +1,4 @@
-using System.Collections.Generic;
+using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
@@ -41,7 +41,9 @@ namespace MQTTnet.Tests.Server
                 }, CancellationToken.None);
 
                 MqttApplicationMessage receivedMessage = null;
-                receiver.UseApplicationMessageReceivedHandler(e => receivedMessage = e.ApplicationMessage);
+                receiver.UseApplicationMessageReceivedHandler(e =>
+                
+                receivedMessage = e.ApplicationMessage);
 
                 await sender.PublishAsync(message.Build(), CancellationToken.None);
 

Tests/MQTTnet.TestApp.NetCore/ClientTest.cs

@@ -1,4 +1,4 @@
-using MQTTnet.Client;
+using MQTTnet.Client;
 using MQTTnet.Client.Connecting;
 using MQTTnet.Client.Disconnecting;
 using MQTTnet.Client.Options;
@@ -26,7 +26,7 @@ namespace MQTTnet.TestApp.NetCore
                 {
                     ChannelOptions = new MqttClientTcpOptions
                     {
-                        Server = "127.0.0.1"
+                        Server = "10.2.1.21"
                     }
                 };