feat: 使用Jmuxer在浏览器中播放h264 nalu stream

Author: nintha

Cargo.lock

@@ -21,4 +21,6 @@ num-traits = "0.2"
 anyhow = "1.0.33"
 dashmap = "4.0.0-rc6"
 crossbeam-utils = "0.8"
-once_cell = "1.4.1"
+once_cell = "1.4.1"
+async-tungstenite = "0.10.0"
+futures = "0.3"

Cargo.toml

@@ -7,20 +7,26 @@ OBS
 
 ## Play
 
+### ffplay
 ```shell
 ffplay -fflags nobuffer -analyzeduration 100000 rtmp://localhost:11935/channel/token
 ```
-
 使用`-fflags nobuffer -analyzeduration 100000`可以有效降低播放的延迟，目前本地测试延迟大概为1秒
 
+### JMuxer（video only）
+使用[Jmuxer](https://github.com/samirkumardas/jmuxer)在浏览器中播放，详见`example/h264-nalu-stream`目录。
+
+如果是使用x264编码推流，建议profile=baseline，可以避免视频频繁抖动，目前本地测试延迟大概为1秒
+
 ## Completed
 - [x] 支持不同分辨率的推流和拉流（之前默认1028x720）
 - [x] 支持音频传输
 - [x] 支持HTTP-FLV输出
+- [x] 输出H264流，使用[Jmuxer](https://github.com/samirkumardas/jmuxer)在浏览器中播放
 
 ## TODO
 - [ ] Web GUI 播放界面(FLV)
-- [ ] 输出H264流，使用[Jmuxer](https://github.com/samirkumardas/jmuxer)在浏览器中播放
+
 
 ## 参考资料
 - [RTMP推送AAC ADTS音频流](https://www.jianshu.com/p/1a6f195863c7)

README.md

@@ -2,8 +2,8 @@ use crate::util::spawn_and_log_error;
 use smol::io::{AsyncReadExt, AsyncWriteExt};
 use smol::net::{TcpListener, TcpStream};
 use smol::stream::StreamExt;
-use crate::rtmp_server::{eventbus_map, video_header_map, audio_header_map, meta_data_map};
-use crate::protocol::flv::{FLV_HEADER_WITH_TAG0, FLV_HEADER_ONLY_VIDEO_WITH_TAG0};
+use crate::rtmp_server::{eventbus_map, video_header_map};
+use crate::protocol::flv::{FLV_HEADER_ONLY_VIDEO_WITH_TAG0};
 use crate::protocol::flv::FlvTag;
 use chrono::Local;
 use std::convert::TryFrom;

src/http_flv.rs

@@ -1,7 +1,6 @@
 #[macro_use]
 extern crate num_derive;
 
-use crate::http_flv::run_server;
 use crate::rtmp_server::accept_loop;
 use crate::util::spawn_and_log_error;
 
@@ -10,11 +9,13 @@ mod http_flv;
 mod protocol;
 mod rtmp_server;
 mod util;
+mod ws_h264;
 
 fn main() -> anyhow::Result<()> {
     util::init_logger();
 
-    spawn_and_log_error(run_server("0.0.0.0:8080"));
+    spawn_and_log_error(http_flv::run_server("0.0.0.0:8080"));
+    spawn_and_log_error(ws_h264::run_server("0.0.0.0:8081"));
     let server = "0.0.0.0:11935";
     smol::block_on(accept_loop(server))
 }

src/main.rs

@@ -1,3 +1,3 @@
 pub mod flv;
-pub mod h264;
 pub mod rtmp;
+pub mod nalu;

src/protocol/mod.rs

@@ -0,0 +1,246 @@
+use byteorder::{BigEndian, ByteOrder};
+
+use crate::protocol::rtmp::{RtmpContext, RtmpMessage};
+
+/// H264编码数据存储或传输的基本单元
+pub struct Nalu {
+    inner: Vec<u8>,
+    pub is_key_frame: bool,
+}
+
+impl Nalu {
+    /// RtmpMessage to Nalus
+    pub fn from_rtmp_message(msg: &RtmpMessage) -> Vec<Nalu> {
+        let bytes = &msg.body;
+        let mut nalus = vec![];
+
+        let frame_type = bytes[0];
+        let is_key_frame = frame_type == 0x17;
+        let mut read_index = 1;
+        let acv_packet_type = bytes[read_index];
+        read_index += 4;
+
+        // AVCDecoderConfigurationRecord（AVC sequence header）
+        if acv_packet_type == 0 {
+            read_index += 5;
+            let sps_num = &bytes[read_index] & 0x1F;
+            read_index += 1;
+            for _ in 0..sps_num as usize {
+                let data_len = BigEndian::read_u16(&bytes[read_index..]);
+                read_index += 2;
+                let data = &bytes[read_index..(read_index + data_len as usize)];
+                read_index += data_len as usize;
+
+                let mut nalu_bytes: Vec<u8> = vec![0x00, 0x00, 0x00, 0x01];
+                nalu_bytes.extend_from_slice(data);
+                nalus.push(Self { inner: nalu_bytes, is_key_frame });
+            }
+            let num_of_pps = &bytes[read_index] & 0x1F;
+            read_index += 1;
+            for _ in 0..num_of_pps as usize {
+                let data_len = BigEndian::read_u16(&bytes[read_index..]);
+                read_index += 2;
+                let data = &bytes[read_index..(read_index + data_len as usize)];
+                read_index += data_len as usize;
+
+                let mut nalu_bytes: Vec<u8> = vec![0x00, 0x00, 0x00, 0x01];
+                nalu_bytes.extend_from_slice(data);
+                nalus.push(Self { inner: nalu_bytes, is_key_frame });
+            }
+        }
+        // One or more NALUs (Full frames are required)
+        else if acv_packet_type == 1 {
+            loop {
+                if read_index >= bytes.len() {
+                    break;
+                }
+                let data_len = BigEndian::read_u32(&bytes[read_index..]);
+                read_index += 4;
+                let data = &bytes[read_index..(read_index + data_len as usize)];
+                read_index += data_len as usize;
+
+                let mut nalu_bytes: Vec<u8> = vec![0x00, 0x00, 0x00, 0x01];
+                nalu_bytes.extend_from_slice(data);
+                nalus.push(Self { inner: nalu_bytes, is_key_frame });
+            }
+        } else {
+            log::warn!("unknown acv packet type");
+        };
+        nalus
+    }
+
+    /// 帧优先级
+    #[allow(unused)]
+    pub fn get_nal_ref_idc(&self) -> u8 {
+        self.inner[0] >> 5
+    }
+
+    /// 帧类型
+    #[allow(unused)]
+    pub fn get_nal_unit_type(&self) -> u8 {
+        self.inner[0] & 0x1F
+    }
+
+    #[allow(unused)]
+    pub fn nalu_type_desc(&self) -> String {
+        let priority: String = match self.get_nal_ref_idc() {
+            0 => "DISPOSABLE".into(),
+            1 => "LOW".into(),
+            2 => "HIGH".into(),
+            3 => "HIGHEST".into(),
+            _ => "UNKNOWN".into(),
+        };
+
+        let t: String = match self.get_nal_unit_type() {
+            1 => "SLICE".into(),
+            2 => "DPA".into(),
+            3 => "DPB".into(),
+            4 => "DPC".into(),
+            5 => "IDR".into(),
+            6 => "SEI".into(),
+            7 => "SPS".into(),
+            8 => "PPS".into(),
+            9 => "AUD".into(),
+            10 => "EOSEQ".into(),
+            11 => "EOSTREAM".into(),
+            12 => "FILL".into(),
+            _ => "UNKNOWN".into(),
+        };
+
+        format!("{}::{}", priority, t)
+    }
+}
+
+impl AsRef<[u8]> for Nalu {
+    fn as_ref(&self) -> &[u8] {
+        self.inner.as_ref()
+    }
+}
+
+/// # VideoTagHeader
+///
+/// ## Frame Type
+///
+/// Type: UB [4]
+///
+/// Type of video frame. The following values are defined:
+/// 1 = key frame (for AVC, a seekable frame)
+/// 2 = inter frame (for AVC, a non-seekable frame)
+/// 3 = disposable inter frame (H.263 only)
+/// 4 = generated key frame (reserved for server use only)
+/// 5 = video info/command frame
+///
+/// ## CodecID
+///
+/// Type: UB [4]
+///
+/// Codec Identifier. The following values are defined:
+/// 2 = Sorenson H.263
+/// 3 = Screen video
+/// 4 = On2 VP6
+/// 5 = On2 VP6 with alpha channel
+/// 6 = Screen video version 2
+/// 7 = AVC
+///
+/// ## AVCPacketType
+///
+/// Type: IF CodecID == 7, UI8
+///
+/// The following values are defined:
+/// 0 = AVC sequence header
+/// 1 = AVC NALU
+/// 2 = AVC end of sequence (lower level NALU sequence ender is not required or supported)
+///
+/// ## CompositionTime
+///
+/// Type: IF CodecID == 7, SI24
+///
+/// IF AVCPacketType == 1
+///     Composition time offset
+/// ELSE
+///     0
+/// See ISO 14496-12, 8.15.3 for an explanation of composition
+/// times. The offset in an FLV file is always in milliseconds.
+#[allow(unused)]
+pub fn handle_video_data(bytes: &[u8], ctx: &RtmpContext) {
+    let frame_type = bytes[0];
+    let mut read_index = 1;
+    let acv_packet_type = bytes[read_index];
+    read_index += 1;
+
+    // AVC时，全0，无意义（作业时间）
+    let _composition_time_offset = &bytes[read_index..read_index + 3];
+    read_index += 3;
+
+    log::debug!(
+        "[peer={}] video frame type = {:#04X}, acv_packet_type={:#04X}",
+        &ctx.peer_addr,
+        frame_type,
+        acv_packet_type
+    );
+
+    // AVCDecoderConfigurationRecord（AVC sequence header）
+    if acv_packet_type == 0 {
+        // let config_version = &bytes[read_index];
+        // let avc_profile_indication = &bytes[read_index + 1];
+        // let profile_compatibility = &bytes[read_index + 2];
+        // let avc_level_indication = &bytes[read_index + 3];
+        // let length_size_minus_one = &bytes[read_index + 4];
+        read_index += 5;
+        let sps_num = &bytes[read_index] & 0x1F;
+        read_index += 1;
+        // println!("sps_num={}", sps_num);
+        // println!("config_version={:#04X}", config_version);
+        // println!("avc_profile_indication={:#04X}", avc_profile_indication);
+        // println!("profile_compatibility={:#04X}", profile_compatibility);
+        // println!("avc_level_indication={:#04X}", avc_level_indication);
+        // println!("length_size_minus_one={:#04X}", length_size_minus_one);
+        for _ in 0..sps_num as usize {
+            let data_len = BigEndian::read_u16(&bytes[read_index..]);
+            read_index += 2;
+            let data = &bytes[read_index..(read_index + data_len as usize)];
+            read_index += data_len as usize;
+            // println!("len={}, sps data:\n{}", data_len, bytes_hex_format(data));
+
+            let mut nalu_bytes: Vec<u8> = vec![0x00, 0x00, 0x00, 0x01];
+            nalu_bytes.extend_from_slice(data);
+            handle_nalu(nalu_bytes);
+        }
+        let num_of_pps = &bytes[read_index] & 0x1F;
+        read_index += 1;
+        // println!("pps num = {}", num_of_pps);
+        for _ in 0..num_of_pps as usize {
+            let data_len = BigEndian::read_u16(&bytes[read_index..]);
+            read_index += 2;
+            let data = &bytes[read_index..(read_index + data_len as usize)];
+            read_index += data_len as usize;
+            // println!("len={}, pps data:\n{}", data_len, bytes_hex_format(data));
+
+            let mut nalu_bytes: Vec<u8> = vec![0x00, 0x00, 0x00, 0x01];
+            nalu_bytes.extend_from_slice(data);
+            handle_nalu(nalu_bytes);
+        }
+    }
+    // One or more NALUs (Full frames are required)
+    else if acv_packet_type == 1 {
+        loop {
+            if read_index >= bytes.len() {
+                break;
+            }
+            let data_len = BigEndian::read_u32(&bytes[read_index..]);
+            read_index += 4;
+            let data = &bytes[read_index..(read_index + data_len as usize)];
+            read_index += data_len as usize;
+            // println!("NALU Type: {}, len={}", nalu_type_desc(&data[0]), data_len);
+            // println!("len={}, nalu data:\n{}", data_len, bytes_hex_format(data));
+
+            let mut nalu_bytes: Vec<u8> = vec![0x00, 0x00, 0x01];
+            nalu_bytes.extend_from_slice(data);
+            handle_nalu(nalu_bytes);
+        }
+    } else {
+        unreachable!("unknown acv packet type")
+    };
+
+    fn handle_nalu(nalu_bytes: Vec<u8>) {}
+}