完善flv首包和其他包组包待测试

src/JT1078.Gateway.Abstractions/JT1078.Gateway.Abstractions.csproj

@@ -26,10 +26,10 @@
 
   <ItemGroup>
     <PackageReference Include="JT1078" Version="1.0.3" />
-    <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="3.1.5" />
-    <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="3.1.5" />
-    <PackageReference Include="Microsoft.Extensions.Options.ConfigurationExtensions" Version="3.1.5" />
-    <PackageReference Include="Microsoft.Extensions.Options" Version="3.1.5" />
+    <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="3.1.7" />
+    <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="3.1.7" />
+    <PackageReference Include="Microsoft.Extensions.Options.ConfigurationExtensions" Version="3.1.7" />
+    <PackageReference Include="Microsoft.Extensions.Options" Version="3.1.7" />
     <PackageReference Include="System.Text.Json" Version="4.7.2" />
   </ItemGroup>
 </Project>

src/JT1078.Gateway.Tests/JT1078.Gateway.Test/JT1078.Gateway.Test.csproj

@@ -7,10 +7,10 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.6.1" />
+    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.7.0" />
     <PackageReference Include="System.IO.Pipelines" Version="4.7.2" />
     <PackageReference Include="xunit" Version="2.4.1" />
-    <PackageReference Include="xunit.runner.visualstudio" Version="2.4.2">
+    <PackageReference Include="xunit.runner.visualstudio" Version="2.4.3">
       <PrivateAssets>all</PrivateAssets>
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
     </PackageReference>

src/JT1078.Gateway.Tests/JT1078.Gateway.TestNormalHosting/JT1078.Gateway.TestNormalHosting.csproj

@@ -6,9 +6,10 @@
   </PropertyGroup>
   
   <ItemGroup>
-    <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="3.1.5" />
-    <PackageReference Include="Microsoft.Extensions.Hosting" Version="3.1.5" />
-    <PackageReference Include="Microsoft.Extensions.Logging.Console" Version="3.1.5" />
+    <PackageReference Include="JT1078.Flv" Version="1.0.0-preview6" />
+    <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="3.1.7" />
+    <PackageReference Include="Microsoft.Extensions.Hosting" Version="3.1.7" />
+    <PackageReference Include="Microsoft.Extensions.Logging.Console" Version="3.1.7" />
   </ItemGroup>
 
   <ItemGroup>

src/JT1078.Gateway.Tests/JT1078.Gateway.TestNormalHosting/Program.cs

@@ -1,4 +1,5 @@
-using JT1078.Gateway.InMemoryMQ;
+using JT1078.Flv;
+using JT1078.Gateway.InMemoryMQ;
 using JT1078.Gateway.TestNormalHosting.Services;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
