1.增加websocket消息推送demo

2.添加缺失的配置文件
3.修改文档

.gitignore

@@ -328,6 +328,5 @@ ASALocalRun/
 
 # MFractors (Xamarin productivity tool) working folder 
 .mfractor/
-*.json
 /nupkgs
 /src/JT808.DotNetty.Admin/tools/protoc-gen-grpc-web-1.0.3-windows-x86_64.exe

README.md

@@ -128,7 +128,10 @@ Pipeline分为两种方式使用，一种是使用队列的方式，一种是网
 
 3.JT808.Gateway.SimpleQueueService项目下的Debug目录运行消息处理服务
 
-4.进入JT808.Gateway.SimpleClient项目下的Debug目录运行客户端
+4.JT808.Gateway.SimpleQueueNotification项目下的Debug目录运行WebSocket服务
+从浏览器中打开localhost:5000查看数据
+
+5.进入JT808.Gateway.SimpleClient项目下的Debug目录运行客户端
 
 > 注意：需要安装kafka和zookeeper
 

simples/JT808.Gateway.SimpleQueueNotification/Configs/AuthOptions.cs

@@ -0,0 +1,12 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+
+namespace JT808.Gateway.SimpleQueueNotification.Configs
+{
+    public class AuthOptions
+    {
+        public string SuperToken { get; set; }
+    }
+}

simples/JT808.Gateway.SimpleQueueNotification/Hubs/JT808MsgHub.cs

@@ -0,0 +1,15 @@
+using Microsoft.AspNetCore.SignalR;
+using System.Threading.Tasks;
+
+
+namespace JT808.Gateway.SimpleQueueNotification.Hubs
+{
+    public class JT808MsgHub : Hub
+    {
+        public Task ReceiveMessage(string token, string msg)
+		{
+			Clients.Caller.SendAsync("ReceiveMessage", token, "Heartbeat");
+			return Task.CompletedTask;
+		}
+	}
+}

simples/JT808.Gateway.SimpleQueueNotification/Impl/JT808MsgIdHandlerImpl.cs

@@ -0,0 +1,44 @@
+using JT808.Gateway.MsgIdHandler;
+using JT808.Gateway.SimpleQueueNotification.Hubs;
+using JT808.Protocol.Extensions;
+using Microsoft.AspNetCore.SignalR;
+using Microsoft.Extensions.Logging;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text.Json;
+using System.Threading.Tasks;
+
+namespace JT808.Gateway.SimpleQueueNotification.Impl
+{
+    public class JT808MsgIdHandlerImpl : IJT808MsgIdHandler
+    {
+        private readonly ILogger<JT808MsgIdHandlerImpl> logger;
+
+        private readonly IHubContext<JT808MsgHub> _hubContext;
+
+        public JT808MsgIdHandlerImpl(
+            ILoggerFactory loggerFactory,
+            IHubContext<JT808MsgHub> hubContext
+            )
+        {
+            this._hubContext = hubContext;
+            logger = loggerFactory.CreateLogger<JT808MsgIdHandlerImpl>();
+        }
+        public void Processor((string TerminalNo, byte[] Data) parameter)
+        {
+            try
+            {
+                if (logger.IsEnabled(LogLevel.Trace))
+                {
+                    logger.LogTrace($"{parameter.TerminalNo}-{parameter.Data.ToHexString()}");
+                }
+               _hubContext.Clients.All.SendAsync("ReceiveMessage", parameter.TerminalNo, parameter.Data.ToHexString());
+            }
+            catch (Exception ex)
+            {
+                logger.LogError(ex, "");
+            }
+        }
+    }
+}

simples/JT808.Gateway.SimpleQueueNotification/JT808.Gateway.SimpleQueueNotification.csproj

