1.添加fmp4测试（待测试）

2.整理前端demo
3.修改文档

README.md

@@ -7,6 +7,7 @@
 3. 了解WebSocket消息推送
 4. [了解flv.js](https://github.com/bilibili/flv.js)
 5. [了解hls.js](https://github.com/video-dev/hls.js)
+6. 了解fmp4
 
 > 注意：暂不支持音频
 
@@ -14,6 +15,7 @@
 | --- | ---| --- |---|---|
 | flv  | 😀| ☹ |😀|http-flv、ws-flv|
 | m3u8 | 😀| ☹ |😀|http|
+| fmp4 | 😀| ☹ |☹|http-fmp4、ws-fmp4|
 
 ## NuGet安装
 

src/JT1078.Gateway.Abstractions/IJT1078MsgProducer.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Generic;
 using System.Text;
+using System.Threading;
 using System.Threading.Tasks;
 
 namespace JT1078.Gateway.Abstractions
@@ -12,6 +13,7 @@ namespace JT1078.Gateway.Abstractions
         /// </summary>
         /// <param name="sim">设备sim终端号</param>
         /// <param name="data">jt1078 hex data</param>
-        ValueTask ProduceAsync(string sim, byte[] data);
+        /// <param name="cancellationToken">cts</param>
+        ValueTask ProduceAsync(string sim, byte[] data, CancellationToken cancellationToken = default);
     }
 }

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

@@ -33,7 +33,7 @@
   </ItemGroup>
 
   <ItemGroup>
-    <PackageReference Include="JT1078" Version="1.1.0" />
+    <PackageReference Include="JT1078" Version="1.1.1-preview1" />
     <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="5.0.0" />
     <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="5.0.0" />
     <PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="5.0.0" />

src/JT1078.Gateway.Abstractions/JT1078.Gateway.Abstractions.xml

@@ -9,12 +9,13 @@
             传输协议类型
             </summary>
         </member>
-        <member name="M:JT1078.Gateway.Abstractions.IJT1078MsgProducer.ProduceAsync(System.String,System.Byte[])">
+        <member name="M:JT1078.Gateway.Abstractions.IJT1078MsgProducer.ProduceAsync(System.String,System.Byte[],System.Threading.CancellationToken)">
             <summary>
             
             </summary>
             <param name="sim">设备sim终端号</param>
             <param name="data">jt1078 hex data</param>
+            <param name="cancellationToken">cts</param>
         </member>
         <member name="P:JT1078.Gateway.Abstractions.IJT1078Session.TerminalPhoneNo">
             <summary>

src/JT1078.Gateway.InMemoryMQ/JT1078MsgProducer.cs

@@ -3,6 +3,7 @@ using JT1078.Protocol;
 using System;
 using System.Collections.Generic;
 using System.Text;
+using System.Threading;
 using System.Threading.Channels;
 using System.Threading.Tasks;
 
@@ -24,9 +25,9 @@ namespace JT1078.Gateway.InMemoryMQ
             
         }
 
-        public async ValueTask ProduceAsync(string sim, byte[] data)
+        public async ValueTask ProduceAsync(string sim, byte[] data, CancellationToken cancellationToken = default)
         {
-            await Channel.Channel.Writer.WriteAsync((sim, data));
+            await Channel.Channel.Writer.WriteAsync((sim, data), cancellationToken);
         }
     }
 }

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

@@ -31,11 +31,13 @@
     <None Update="Configs\NLog.xsd">
       <CopyToOutputDirectory>Always</CopyToOutputDirectory>
     </None>
-    <None Update="wwwroot\demo\demo.m3u8">
+    <None Update="wwwroot\hls_demo\demo.m3u8">
       <CopyToOutputDirectory>Always</CopyToOutputDirectory>
     </None>
-    <None Update="wwwroot\demo\demo.ts">
+    <None Update="wwwroot\hls_demo\demo.ts">
       <CopyToOutputDirectory>Always</CopyToOutputDirectory>
     </None>
   </ItemGroup>
+
+  <ProjectExtensions><VisualStudio><UserProperties appsettings_1json__JsonSchema="" /></VisualStudio></ProjectExtensions>
 </Project>

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

@@ -1,8 +1,10 @@
 using JT1078.Flv;
+using JT1078.FMp4;
 using JT1078.Gateway.InMemoryMQ;
 using JT1078.Gateway.TestNormalHosting.Services;
 using JT1078.Hls;
 using JT1078.Hls.Options;