@@ -28,12 +29,13 @@ namespace JT1078.Gateway.TestNormalHosting
                 {
                     services.AddSingleton<ILoggerFactory, LoggerFactory>();
                     services.AddSingleton(typeof(ILogger<>), typeof(Logger<>));
+                    services.AddSingleton<FlvEncoder>();
                     //使用内存队列实现会话通知
                     services.AddJT1078Gateway(hostContext.Configuration)
                             .AddTcp()
                             .AddUdp()
                             .AddHttp()
-                            .AddCoordinatorHttpClient()
+                            //.AddCoordinatorHttpClient()
                             .AddNormal()
                             .AddMsgProducer()
                             .AddMsgConsumer();

src/JT1078.Gateway.Tests/JT1078.Gateway.TestNormalHosting/Services/JT1078NormalMsgHostedService.cs

@@ -1,25 +1,51 @@
 using JT1078.Gateway.Abstractions;
+using JT1078.Gateway.Sessions;
+using JT1078.Flv;
 using Microsoft.Extensions.Hosting;
 using System;
 using System.Collections.Generic;
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
+using System.Linq;
 
 namespace JT1078.Gateway.TestNormalHosting.Services
 {
     public class JT1078NormalMsgHostedService : BackgroundService
     {
         private IJT1078PackageConsumer PackageConsumer;
-        public JT1078NormalMsgHostedService(IJT1078PackageConsumer packageConsumer)
+        private JT1078HttpSessionManager HttpSessionManager;
+        private FlvEncoder FlvEncoder;
+        public JT1078NormalMsgHostedService(
+            FlvEncoder flvEncoder,
+            JT1078HttpSessionManager httpSessionManager,
+            IJT1078PackageConsumer packageConsumer)
         {
             PackageConsumer = packageConsumer;
+            HttpSessionManager = httpSessionManager;
+            FlvEncoder = flvEncoder;
         }
         protected override Task ExecuteAsync(CancellationToken stoppingToken)
         {
-            PackageConsumer.OnMessage((Message) => 
+            PackageConsumer.OnMessage((Message) =>
             {
-
+                var merge = JT1078.Protocol.JT1078Serializer.Merge(Message.Data);
+                if (merge != null)
+                {
+                    var httpSessions = HttpSessionManager.GetAllBySimAndChannelNo(Message.Data.SIM, Message.Data.LogicChannelNumber);
+                    var firstHttpSessions = httpSessions.Where(w => !w.FirstSend).ToList();
+                    if (firstHttpSessions.Count > 0)
+                    {
+                        var flvVideoBuffer = FlvEncoder.EncoderVideoTag(merge, true);
+                        HttpSessionManager.SendAVData(firstHttpSessions, flvVideoBuffer, true);
+                    }
+                    var otherHttpSessions = httpSessions.Where(w => w.FirstSend).ToList();
+                    if (otherHttpSessions.Count > 0)
+                    {
+                        var flvVideoBuffer = FlvEncoder.EncoderVideoTag(merge, false);
+                        HttpSessionManager.SendAVData(firstHttpSessions, flvVideoBuffer, false);
+                    }
+                }
             });
             return Task.CompletedTask;
         }

src/JT1078.Gateway/Metadata/JT1078HttpContext.cs

@@ -23,14 +23,14 @@ namespace JT1078.Gateway.Metadata
             }
         }
         public DateTime StartTime { get; set; }
-        public bool SendChunked { get; set; }
+        public bool FirstSend { get; set; }
         public JT1078HttpContext(HttpListenerContext context, IPrincipal user)
         {
             Context = context;
             User = user;
             StartTime = DateTime.Now;
             SessionId = Guid.NewGuid().ToString("N");
-            SendChunked = false;
+            FirstSend = false;
         }
         public JT1078HttpContext(HttpListenerContext context, HttpListenerWebSocketContext webSocketContext, IPrincipal user)
         {
@@ -39,7 +39,7 @@ namespace JT1078.Gateway.Metadata
             User = user;
             StartTime = DateTime.Now;
             SessionId = Guid.NewGuid().ToString("N");
-            SendChunked = false;
+            FirstSend = false;
         }
     }
 }

src/JT1078.Gateway/Sessions/JT1078HttpSessionManager.cs

@@ -88,16 +88,20 @@ namespace JT1078.Gateway.Sessions
         /// <summary>
         /// 发送音视频数据
         /// </summary>
-        /// <param name="sim"></param>
-        /// <param name="channelNo"></param>
+        /// <param name="httpContexts"></param>
         /// <param name="data"></param>