@@ -0,0 +1,33 @@
+<Project Sdk="Microsoft.NET.Sdk.Web">
+
+  <PropertyGroup>
+    <TargetFramework>netcoreapp3.1</TargetFramework>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <Content Remove="appsettings.Development.json" />
+    <Content Remove="appsettings.json" />
+    <Content Remove="wwwroot\index.html" />
+    <Content Remove="wwwroot\signalr.min.js" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <None Include="appsettings.Development.json">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </None>
+    <None Include="appsettings.json">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </None>
+    <None Include="wwwroot\index.html">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </None>
+    <None Include="wwwroot\signalr.min.js">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </None>
+  </ItemGroup>
+
+  <ItemGroup>
+    <PackageReference Include="JT808.Gateway.Kafka" Version="1.0.0-preview8" />
+    <PackageReference Include="JT808.Gateway.MsgIdHandler" Version="1.0.0-preview8" />
+  </ItemGroup>
+</Project>

simples/JT808.Gateway.SimpleQueueNotification/Middlewares/JT808Jwtiddleware.cs

@@ -0,0 +1,105 @@
+using JT808.Gateway.SimpleQueueNotification.Configs;
+using Microsoft.AspNetCore.Http;
+using Microsoft.AspNetCore.SignalR;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Net;
+using System.Security.Claims;
+using System.Security.Principal;
+using System.Text;
+using System.Text.Json;
+using System.Threading.Tasks;
+
+namespace JT808.Gateway.SimpleQueueNotification.Middlewares
+{
+    public class JT808JwtMiddlewares
+    {
+        private readonly RequestDelegate next;
+
+        private readonly ILogger logger;
+
+        IOptionsMonitor<AuthOptions> authOptionsMonitor;
+
+        public JT808JwtMiddlewares(RequestDelegate next,
+            IOptionsMonitor<AuthOptions> authOptionsMonitor,
+            ILoggerFactory loggerFactory)
+        {
+            this.next = next;
+            this.authOptionsMonitor = authOptionsMonitor;
+            logger = loggerFactory.CreateLogger<JT808JwtMiddlewares>();
+        }
+
+        public async Task Invoke(HttpContext context)
+        {
+            StringBuilder sb = new StringBuilder();
+            sb.Append(getIp(context));
+            sb.Append(",");
+            sb.Append(getBrowser(context));
+            sb.Append(",");
+            if (context.Request.Query.TryGetValue("access_token", out var token))
+            {
+                if (token == "")
+                {
+                    context.Response.StatusCode = (int)HttpStatusCode.OK;
+                    await context.Response.WriteAsync(JsonSerializer.Serialize(new { Code = (int)HttpStatusCode.Unauthorized, Msg = HttpStatusCode.Unauthorized.ToString() }));
+                    return;
+                }
+                else
+                {
+                    sb.Append(token);
+                }
+                try
+                {
+                    // 去认证服务器请求结果
+                    // 如果超级Token有值就不去jwt认证
+                    if (!string.IsNullOrEmpty(authOptionsMonitor.CurrentValue.SuperToken))
+                    {
+                        if(authOptionsMonitor.CurrentValue.SuperToken != token)
+                        {
+                            context.Response.StatusCode = (int)HttpStatusCode.OK;
+                            await context.Response.WriteAsync(JsonSerializer.Serialize(new { Code = (int)HttpStatusCode.Unauthorized, Msg = HttpStatusCode.Unauthorized.ToString() }));
+                            return;
+                        }
+                    }
+                }
+                catch (Exception ex)
+                {
+                    logger.LogError(ex, ex.Message);
+                    context.Response.StatusCode = (int)HttpStatusCode.OK;
+                    await context.Response.WriteAsync(JsonSerializer.Serialize(new { Code = (int)HttpStatusCode.ExpectationFailed, Msg = HttpStatusCode.ExpectationFailed.ToString() }));
+                    return;
+                }
+                if(logger.IsEnabled(LogLevel.Debug))
+                    logger.LogDebug($"认证成功:{sb.ToString()}");
+                await next(context);
+            }
+            else
+            {
+                context.Response.StatusCode = (int)HttpStatusCode.OK;
+                await context.Response.WriteAsync(JsonSerializer.Serialize(new { Code = (int)HttpStatusCode.Unauthorized, Msg = HttpStatusCode.Unauthorized.ToString() }));
+            }
+        }
+
+        private string getIp(HttpContext context)
+        {
+            Microsoft.Extensions.Primitives.StringValues ips;
+            if (context.Request.Headers.TryGetValue("X-Real-IP", out ips))
+            {
+                return ips.FirstOrDefault() ?? "";
+            }
+            else
+            {
+                return context.Connection.RemoteIpAddress?.ToString() ?? "";
+            }
+        }
+
+        private static string getBrowser(HttpContext context)
+        {
+            return context.Request.Headers["User-Agent"].FirstOrDefault();
+        }
+    }
+}

