From 1356bdda473f1213d971791fb1942dc25497de22 Mon Sep 17 00:00:00 2001
From: haorui <1905811497@qq.com>
Date: Mon, 11 Mar 2024 16:33:55 +0800
Subject: [PATCH] =?UTF-8?q?[what][bugfix][live]=20=E4=BF=AE=E5=A4=8D?=
 =?UTF-8?q?=E5=90=91=E6=9F=90=E4=BA=9B=20FMS3.0=20=E6=8E=A8=E6=B5=81?=
 =?UTF-8?q?=E5=A4=B1=E8=B4=A5=E9=97=AE=E9=A2=98?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

[how]
1. AMF 为简单类型时填插 AMF Null (参考 OBS 以及实际测试)
2. createStream 前附加 releaseStream 和 FCPublish, 兼容旧 FMS3.0
3. 正确处理 RTMP Header fmt 为 1 和 2 的业务逻辑
---
 src/Rtmp/RtmpProtocol.cpp | 27 ++++++++++++++++++++++-----
 src/Rtmp/RtmpProtocol.h   |  4 ++++
 src/Rtmp/RtmpPusher.cpp   | 30 ++++++++++++++++++++++--------
 3 files changed, 48 insertions(+), 13 deletions(-)

diff --git a/src/Rtmp/RtmpProtocol.cpp b/src/Rtmp/RtmpProtocol.cpp
index 010e30d4..800f0c9d 100644
--- a/src/Rtmp/RtmpProtocol.cpp
+++ b/src/Rtmp/RtmpProtocol.cpp
@@ -165,7 +165,14 @@ void RtmpProtocol::sendResponse(int type, const string &str) {
 
 void RtmpProtocol::sendInvoke(const string &cmd, const AMFValue &val) {
     AMFEncoder enc;
-    enc << cmd << ++_send_req_id << val;
+    if (val.type() == AMFType::AMF_OBJECT || val.type() == AMFType::AMF_NULL)
+    {
+        enc << cmd << ++_send_req_id << val;
+    }
+    else
+    {
+        enc << cmd << ++_send_req_id << AMFValue() << val;
+    }
     sendRequest(MSG_CMD, enc.data());
 }
 
@@ -615,16 +622,26 @@ const char* RtmpProtocol::handle_rtmp(const char *data, size_t len) {
         }
         auto &chunk_data = *now_packet;
         chunk_data.chunk_id = _now_chunk_id;
-        switch (header_len) {
-            case 12:
+        switch (header->fmt) {
+            case 0:
                 chunk_data.is_abs_stamp = true;
                 chunk_data.stream_index = load_le32(header->stream_index);
-            case 8:
+                _last_stream_index = chunk_data.stream_index;
+            case 1:
                 chunk_data.body_size = load_be24(header->body_size);
                 chunk_data.type_id = header->type_id;
-            case 4:
+                _last_body_size = chunk_data.body_size;
+                _last_type_id = chunk_data.type_id;
+            case 2:
                 chunk_data.ts_field = load_be24(header->time_stamp);
         }
+        switch (header->fmt) {
+            case 2:
+                chunk_data.type_id = _last_type_id;
+                chunk_data.body_size = _last_body_size;
+            case 1:
+                chunk_data.stream_index = _last_stream_index;
+        }
 
         auto time_stamp = chunk_data.ts_field;
         if (chunk_data.ts_field == 0xFFFFFF) {
diff --git a/src/Rtmp/RtmpProtocol.h b/src/Rtmp/RtmpProtocol.h
index 7ace93eb..2b7adfff 100644
--- a/src/Rtmp/RtmpProtocol.h
+++ b/src/Rtmp/RtmpProtocol.h
@@ -11,6 +11,7 @@
 #ifndef SRC_RTMP_RTMPPROTOCOL_H_
 #define SRC_RTMP_RTMPPROTOCOL_H_
 
+#include <cstdint>
 #include <memory>
 #include <string>
 #include <functional>
@@ -87,6 +88,9 @@ protected:
 private:
     bool _data_started = false;
     int _now_chunk_id = 0;
+    uint32_t _last_stream_index = 0;
+    size_t _last_body_size = 0;
+    uint8_t _last_type_id = 0;
     ////////////ChunkSize////////////
     size_t _chunk_size_in = DEFAULT_CHUNK_LEN;
     size_t _chunk_size_out = DEFAULT_CHUNK_LEN;
diff --git a/src/Rtmp/RtmpPusher.cpp b/src/Rtmp/RtmpPusher.cpp
index 92363241..8f3596fc 100644
--- a/src/Rtmp/RtmpPusher.cpp
+++ b/src/Rtmp/RtmpPusher.cpp
@@ -163,14 +163,28 @@ void RtmpPusher::send_connect() {
 }
 
 void RtmpPusher::send_createStream() {
-    AMFValue obj(AMF_NULL);
-    sendInvoke("createStream", obj);
-    addOnResultCB([this](AMFDecoder &dec) {
-        //TraceL << "createStream result";
-        dec.load<AMFValue>();
-        _stream_index = dec.load<int>();
-        send_publish();
-    });
+    // Workaround : 兼容较旧的 FMS3.0
+    {
+        {
+            AMFValue obj(_stream_id);
+            sendInvoke("releaseStream", obj);
+        }
+        {
+            AMFValue obj(_stream_id);
+            sendInvoke("FCPublish", obj);
+        }
+    }
+    {
+        AMFValue obj(AMF_NULL);
+        sendInvoke("createStream", obj);
+        addOnResultCB([this](AMFDecoder &dec) {
+            //TraceL << "createStream result";
+            dec.load<AMFValue>();
+            _stream_index = dec.load<int>();
+            send_publish();
+        });
+    }
+
 }
 
 #define RTMP_STREAM_LIVE    "live"