|
|
@@ -79,6 +79,7 @@ namespace YSAI.Mqtt.service.websocket
|
|
|
public MqttWebSocketServiceOperate(MqttWebSocketServiceData.Basics basics)
|
|
|
{
|
|
|
mqttServiceData_Static = this.basics = basics;
|
|
|
+ OnEventHandler_Static = OnEventHandler;
|
|
|
}
|
|
|
|
|
|
/// <summary>
|
|
|
@@ -99,6 +100,11 @@ namespace YSAI.Mqtt.service.websocket
|
|
|
";
|
|
|
}
|
|
|
|
|
|
+
|
|
|
+ public static event EventHandler<EventResult> OnEventHandler_Static;
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
public void Dispose()
|
|
|
{
|
|
|
GC.Collect();
|
|
|
@@ -106,20 +112,6 @@ namespace YSAI.Mqtt.service.websocket
|
|
|
ThisObjList.Remove(this);
|
|
|
}
|
|
|
|
|
|
- /// <summary>
|
|
|
- /// 信息传递
|
|
|
- /// </summary>
|
|
|
- public event EventHandler<EventResult> OnEvent
|
|
|
- {
|
|
|
- add { OnEventHandler += value; }
|
|
|
- remove { OnEventHandler -= value; }
|
|
|
- }
|
|
|
-
|
|
|
- /// <summary>
|
|
|
- /// 信息传递
|
|
|
- /// </summary>
|
|
|
- public static EventHandler<EventResult>? OnEventHandler;
|
|
|
-
|
|
|
public Task<OperateResult> OnAsync()
|
|
|
{
|
|
|
return Task.Run(() => On());
|
|
|
@@ -151,7 +143,6 @@ namespace YSAI.Mqtt.service.websocket
|
|
|
// MQTT WEBSOCKET 服务端口
|
|
|
o.ListenAnyIP(mqttServiceData_Static.WsPort);
|
|
|
});
|
|
|
-
|
|
|
webBuilder.UseStartup<Startup>();
|
|
|
}).RunConsoleAsync();
|
|
|
return Break("On", true);
|
|
|
@@ -161,6 +152,77 @@ namespace YSAI.Mqtt.service.websocket
|
|
|
return Break("On", false, ex.Message, Exception: ex);
|
|
|
}
|
|
|
}
|
|
|
+ /// <summary>
|
|
|
+ /// 启动
|
|
|
+ /// </summary>
|
|
|
+ private sealed class Startup
|
|
|
+ {
|
|
|
+ /// <summary>
|
|
|
+ /// 配置
|
|
|
+ /// </summary>
|
|
|
+ public void Configure(IApplicationBuilder app, IWebHostEnvironment environment, MqttController mqttController)
|
|
|
+ {
|
|
|
+ try
|
|
|
+ {
|
|
|
+ app.UseRouting();
|
|
|
+
|
|
|
+ app.UseEndpoints(
|
|
|
+ endpoints =>
|
|
|
+ {
|
|
|
+ endpoints.MapConnectionHandler<MqttConnectionHandler>(
|
|
|
+ mqttServiceData_Static.Uri, //指定地址
|
|
|
+ httpConnectionDispatcherOptions => httpConnectionDispatcherOptions.WebSockets.SubProtocolSelector = protocolList => protocolList.FirstOrDefault() ?? string.Empty);
|
|
|
+ });
|
|
|
+
|
|
|
+ app.UseMqttServer(
|
|
|
+ server =>
|
|
|
+ {
|
|
|
+ //身份验证(异步)
|
|
|
+ server.ValidatingConnectionAsync += mqttController.MqttServer_ValidatingConnectionAsync;
|
|
|
+ //消息接收(异步)
|
|
|
+ server.ApplicationMessageNotConsumedAsync += mqttController.MqttServer_ApplicationMessageNotConsumedAsync;
|
|
|
+ //客户端连接(异步)
|
|
|
+ server.ClientConnectedAsync += mqttController.MqttServer_ClientConnectedAsync;
|
|
|
+ //客户端断开(异步)
|
|
|
+ server.ClientDisconnectedAsync += mqttController.MqttServer_ClientDisconnectedAsync;
|
|
|
+ //客户端订阅事件(异步)
|
|
|
+ server.ClientSubscribedTopicAsync += mqttController.MqttServer_ClientSubscribedTopicAsync;
|
|
|
+ //客户端取消订阅(异步)
|
|
|
+ server.ClientUnsubscribedTopicAsync += mqttController.MqttServer_ClientUnsubscribedTopicAsync;
|
|
|
+ });
|
|
|
+
|
|
|
+ OnEventHandler_Static?.Invoke(this, new EventResult(true, $"[ {Steps.系统信息} ]MQTT 服务端地址 ( 127.0.0.1:{mqttServiceData_Static.Port} )"));
|
|
|
+ OnEventHandler_Static?.Invoke(this, new EventResult(true, $"[ {Steps.系统信息} ]MQTT WebSocket 服务端地址 ( ws://127.0.0.1:{mqttServiceData_Static.WsPort}/{mqttServiceData_Static.Uri} )"));
|
|
|
+ }
|
|
|
+ catch (Exception ex)
|
|
|
+ {
|
|
|
+ OnEventHandler_Static?.Invoke(this, new EventResult(false, $"[ {Steps.系统异常信息} ]配置服务异常:{ex.Message}"));
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /// <summary>
|
|
|
+ /// 配置服务
|
|
|
+ /// </summary>
|
|
|
+ /// <param name="services"></param>
|
|
|
+ public void ConfigureServices(IServiceCollection services)
|
|
|
+ {
|
|
|
+ try
|
|
|
+ {
|
|
|
+ services.AddHostedMqttServer(
|
|
|
+ optionsBuilder =>
|
|
|
+ {
|
|
|
+ optionsBuilder.WithDefaultEndpoint();
|
|
|
+ });
|
|
|
+ services.AddMqttConnectionHandler();
|
|
|
+ services.AddConnections();
|
|
|
+ services.AddSingleton<MqttController>();
|
|
|
+ }
|
|
|
+ catch (Exception ex)
|
|
|
+ {
|
|
|
+ OnEventHandler_Static?.Invoke(this, new EventResult(false, $"[ {Steps.系统异常信息} ]配置服务异常:{ex.Message}"));
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
|
|
|
/// <summary>
|
|
|
/// MQTT控制器
|
|
|
@@ -174,7 +236,7 @@ namespace YSAI.Mqtt.service.websocket
|
|
|
{
|
|
|
return Task.Run(() =>
|
|
|
{
|
|
|
- OnEventHandler(this, new EventResult(true, $"[ {Steps.客户端取消订阅事件} ]( {arg.ClientId} ) 取消了 ( {arg.TopicFilter} ) 的订阅", arg.ClientId, Core.@enum.ResultType.String));
|
|
|
+ OnEventHandler_Static?.Invoke(this, new EventResult(true, $"[ {Steps.客户端取消订阅事件} ]( {arg.ClientId} ) 取消了 ( {arg.TopicFilter} ) 的订阅", arg.ClientId, Core.@enum.ResultType.String));
|
|
|
});
|
|
|
}
|
|
|
|
|
|
@@ -185,7 +247,7 @@ namespace YSAI.Mqtt.service.websocket
|
|
|
{
|
|
|
return Task.Run(() =>
|
|
|
{
|
|
|
- OnEventHandler(this, new EventResult(true, $"[ {Steps.客户端订阅事件} ]( {arg.ClientId} ) 订阅了主题 ( {arg.TopicFilter.Topic} )", arg.ClientId, Core.@enum.ResultType.String));
|
|
|
+ OnEventHandler_Static?.Invoke(this, new EventResult(true, $"[ {Steps.客户端订阅事件} ]( {arg.ClientId} ) 订阅了主题 ( {arg.TopicFilter.Topic} )", arg.ClientId, Core.@enum.ResultType.String));
|
|
|
});
|
|
|
}
|
|
|
|
|
|
@@ -205,7 +267,7 @@ namespace YSAI.Mqtt.service.websocket
|
|
|
dynamic DynamicObj = new ExpandoObject();
|
|
|
DynamicObj.Messages = arg.ApplicationMessage;
|
|
|
DynamicObj.SenderID = arg.SenderId;
|
|
|
- OnEventHandler(this, new EventResult(true, $"[ {Steps.客户端消息事件} ]( {arg.SenderId} ) 发布了主题 ( {arg.ApplicationMessage.Topic} ) 内容 ( {Encoding.UTF8.GetString(arg.ApplicationMessage.Payload)} )", DynamicObj, Core.@enum.ResultType.Dynamic));
|
|
|
+ OnEventHandler_Static?.Invoke(this, new EventResult(true, $"[ {Steps.客户端消息事件} ]( {arg.SenderId} ) 发布了主题 ( {arg.ApplicationMessage.Topic} ) 内容 ( {Encoding.UTF8.GetString(arg.ApplicationMessage.Payload)} )", DynamicObj, Core.@enum.ResultType.Dynamic));
|
|
|
break;
|
|
|
}
|
|
|
});
|
|
|
@@ -218,7 +280,7 @@ namespace YSAI.Mqtt.service.websocket
|
|
|
{
|
|
|
return Task.Run(() =>
|
|
|
{
|
|
|
- OnEventHandler(this, new EventResult(true, $"[ {Steps.客户端断开事件} ]( {arg.ClientId} ) 已断开", arg.ClientId, Core.@enum.ResultType.String));
|
|
|
+ OnEventHandler_Static?.Invoke(this, new EventResult(true, $"[ {Steps.客户端断开事件} ]( {arg.ClientId} ) 已断开", arg.ClientId, Core.@enum.ResultType.String));
|
|
|
});
|
|
|
}
|
|
|
|
|
|
@@ -229,7 +291,7 @@ namespace YSAI.Mqtt.service.websocket
|
|
|
{
|
|
|
return Task.Run(() =>
|
|
|
{
|
|
|
- OnEventHandler(this, new EventResult(true, $"[ {Steps.客户端连接事件} ]( {arg.ClientId} ) 已连接", arg.ClientId, Core.@enum.ResultType.String));
|
|
|
+ OnEventHandler_Static?.Invoke(this, new EventResult(true, $"[ {Steps.客户端连接事件} ]( {arg.ClientId} ) 已连接", arg.ClientId, Core.@enum.ResultType.String));
|
|
|
});
|
|
|
}
|
|
|
|
|
|
@@ -244,90 +306,18 @@ namespace YSAI.Mqtt.service.websocket
|
|
|
return Task.Run((Action)(() =>
|
|
|
{
|
|
|
//验证账号密码是否正确
|
|
|
- if (!arg.UserName.Equals((string?)mqttServiceData_Static.UserName) || !arg.Password.Equals(mqttServiceData_Static.Password))
|
|
|
+ if (arg.UserName != mqttServiceData_Static.UserName || arg.Password != mqttServiceData_Static.Password)
|
|
|
{
|
|
|
arg.ReasonCode = MQTTnet.Protocol.MqttConnectReasonCode.BadUserNameOrPassword;
|
|
|
- OnEventHandler(this, new EventResult(false, $"[ {Steps.客户端身份验证事件} ]( {arg.ClientId} ) 身份验证异常:{arg.ReasonCode}", arg.ClientId, Core.@enum.ResultType.String));
|
|
|
+ OnEventHandler_Static?.Invoke(this, new EventResult(false, $"[ {Steps.客户端身份验证事件} ]( {arg.ClientId} ) 身份验证异常:{arg.ReasonCode}", arg.ClientId, Core.@enum.ResultType.String));
|
|
|
}
|
|
|
else
|
|
|
{
|
|
|
arg.ReasonCode = MQTTnet.Protocol.MqttConnectReasonCode.Success;
|
|
|
- OnEventHandler(this, new EventResult(true, $"[ {Steps.客户端身份验证事件} ]( {arg.ClientId} ) 身份验证成功", arg.ClientId, Core.@enum.ResultType.String));
|
|
|
+ OnEventHandler_Static?.Invoke(this, new EventResult(true, $"[ {Steps.客户端身份验证事件} ]( {arg.ClientId} ) 身份验证成功", arg.ClientId, Core.@enum.ResultType.String));
|
|
|
}
|
|
|
}));
|
|
|
}
|
|
|
}
|
|
|
-
|
|
|
- /// <summary>
|
|
|
- /// 启动
|
|
|
- /// </summary>
|
|
|
- private sealed class Startup
|
|
|
- {
|
|
|
- /// <summary>
|
|
|
- /// 配置
|
|
|
- /// </summary>
|
|
|
- public void Configure(IApplicationBuilder app, IWebHostEnvironment environment, MqttController mqttController)
|
|
|
- {
|
|
|
- try
|
|
|
- {
|
|
|
- app.UseRouting();
|
|
|
-
|
|
|
- app.UseEndpoints(
|
|
|
- endpoints =>
|
|
|
- {
|
|
|
- endpoints.MapConnectionHandler<MqttConnectionHandler>(
|
|
|
- mqttServiceData_Static.Uri, //指定地址
|
|
|
- httpConnectionDispatcherOptions => httpConnectionDispatcherOptions.WebSockets.SubProtocolSelector = protocolList => protocolList.FirstOrDefault() ?? string.Empty);
|
|
|
- });
|
|
|
-
|
|
|
- app.UseMqttServer(
|
|
|
- server =>
|
|
|
- {
|
|
|
- //身份验证(异步)
|
|
|
- server.ValidatingConnectionAsync += mqttController.MqttServer_ValidatingConnectionAsync;
|
|
|
- //消息接收(异步)
|
|
|
- server.ApplicationMessageNotConsumedAsync += mqttController.MqttServer_ApplicationMessageNotConsumedAsync;
|
|
|
- //客户端连接(异步)
|
|
|
- server.ClientConnectedAsync += mqttController.MqttServer_ClientConnectedAsync;
|
|
|
- //客户端断开(异步)
|
|
|
- server.ClientDisconnectedAsync += mqttController.MqttServer_ClientDisconnectedAsync;
|
|
|
- //客户端订阅事件(异步)
|
|
|
- server.ClientSubscribedTopicAsync += mqttController.MqttServer_ClientSubscribedTopicAsync;
|
|
|
- //客户端取消订阅(异步)
|
|
|
- server.ClientUnsubscribedTopicAsync += mqttController.MqttServer_ClientUnsubscribedTopicAsync;
|
|
|
- });
|
|
|
-
|
|
|
- OnEventHandler(this, new EventResult(true, $"[ {Steps.系统信息} ]MQTT 服务端地址 ( 127.0.0.1:{mqttServiceData_Static.Port} )"));
|
|
|
- OnEventHandler(this, new EventResult(true, $"[ {Steps.系统信息} ]MQTT WebSocket 服务端地址 ( ws://127.0.0.1:{mqttServiceData_Static.WsPort}/{mqttServiceData_Static.Uri} )"));
|
|
|
- }
|
|
|
- catch (Exception ex)
|
|
|
- {
|
|
|
- OnEventHandler(this, new EventResult(false, $"[ {Steps.系统异常信息} ]配置异常:{ex.Message}"));
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- /// <summary>
|
|
|
- /// 配置服务
|
|
|
- /// </summary>
|
|
|
- /// <param name="services"></param>
|
|
|
- public void ConfigureServices(IServiceCollection services)
|
|
|
- {
|
|
|
- try
|
|
|
- {
|
|
|
- services.AddHostedMqttServer(
|
|
|
- optionsBuilder =>
|
|
|
- {
|
|
|
- optionsBuilder.WithDefaultEndpoint();
|
|
|
- });
|
|
|
- services.AddMqttConnectionHandler();
|
|
|
- services.AddConnections();
|
|
|
- services.AddSingleton<MqttController>();
|
|
|
- }
|
|
|
- catch (Exception ex)
|
|
|
- {
|
|
|
- OnEventHandler(this, new EventResult(false, $"[ {Steps.系统异常信息} ]配置服务异常:{ex.Message}"));
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
}
|
|
|
}
|