simples/JT808.Gateway.SimpleQueueNotification/Middlewares/JT808JwtiddlewareExtensions.cs

@@ -0,0 +1,14 @@
+using Microsoft.AspNetCore.Builder;
+
+
+namespace JT808.Gateway.SimpleQueueNotification.Middlewares
+{
+    public static class JT808JwtiddlewareExtensions
+    {
+        public static IApplicationBuilder UseJT808JwtVerify(
+            this IApplicationBuilder builder)
+        {
+            return builder.UseMiddleware<JT808JwtMiddlewares>();
+        }
+    }
+}

simples/JT808.Gateway.SimpleQueueNotification/Program.cs

@@ -0,0 +1,26 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using Microsoft.AspNetCore.Hosting;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+
+namespace JT808.Gateway.SimpleQueueNotification
+{
+    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>();
+                });
+    }
+}

simples/JT808.Gateway.SimpleQueueNotification/Startup.cs

@@ -0,0 +1,79 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using JT808.Gateway.SimpleQueueNotification.Configs;
+using JT808.Gateway.SimpleQueueNotification.Hubs;
+using JT808.Gateway.SimpleQueueNotification.Middlewares;
+using Microsoft.AspNetCore.Builder;
+using Microsoft.AspNetCore.Hosting;
+using Microsoft.AspNetCore.Http;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using JT808.Protocol;
+using JT808.Gateway.Kafka;
+using JT808.Gateway.MsgIdHandler;
+using JT808.Gateway.SimpleQueueNotification.Impl;
+
+namespace JT808.Gateway.SimpleQueueNotification
+{
+    public class Startup
+    {
+        public Startup(IHostEnvironment env)
+        {
+            var builder = new ConfigurationBuilder()
+                           .SetBasePath(AppDomain.CurrentDomain.BaseDirectory)
+                           .AddJsonFile("appsettings.json", optional: true, reloadOnChange: true)
+                           .AddJsonFile($"appsettings.{env.EnvironmentName}.json", optional: true, reloadOnChange: true)
+                           .AddEnvironmentVariables();
+            Configuration = builder.Build();
+        }
+
+        public IConfiguration Configuration { get; }
+        // This method gets called by the runtime. Use this method to add services to the container.
+        // For more information on how to configure your application, visit https://go.microsoft.com/fwlink/?LinkID=398940
+        public void ConfigureServices(IServiceCollection services)
+        {
+            services.AddSingleton<ILoggerFactory, LoggerFactory>();
+            services.AddSingleton(typeof(ILogger<>), typeof(Logger<>));
+            services.AddLogging((configure) => {
+                configure.AddConsole();
+                configure.SetMinimumLevel(LogLevel.Trace);
+            });
+            services.AddSignalR();
+            services.AddCors(options => options.AddPolicy("CorsPolicy", builder =>
+            {
+                //builder.AllowAnyOrigin()
+                //        .AllowAnyMethod()
+                //        .AllowAnyHeader()
+                //        .AllowAnyOrigin();
+                builder.AllowAnyMethod()
+                        .AllowAnyHeader()
+                        .WithOrigins(Configuration.GetSection("CorsHosts").Get<List<string>>().ToArray())
+                        .AllowCredentials();
+            }));
+            services.Configure<AuthOptions>(Configuration.GetSection("AuthOptions"));
+            services.AddJT808Configure()
+                    .AddClientKafka()
+                    .AddMsgConsumer(Configuration)
+                    .AddMsgIdHandler<JT808MsgIdHandlerImpl>();
+        }
+
+
+        // This method gets called by the runtime. Use this method to configure the HTTP request pipeline.
+        public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
+        {
+            app.UseDefaultFiles();
+            app.UseStaticFiles();
+            app.UseRouting();
+            app.UseJT808JwtVerify();
+            app.UseCors("CorsPolicy");
+            app.UseEndpoints(endpoints =>
+            {
+                endpoints.MapHub<JT808MsgHub>("/JT808MsgHub");
+            });
+        }
+    }
+}