+using JT1078.Protocol.H264;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
@@ -35,20 +37,19 @@ namespace JT1078.Gateway.TestNormalHosting
                 .ConfigureServices((hostContext, services) =>
                 {
                     services.AddMemoryCache();                  
-                    services.AddSingleton<ILoggerFactory, LoggerFactory>();
-                    services.AddSingleton(typeof(ILogger<>), typeof(Logger<>));
                     //flv视频解码器
                     services.AddSingleton<FlvEncoder>();
                     //hls视频解码器
                     services.AddSingleton<TSEncoder>();
                     services.AddSingleton<M3U8FileManage>();
+                    services.AddSingleton<H264Decoder>();
+                    services.AddSingleton<FMp4Encoder>();
                     //添加hls依赖项
                     services.AddHlsGateway(hostContext.Configuration);
                     services.Configure<M3U8Option>(hostContext.Configuration.GetSection("M3U8Option"));
                     var m3u8Option = services.BuildServiceProvider().GetRequiredService<IOptions<M3U8Option>>().Value;
                     services.AddSingleton(m3u8Option);
 
-                    
                     //使用内存队列实现会话通知
                     services.AddJT1078Gateway(hostContext.Configuration)
                             .AddTcp()
@@ -60,6 +61,7 @@ namespace JT1078.Gateway.TestNormalHosting
                     //内存队列没有做分发，可以自己实现。
                     services.AddHostedService<JT1078FlvNormalMsgHostedService>();
                     services.AddHostedService<JT1078HlsNormalMsgHostedService>();
+                    services.AddHostedService<JT1078FMp4NormalMsgHostedService>();
 
                     services.AddSingleton<MessageDispatchDataService>();
                     services.AddHostedService<MessageDispatchHostedService>();

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

@@ -15,33 +15,37 @@ using JT1078.Protocol;
 using System.Text.Json;
 using System.Text.Json.Serialization;
 using JT1078.FMp4;
+using JT1078.Protocol.H264;
+using System.Collections.Concurrent;
 
 namespace JT1078.Gateway.TestNormalHosting.Services
 {
     public class JT1078FMp4NormalMsgHostedService : BackgroundService
     {
-        private IJT1078MsgConsumer JT1078MsgConsumer;
         private JT1078HttpSessionManager HttpSessionManager;
         private FMp4Encoder FM4Encoder;
+        private readonly H264Decoder H264Decoder;
         private ILogger Logger;
         private IMemoryCache memoryCache;
-        private const string ikey = "IKEY";
+        private const string ikey = "IFMp4KEY";
         private MessageDispatchDataService messageDispatchDataService;
+        private ConcurrentDictionary<string, List<H264NALU>> avFrameDict;
 
         public JT1078FMp4NormalMsgHostedService(
             MessageDispatchDataService messageDispatchDataService,
             IMemoryCache memoryCache,
             ILoggerFactory loggerFactory,
+            H264Decoder h264Decoder,
             FMp4Encoder fM4Encoder,
-            JT1078HttpSessionManager httpSessionManager,
-            IJT1078MsgConsumer msgConsumer)
+            JT1078HttpSessionManager httpSessionManager)
         {
             Logger = loggerFactory.CreateLogger<JT1078FMp4NormalMsgHostedService>();
-            JT1078MsgConsumer = msgConsumer;
             HttpSessionManager = httpSessionManager;
             FM4Encoder = fM4Encoder;
+            H264Decoder = h264Decoder;
             this.memoryCache = memoryCache;
             this.messageDispatchDataService = messageDispatchDataService;
+            avFrameDict = new ConcurrentDictionary<string, List<H264NALU>>();
         }
         protected async override Task ExecuteAsync(CancellationToken stoppingToken)
         {
@@ -49,7 +53,8 @@ namespace JT1078.Gateway.TestNormalHosting.Services
             {
                 var data = await messageDispatchDataService.FlvChannel.Reader.ReadAsync();
                 try
-                {      
+                {
+                    var nalus = H264Decoder.ParseNALU(data);
                     if (Logger.IsEnabled(LogLevel.Debug))
                     {
                         Logger.LogDebug(JsonSerializer.Serialize(HttpSessionManager.GetAll()));
@@ -58,30 +63,66 @@ namespace JT1078.Gateway.TestNormalHosting.Services
                     string key = $"{data.GetKey()}_{ikey}";
                     if (data.Label3.DataType == Protocol.Enums.JT1078DataType.视频I帧)
                     {
-                        memoryCache.Set(key, data);
+                        var moov = FM4Encoder.EncoderMoovBox(nalus.FirstOrDefault(f => f.NALUHeader.NalUnitType == NalUnitType.SPS),
+                            nalus.FirstOrDefault(f => f.NALUHeader.NalUnitType == NalUnitType.PPS));
+                        memoryCache.Set(key, moov);
                     }
                     var httpSessions = HttpSessionManager.GetAllBySimAndChannelNo(data.SIM.TrimStart('0'), data.LogicChannelNumber);
                     var firstHttpSessions = httpSessions.Where(w => !w.FirstSend).ToList();
                     if (firstHttpSessions.Count > 0)
                     {
-                        if (memoryCache.TryGetValue(key, out JT1078Package idata))
+                        try
                         {
-                            try
-                            {
-                                
-                            }
-                            catch (Exception ex)
+                            var flvVideoBuffer = FM4Encoder.EncoderFtypBox();
+                            memoryCache.TryGetValue(key, out byte[] moovBuffer);
+                            foreach (var session in firstHttpSessions)
                             {
-                                Logger.LogError(ex, $"{data.SIM},{true},{data.SN},{data.LogicChannelNumber},{data.Label3.DataType.ToString()},{data.Label3.SubpackageType.ToString()},{data.Bodies.ToHexString()}");
+                                HttpSessionManager.SendAVData(session, flvVideoBuffer, true);
+                                if (moovBuffer != null)
+                                {
+                                    HttpSessionManager.SendAVData(session, moovBuffer, false);
+                                }
                             }
                         }
+                        catch (Exception ex)
+                        {
+                            Logger.LogError(ex, $"{data.SIM},{true},{data.SN},{data.LogicChannelNumber},{data.Label3.DataType.ToString()},{data.Label3.SubpackageType.ToString()},{data.Bodies.ToHexString()}");
+                        }
                     }
                     var otherHttpSessions = httpSessions.Where(w => w.FirstSend).ToList();
                     if (otherHttpSessions.Count > 0)
                     {
                         try
                         {
-                           
+                            if(!avFrameDict.TryGetValue(key, out List<H264NALU> frames))
+                            {
+                                frames = new List<H264NALU>();
+                                avFrameDict.TryAdd(key, frames);
+                            }
+                            foreach (var nalu in nalus)
+                            {
+                                if (nalu.Slice)
+                                {
+                                    //H264 NALU slice first_mb_in_slice
+                                    frames.Add(nalu);
+                                }
+                                else
+                                {
+                                    if (nalus.Count > 0)
+                                    {
+                                        var otherBuffer = FM4Encoder.EncoderOtherVideoBox(frames);
+                                        foreach (var session in otherHttpSessions)
+                                        {
+                                            if (otherBuffer != null)
+                                            {
+                                                HttpSessionManager.SendAVData(session, otherBuffer, false);
+                                            }
+                                        }
+                                        frames.Clear();
+                                    }
+                                    frames.Add(nalu);
+                                }
+                            }
                         }
                         catch (Exception ex)
                         {

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

@@ -12,5 +12,6 @@ namespace JT1078.Gateway.TestNormalHosting.Services
     {
         public Channel<JT1078Package> HlsChannel = Channel.CreateUnbounded<JT1078Package>();
         public Channel<JT1078Package> FlvChannel = Channel.CreateUnbounded<JT1078Package>();
+        public Channel<JT1078Package> FMp4Channel = Channel.CreateUnbounded<JT1078Package>();
     }
 }

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

@@ -20,7 +20,7 @@ namespace JT1078.Gateway.TestNormalHosting.Services
         private readonly MessageDispatchDataService messageDispatchDataService;
 
         public MessageDispatchHostedService(IJT1078MsgConsumer JT1078MsgConsumer,
-                                                                        MessageDispatchDataService messageDispatchDataService) {
+                                            MessageDispatchDataService messageDispatchDataService) {
             this.JT1078MsgConsumer = JT1078MsgConsumer;
             this.messageDispatchDataService = messageDispatchDataService;
         }
@@ -33,8 +33,9 @@ namespace JT1078.Gateway.TestNormalHosting.Services
                 var merge = JT1078.Protocol.JT1078Serializer.Merge(package);
                 if (merge != null)
                 {
-                    await messageDispatchDataService.HlsChannel.Writer.WriteAsync(merge);
-                    await messageDispatchDataService.FlvChannel.Writer.WriteAsync(merge);
+                    await messageDispatchDataService.HlsChannel.Writer.WriteAsync(merge, stoppingToken);
+                    await messageDispatchDataService.FlvChannel.Writer.WriteAsync(merge, stoppingToken);
+                    await messageDispatchDataService.FMp4Channel.Writer.WriteAsync(merge, stoppingToken);
                 }
             });
             return Task.CompletedTask;

src/JT1078.Gateway.Tests/JT1078.Gateway.TestNormalHosting/wwwroot/fmp4_demo/index.html

@@ -0,0 +1,201 @@
+<!DOCTYPE html>
+
+<html lang="en" xmlns="http://www.w3.org/1999/xhtml">
+<head>
+    <meta charset="utf-8" />
+    <title>WebSocket MSE Fmp4 demo</title>
+</head>
+<body>
+    <h1>MSE FMp4 Demo</h1>
+    <video id="stream_live" width="640" height="480" controls="false" autoplay="true"
+           muted="muted"
+           preload="auto">
+        浏览器不支持
+    </video>
+    <ul id="messagesList"></ul>
+    <script>
+        //var mimeCodec = 'video/mp4;codecs="avc1.4D0014, mp4a.40.2"';
+        // *** USER PARAMETERS ***
+        var verbose = true;
+        // var verbose = true; // enable for saturating the console ..
+        var buffering_sec = 1; // use some reasonable value
+        var buffering_sec_seek = buffering_sec * 0.9;
+        // ..seek the stream if it's this much away or
+        // from the last available timestamp
+        var buffering_sec_seek_distance = buffering_sec * 0.5;
+        // .. jump to this distance from the last avail. timestamp
+        // *** INTERNAL PARAMETERS ***
+        // set mimetype and codec
+        var mimeType = "video/mp4";
+        var codecs = "avc1.4D0014"; // https://wiki.whatwg.org/wiki/Video_type_parameters
+        // if your stream has audio, remember to include it in these definitions.. otherwise your mse goes sour
+        var codecPars = mimeType + ';codecs="' + codecs + '"';
+        var stream_started = false; // is the source_buffer updateend callback active nor not
+        // create media source instance
+        var ms = new MediaSource();
+        // queue for incoming media packets
+        var queue = [];
+        var stream_live; // the HTMLMediaElement (i.e. <video> element)
+        var ws; // websocket
+        var seeked = false; // have have seeked manually once ..
+        var cc = 0;
+        var source_buffer; // source_buffer instance
+        var pass = 0;
+        // *** MP4 Box manipulation functions ***
+        // taken from here: https://stackoverflow.com/questions/54186634/sending-periodic-metadata-in-fragmented-live-mp4-stream/
+        function toInt(arr, index) { // From bytes to big-endian 32-bit integer.  Input: Uint8Array, index
+            var dv = new DataView(arr.buffer, 0);
+            return dv.getInt32(index, false); // big endian
+        }
+        function toString(arr, fr, to) { // From bytes to string.  Input: Uint8Array, start index, stop index.
+            // https://developers.google.com/web/updates/2012/06/How-to-convert-ArrayBuffer-to-and-from-String
+            return String.fromCharCode.apply(null, arr.slice(fr, to));
+        }
+        function getBox(arr, i) { // input Uint8Array, start index
+            return [toInt(arr, i), toString(arr, i + 4, i + 8)]
+        }
+        function getSubBox(arr, box_name) { // input Uint8Array, box name
+            var i = 0;
+            res = getBox(arr, i);
+            main_length = res[0]; name = res[1]; // this boxes length and name
+            i = i + 8;
+            var sub_box = null;
+            while (i < main_length) {
+                res = getBox(arr, i);
+                l = res[0]; name = res[1];
+
+                if (box_name == name) {
+                    sub_box = arr.slice(i, i + l)
+                }
+                i = i + l;
+            }
+            return sub_box;
+        }
+        function hasFirstSampleFlag(arr) { // input Uint8Array
+            // [moof [mfhd] [traf [tfhd] [tfdt] [trun]]]
+            var traf = getSubBox(arr, "traf");
+            if (traf == null) { return false; }
+            var trun = getSubBox(traf, "trun");
+            if (trun == null) { return false; }
+            // ISO/IEC 14496-12:2012(E) .. pages 5 and 57
+            // bytes: (size 4), (name 4), (version 1 + tr_flags 3)
+            var flags = trun.slice(10, 13); // console.log(flags);
+            f = flags[1] & 4; // console.log(f);
+            return f == 4;
+        }
+        // consider these callbacks:
+        // - putPacket : called when websocket receives data
+        // - loadPacket : called when source_buffer is ready for more data
+        // Both operate on a common fifo
+        function putPacket(arr) {
+            // receives ArrayBuffer.  Called when websocket gets more data
+            // first packet ever to arrive: write directly to source_buffer
+            // source_buffer ready to accept: write directly to source_buffer
+            // otherwise insert it to queue
+            var memview = new Uint8Array(arr);
+            if (verbose) { console.log("got", arr.byteLength, "bytes.  Values=", memview[0], memview[1], memview[2], memview[3], memview[4]); }
+            res = getBox(memview, 0);
+            main_length = res[0]; name = res[1]; // this boxes length and name
+            // if ((name == "ftyp") && (pass == 0)) {
+            //    pass = pass + 1;
+            //    console.log("got ftyp");
+            // }
+            // else if ((name == "moov") && (pass == 1)) {
+            //    pass = pass + 1;
+            //    console.log("got moov");
+            // }
+            // else if ((name == "moof") && (pass == 2)) {
+            //    if (hasFirstSampleFlag(memview)) {
+            //        pass = pass + 1;
+            //        console.log("got that special moof");
+            //    }
+            //    else {
+            //        return;
+            //    }
+            // }
+            // else if (pass < 3) {
+            //    return;
+            // }
+            // keep the latency to minimum
+            let latest = stream_live.duration;
+            if ((stream_live.duration >= buffering_sec) &&
+                ((latest - stream_live.currentTime) > buffering_sec_seek)) {
+                console.log("seek from ", stream_live.currentTime, " to ", latest);
+                df = (stream_live.duration - stream_live.currentTime); // this much away from the last available frame
+                if ((df > buffering_sec_seek)) {
+                    seek_to = stream_live.duration - buffering_sec_seek_distance;
+                    stream_live.currentTime = seek_to;
+                }
+            }
+            data = arr;
+            if (!stream_started) {
+                if (verbose) { console.log("Streaming started: ", memview[0], memview[1], memview[2], memview[3], memview[4]); }
+                stream_started = true;
+                source_buffer.appendBuffer(data);
+                
+                cc = cc + 1;
+                return;
+            }
+            queue.push(data); // add to the end
+            if (verbose) { console.log("queue push:", queue.length); }
+        }
+
+        function loadPacket() { // called when source_buffer is ready for more
+            if (!source_buffer.updating) { // really, really ready
+                if (queue.length > 0) {
+                    inp = queue.shift(); // pop from the beginning
+                    if (verbose) { console.log("queue pop:", queue.length); }
+                    var memview = new Uint8Array(inp);
+                    if (verbose) { console.log(" ==> writing buffer with", memview[0], memview[1], memview[2], memview[3]); }
+                    source_buffer.appendBuffer(inp);
+                    cc = cc + 1;
+                }
+                else { // the queue runs empty, so the next packet is fed directly
+                    stream_started = false;
+                }
+            }
+            else { // so it was not?
+            }
+        }
+
+        function opened() { // MediaSource object is ready to go
+            // https://developer.mozilla.org/en-US/docs/Web/API/MediaSource/duration
+            ms.duration = buffering_sec;
+            source_buffer = ms.addSourceBuffer(codecPars);
+            // https://developer.mozilla.org/en-US/docs/Web/API/source_buffer/mode
+            var myMode = source_buffer.mode;
+            source_buffer.mode = 'sequence';
+            // source_buffer.mode = 'segments';
+            source_buffer.addEventListener("updateend", loadPacket);
+
+            ws = new WebSocket("ws://127.0.0.1:15555/live.mp4?sim=12345678901&channel=3&token=123456"); //创建WebSocket连接
+            ws.onmessage = function (e) {
+                //当客户端收到服务端发来的消息时，触发onmessage事件，参数e.data包含server传递过来的数据
+                console.log(e.data);
+                putPacket(e.data);
+            }
+        }
+
+        function startup() {
+            ms.addEventListener('sourceopen', opened, false);
+            // get reference to video
+            stream_live = document.getElementById('stream_live');
+            // set mediasource as source of video
+            stream_live.src = window.URL.createObjectURL(ms);
+        }
+        function base64ToArrayBuffer(base64) {
+            var binary_string = window.atob(base64);
+            var len = binary_string.length;
+            var bytes = new Uint8Array(len);
+            for (var i = 0; i < len; i++) {
+                bytes[i] = binary_string.charCodeAt(i);
+            }
+            return bytes.buffer;
+        }
+        window.onload = function () {
+            startup();
+        }
+
+    </script>
+</body>
+</html>