-        public void SendAVData(string sim,int channelNo,byte[] data)
+        /// <param name="firstSend"></param>
+        public void SendAVData(List<JT1078HttpContext> httpContexts, byte[] data, bool firstSend)
         {
-            var contexts = Sessions.Select(s => s.Value).Where(w => w.Sim == sim && w.ChannelNo == channelNo).ToList();
-            ParallelLoopResult parallelLoopResult= Parallel.ForEach(contexts, async(context) => 
+            ParallelLoopResult parallelLoopResult = Parallel.ForEach(httpContexts, async (context) =>
             {
                 if (context.IsWebSocket)
                 {
+                    if (firstSend)
+                    {
+                        context.FirstSend = firstSend;
+                        Sessions.TryUpdate(context.SessionId, context, context);
+                    }
                     try
                     {
                         await context.WebSocketSendBinaryAsync(data);
@@ -109,13 +113,13 @@ namespace JT1078.Gateway.Sessions
                             Logger.LogInformation($"[ws close]:{context.SessionId}-{context.Sim}-{context.ChannelNo}-{context.StartTime:yyyyMMddhhmmss}");
                         }
                         remove(context.SessionId);
-                    }
+                    }                   
                 }
                 else
                 {
-                    if (!context.SendChunked)
+                    if (firstSend)
                     {
-                        context.SendChunked = true;
+                        context.FirstSend = firstSend;
                         Sessions.TryUpdate(context.SessionId, context, context);
                         try
                         {
@@ -153,89 +157,6 @@ namespace JT1078.Gateway.Sessions
             }
         }
 
-        /// <summary>
-        /// 发送音视频数据到websocket
-        /// </summary>
-        /// <param name="sim"></param>
-        /// <param name="channelNo"></param>
-        /// <param name="data"></param>
-        public void SendAVData2WebSocket(string sim, int channelNo, byte[] data)
-        {
-            var contexts = Sessions.Select(s => s.Value).Where(w => w.Sim == sim && w.ChannelNo == channelNo && w.IsWebSocket).ToList();
-            ParallelLoopResult parallelLoopResult = Parallel.ForEach(contexts, async (context) =>
-            {
-                if (context.IsWebSocket)
-                {
-                    try
-                    {
-                        await context.WebSocketSendBinaryAsync(data);
-                    }
-                    catch (Exception ex)
-                    {
-                        if (Logger.IsEnabled(LogLevel.Information))
-                        {
-                            Logger.LogInformation($"[ws close]:{context.SessionId}-{context.Sim}-{context.ChannelNo}-{context.StartTime:yyyyMMddhhmmss}");
-                        }
-                        remove(context.SessionId);
-                    }
-                }
-            });
-            if (parallelLoopResult.IsCompleted)
-            {
-
-            }
-        }
-
-        /// <summary>
-        /// 发送音视频数据到Http Chunked中
-        /// </summary>
-        /// <param name="sim"></param>
-        /// <param name="channelNo"></param>
-        /// <param name="data"></param>
-        public void SendAVData2HttpChunked(string sim, int channelNo, byte[] data)
-        {
-            var contexts = Sessions.Select(s => s.Value).Where(w => w.Sim == sim && w.ChannelNo == channelNo && !w.IsWebSocket).ToList();
-            ParallelLoopResult parallelLoopResult = Parallel.ForEach(contexts, async (context) =>
-            {
-                if (!context.SendChunked)
-                {
-                    context.SendChunked = true;
-                    Sessions.TryUpdate(context.SessionId, context, context);
-                    try
-                    {
-                        await context.HttpSendFirstChunked(data);
-                    }
-                    catch (Exception ex)
-                    {
-                        if (Logger.IsEnabled(LogLevel.Information))
-                        {
-                            Logger.LogInformation($"[http close]:{context.SessionId}-{context.Sim}-{context.ChannelNo}-{context.StartTime:yyyyMMddhhmmss}");
-                        }
-                        remove(context.SessionId);
-                    }
-                }
-                else
-                {
-                    try
-                    {
-                        await context.HttpSendChunked(data);
-                    }
-                    catch (Exception ex)
-                    {
-                        if (Logger.IsEnabled(LogLevel.Information))
-                        {
-                            Logger.LogInformation($"[http close]:{context.SessionId}-{context.Sim}-{context.ChannelNo}-{context.StartTime:yyyyMMddhhmmss}");
-                        }
-                        remove(context.SessionId);
-                    }
-                }             
-            });
-            if (parallelLoopResult.IsCompleted)
-            {
-
-            }
-        }
-
         public int SessionCount 
         {
             get
@@ -260,6 +181,11 @@ namespace JT1078.Gateway.Sessions
             }
         }
 
+        public List<JT1078HttpContext> GetAllBySimAndChannelNo(string sim, int channelNo)
+        {
+            return Sessions.Select(s => s.Value).Where(w => w.Sim == sim && w.ChannelNo == channelNo).ToList();
+        }
+
         public List<JT1078HttpContext> GetAll()
         {
             return Sessions.Select(s => s.Value).ToList();