simples/JT808.Gateway.SimpleQueueNotification/appsettings.Development.json

@@ -0,0 +1,24 @@
+{
+  "Logging": {
+    "LogLevel": {
+      "Default": "Information",
+      "Microsoft": "Warning",
+      "Microsoft.Hosting.Lifetime": "Information"
+    }
+  },
+  "AuthOptions": {
+    "SuperToken": "123456"
+  },
+  "JT808MsgConsumerConfig": {
+    "TopicName": "JT808Msg",
+    "EnableAutoCommit": true,
+    "GroupId": "JT808.Notice",
+    "BootstrapServers": "127.0.0.1:9092"
+  },
+  "CorsHosts": [
+    "http://localhost:8055",
+    "http://127.0.0.1:5500",
+    "http://localhost:81",
+    "http://127.0.0.1:81"
+  ]
+}

simples/JT808.Gateway.SimpleQueueNotification/appsettings.json

@@ -0,0 +1,25 @@
+{
+  "Logging": {
+    "LogLevel": {
+      "Default": "Information",
+      "Microsoft": "Warning",
+      "Microsoft.Hosting.Lifetime": "Information"
+    }
+  },
+  "AllowedHosts": "*",
+  "AuthOptions": {
+    "SuperToken": "123456"
+  },
+  "JT808MsgConsumerConfig": {
+    "TopicName": "JT808Msg",
+    "EnableAutoCommit": true,
+    "GroupId": "JT808.Notice",
+    "BootstrapServers": "127.0.0.1:9092"
+  },
+  "CorsHosts": [
+    "http://localhost:8055",
+    "http://127.0.0.1:5500",
+    "http://localhost:81",
+    "http://127.0.0.1:81"
+  ]
+}

simples/JT808.Gateway.SimpleQueueNotification/wwwroot/index.html

@@ -0,0 +1,31 @@
+<!DOCTYPE html>
+<html>
+
+<head>
+    <meta charset="utf-8" />
+    <meta name="viewport" content="width=device-width,initial-scale=1.0" />
+    <meta http-equiv="pragma" content="no-cache" />
+    <meta http-equiv="Cache-Control" content="no-cache, must-revalidate" />
+    <meta http-equiv="expires" content="Wed, 26 Feb 1997 08:21:57 GMT" />
+    <meta http-equiv="Expires" content="-1" />
+    <script type="text/javascript" src="signalr.min.js"></script>
+</head>
+
+<body>
+    <h1>接收消息：</h1>
+    <ul id="msg">
+
+    </ul>
+    <script>
+        var msgDom = document.getElementById("msg");
+        const connection = new signalR.HubConnectionBuilder()
+            .withUrl("http://localhost:5000/JT808MsgHub?access_token=123456")
+            .build();
+        connection.on("ReceiveMessage", (tno, message) => {
+            msgDom.append("<li>" + tno + "=>" + message + "</li>");
+        });
+        connection.start().catch(err => console.error(err));
+    </script>
+</body>
+
+</html>

simples/JT808.Gateway.SimpleQueueServer/appsettings.json

@@ -0,0 +1,35 @@
+{
+  "Logging": {
+    "IncludeScopes": false,
+    "Debug": {
+      "LogLevel": {
+        "Default": "Trace"
+      }
+    },
+    "Console": {
+      "LogLevel": {
+        "Default": "Trace"
+      }
+    }
+  },
+  "JT808Configuration": {
+    "TcpPort": 808,
+    "UdpPort": 808,
+    "WebApiPort": 828,
+    "WebApiToken": "smallchi518"
+  },
+  "JT808MsgProducerConfig": {
+    "TopicName": "JT808Msg",
+    "BootstrapServers": "127.0.0.1:9092"
+  },
+  "JT808MsgReplyConsumerConfig": {
+    "TopicName": "JT808MsgReply",
+    "GroupId": "jt808msgreply",
+    "EnableAutoCommit": true,
+    "BootstrapServers": "127.0.0.1:9092"
+  },
+  "JT808SessionProducerConfig": {
+    "TopicName": "JT808Session",
+    "BootstrapServers": "127.0.0.1:9092"
+  }
+}

simples/JT808.Gateway.SimpleQueueService/appsettings.json

@@ -0,0 +1,30 @@
+{
+  "Logging": {
+    "IncludeScopes": false,
+    "Debug": {
+      "LogLevel": {
+        "Default": "Trace"
+      }
+    },
+    "Console": {
+      "LogLevel": {
+        "Default": "Trace"
+      }
+    }
+  },
+  "JT808MsgConsumerConfig": {
+    "TopicName": "JT808Msg",
+    "EnableAutoCommit": true,
+    "GroupId": "JT808.MsgReply",
+    "BootstrapServers": "127.0.0.1:9092"
+  },
+  "JT808MsgReplyProducerConfig": {
+    "TopicName": "JT808MsgReply",
+    "BootstrapServers": "127.0.0.1:9092"
+  },
+  "JT808SessionConsumerConfig": {
+    "TopicName": "JT808Session",
+    "GroupId": "JT808.Session",
+    "BootstrapServers": "127.0.0.1:9092"
+  }
+}

simples/JT808.Simples.sln

@@ -13,9 +13,11 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "JT808.Gateway.SimpleServer"
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "JT808.Gateway.SimpleClient", "JT808.Gateway.SimpleClient\JT808.Gateway.SimpleClient.csproj", "{09AFAC3E-4E4D-4B51-962D-BF8489D8BEC6}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "JT808.Gateway.SimpleQueueServer", "JT808.Gateway.SimpleQueueServer\JT808.Gateway.SimpleQueueServer.csproj", "{8594AC7F-18B4-439D-B58B-1CEFF0833A1A}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "JT808.Gateway.SimpleQueueServer", "JT808.Gateway.SimpleQueueServer\JT808.Gateway.SimpleQueueServer.csproj", "{8594AC7F-18B4-439D-B58B-1CEFF0833A1A}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "JT808.Gateway.SimpleQueueService", "JT808.Gateway.SimpleQueueService\JT808.Gateway.SimpleQueueService.csproj", "{E2D1CFEF-417A-4C44-BC2E-E5A160602485}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "JT808.Gateway.SimpleQueueService", "JT808.Gateway.SimpleQueueService\JT808.Gateway.SimpleQueueService.csproj", "{E2D1CFEF-417A-4C44-BC2E-E5A160602485}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "JT808.Gateway.SimpleQueueNotification", "JT808.Gateway.SimpleQueueNotification\JT808.Gateway.SimpleQueueNotification.csproj", "{163D2EE2-9A62-4E8A-B203-BF147909E89A}"
 EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
@@ -47,6 +49,10 @@ Global
 		{E2D1CFEF-417A-4C44-BC2E-E5A160602485}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{E2D1CFEF-417A-4C44-BC2E-E5A160602485}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{E2D1CFEF-417A-4C44-BC2E-E5A160602485}.Release|Any CPU.Build.0 = Release|Any CPU
+		{163D2EE2-9A62-4E8A-B203-BF147909E89A}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{163D2EE2-9A62-4E8A-B203-BF147909E89A}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{163D2EE2-9A62-4E8A-B203-BF147909E89A}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{163D2EE2-9A62-4E8A-B203-BF147909E89A}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -58,6 +64,7 @@ Global
 		{09AFAC3E-4E4D-4B51-962D-BF8489D8BEC6} = {2459FB59-8A33-49A4-ADBC-A0B12C5886A6}
 		{8594AC7F-18B4-439D-B58B-1CEFF0833A1A} = {2459FB59-8A33-49A4-ADBC-A0B12C5886A6}
 		{E2D1CFEF-417A-4C44-BC2E-E5A160602485} = {2459FB59-8A33-49A4-ADBC-A0B12C5886A6}
+		{163D2EE2-9A62-4E8A-B203-BF147909E89A} = {2459FB59-8A33-49A4-ADBC-A0B12C5886A6}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {FC0FFCEA-E1EF-4C97-A1C5-F89418B6834B}