fix: support negative RTMP extended timestamp deltas (#2093)

Author: dimiden

src/projects/modules/rtmp/chunk/rtmp_chunk_parser.cpp

@@ -8,37 +8,32 @@
 //==============================================================================
 #pragma once
 
-#include <base/info/info.h>
-
-#include <deque>
-#include <map>
-#include <memory>
-
+#include "../rtmp_chunk_parser_common.h"
 #include "rtmp_datastructure.h"
 #include "rtmp_define.h"
 #include "rtmp_mux_util.h"
 
-class RtmpChunkParser
+class RtmpChunkParser : public RtmpChunkParserCommon<RtmpChunkHeader, RtmpMessage>
 {
 public:
-	enum class ParseResult
-	{
-		Error,
-		NeedMoreData,
-		Parsed,
-	};
-
-	enum class ParseResultForExtendedTimestamp
-	{
-		NeedMoreData,
-		Extended,
-		NotExtended,
-	};
+	using ParseResult = typename RtmpChunkParserCommon<RtmpChunkHeader, RtmpMessage>::ParseResult;
 
-public:
 	RtmpChunkParser(size_t chunk_size);
 	virtual ~RtmpChunkParser();
 
+	/// Parses as much of a single RTMP chunk as possible from the supplied data.
+	///
+	/// @param data Input bytes beginning at the next unread RTMP chunk boundary.
+	/// @param bytes_used Receives the number of bytes consumed from @p data.
+	///
+	/// @return `Parsed` when a chunk payload step completed, `NeedMoreData` when
+	///         more bytes are required, or `Error` when the chunk stream is
+	///         malformed.
+	///
+	/// @note When this returns `NeedMoreData`, @p bytes_used is set to `0` and
+	///       the caller must replay the same unconsumed prefix again with more
+	///       bytes appended. Header parsing is not resumable from the middle of
+	///       a partially received header.
 	ParseResult Parse(const std::shared_ptr<const ov::Data> &data, size_t *bytes_used);
 
 	std::shared_ptr<const RtmpMessage> GetMessage();
@@ -49,47 +44,46 @@ class RtmpChunkParser
 		_chunk_size = chunk_size;
 	}
 
-	info::NamePath GetNamePath() const;
-	void UpdateNamePath(const info::NamePath &stream_name_path);
-
 	void Destroy();
 
 private:
+	/// Returns the most recent completed header for a chunk stream.
+	///
+	/// @param chunk_stream_id RTMP chunk stream identifier.
+	///
+	/// @return The preceding parsed header for @p chunk_stream_id, or `nullptr`
+	///         if no header has been completed on that chunk stream yet.
 	std::shared_ptr<const RtmpChunkHeader> GetPrecedingChunkHeader(const uint32_t chunk_stream_id);
 
+	/// Checks whether the next header belongs to an unfinished message.
+	///
+	/// @param chunk_stream_id RTMP chunk stream identifier for the incoming chunk.
+	///
+	/// @return `true` when the parser is expecting a continuation chunk for the
+	///         same chunk stream, including interleaved pending-message cases.
+	bool IsContinuationChunk(const uint32_t chunk_stream_id) const;
+
+	/// Parses the RTMP basic header.
+	///
+	/// @param stream Input byte stream positioned at the RTMP basic header.
+	/// @param chunk_header Destination header object to populate.
+	///
+	/// @return `Parsed`, `NeedMoreData`, or `Error`.
 	ParseResult ParseBasicHeader(ov::ByteStream &stream, RtmpChunkHeader *chunk_header);
-	ParseResultForExtendedTimestamp ParseExtendedTimestamp(
-		const uint32_t stream_id,
-		ov::ByteStream &stream,
-		RtmpChunkHeader *chunk_header,
-		const int64_t timestamp,
-		RtmpChunkHeader::CompletedHeader *completed_header);
-	ParseResultForExtendedTimestamp ParseExtendedTimestampDelta(
-		const uint32_t stream_id,
-		ov::ByteStream &stream,
-		RtmpChunkHeader *chunk_header,
-		const int64_t preceding_timestamp,
-		const int64_t timestamp_delta,
-		RtmpChunkHeader::CompletedHeader *completed_header);
-	ParseResult ParseMessageHeader(ov::ByteStream &stream, RtmpChunkHeader *chunk_header);
-	ParseResult ParseHeader(ov::ByteStream &stream, RtmpChunkHeader *chunk_header);
-
-	int64_t CalculateRolledTimestamp(const uint32_t stream_id, const int64_t last_timestamp, int64_t parsed_timestamp);
-
-private:
-#if DEBUG
-	uint64_t _chunk_index = 0ULL;
-	uint64_t _total_read_bytes = 0ULL;
-#endif	// DEBUG
 
-	bool _need_to_parse_new_header = true;
-	std::shared_ptr<RtmpMessage> _current_message;
-	std::map<uint32_t, std::shared_ptr<RtmpMessage>> _pending_message_map;
-	std::map<uint32_t, std::shared_ptr<const RtmpChunkHeader>> _preceding_chunk_header_map;
-
-	ov::Queue<std::shared_ptr<const RtmpMessage>> _message_queue{nullptr, 500};
-	size_t _chunk_size;
+	/// Parses the RTMP message header for an already-parsed basic header.
+	///
+	/// @param stream Input byte stream positioned at the RTMP message header.
+	/// @param chunk_header Destination header object to populate.
+	///
+	/// @return `Parsed`, `NeedMoreData`, or `Error`.
+	ParseResult ParseMessageHeader(ov::ByteStream &stream, RtmpChunkHeader *chunk_header);
 
-	mutable std::mutex _name_path_mutex;
-	info::NamePath _name_path;
+	/// Parses both the RTMP basic header and message header.
+	///
+	/// @param stream Input byte stream positioned at the start of an RTMP header.
+	/// @param chunk_header Destination header object to populate.
+	///
+	/// @return `Parsed`, `NeedMoreData`, or `Error`.
+	ParseResult ParseHeader(ov::ByteStream &stream, RtmpChunkHeader *chunk_header);
 };

src/projects/modules/rtmp/chunk/rtmp_chunk_parser.h

@@ -23,8 +23,6 @@ enum class RtmpMessageHeaderType : uint8_t
 #pragma pack(push, 1)
 struct RtmpChunkHeader
 {
-	static constexpr const size_t EXTENDED_TIMESTAMP_SIZE = 4;
-
 	RtmpChunkHeader()
 	{
 		::memset(&message_header, 0, sizeof(message_header));
@@ -39,7 +37,7 @@ struct RtmpChunkHeader
 
 	// Indicates whether the timestamp is extended
 	bool is_extended_timestamp = false;
-	bool is_timestamp_delta = false;
+	bool is_timestamp_delta	   = false;
 
 	// Total chunk header size (basic header + message header + extended timestamp)
 	uint32_t GetTotalHeaderLength() const
@@ -48,16 +46,16 @@ struct RtmpChunkHeader
 	}
 
 #if DEBUG
-	uint64_t chunk_index = 0ULL;
+	uint64_t chunk_index				 = 0ULL;
 
-	uint64_t from_byte_offset = 0ULL;
+	uint64_t from_byte_offset			 = 0ULL;
 	mutable uint64_t message_total_bytes = 0ULL;
 #endif	// DEBUG
 
 	// 1 or 2 or 3
-	uint8_t basic_header_length = 0U;
+	uint8_t basic_header_length	   = 0U;
 	uint32_t message_header_length = 0U;
-	uint32_t message_length = 0U;
+	uint32_t message_length		   = 0U;
 
 	// uint32_t expected_type_3_header{};
 	// The payload size that including type 3 messages
@@ -75,16 +73,16 @@ struct RtmpChunkHeader
 
 	// Extended Timestamp/Timestamp delta
 	uint32_t extended_timestamp = 0U;
-	static_assert(sizeof(RtmpChunkHeader::extended_timestamp) == EXTENDED_TIMESTAMP_SIZE, "Extended timestamp size must be 4 bytes");
+	static_assert(sizeof(RtmpChunkHeader::extended_timestamp) == RTMP_EXTEND_TIMESTAMP_SIZE, "Extended timestamp size must be 4 bytes");
 
 	struct CompletedHeader
 	{
 		// Accumulated timestamp
-		int64_t timestamp = 0LL;
-		uint32_t timestamp_delta = 0U;
+		int64_t timestamp		  = 0LL;
+		uint32_t timestamp_delta  = 0U;
 
 		RtmpMessageTypeID type_id = RtmpMessageTypeID::Unknown;
-		uint32_t stream_id = 0U;
+		uint32_t stream_id		  = 0U;
 	} completed;
 
 	// Message Header

src/projects/modules/rtmp/chunk/rtmp_datastructure.h

@@ -0,0 +1,56 @@
+//==============================================================================
+//
+//  OvenMediaEngine
+//
+//  Created by Hyunjun Jang
+//  Copyright (c) 2026 AirenSoft. All rights reserved.
+//
+//==============================================================================
+#pragma once
+
+#include <base/info/info.h>
+#include <base/ovlibrary/queue.h>
+
+template <typename Theader, typename Tmessage>
+class RtmpChunkParserCommon
+{
+public:
+	enum class ParseResult
+	{
+		Error,
+		NeedMoreData,
+		Parsed,
+	};
+
+public:
+	info::NamePath GetNamePath() const
+	{
+		std::lock_guard lock_guard(_name_path_mutex);
+		return _name_path;
+	}
+
+	virtual void UpdateNamePath(const info::NamePath &stream_name_path)
+	{
+		std::lock_guard lock_guard(_name_path_mutex);
+		_name_path = stream_name_path;
+		_message_queue.SetAlias(ov::String::FormatString("RTMP queue for %s", _name_path.CStr()));
+	}
+
+protected:
+	bool _need_to_parse_new_header = true;
+	std::shared_ptr<Tmessage> _current_message;
+	std::map<uint32_t, std::shared_ptr<Tmessage>> _pending_message_map;
+
+#if DEBUG
+	uint64_t _chunk_index	   = 0ULL;
+	uint64_t _total_read_bytes = 0ULL;
+#endif	// DEBUG
+
+	std::map<uint32_t, std::shared_ptr<const Theader>> _preceding_chunk_header_map;
+	ov::Queue<std::shared_ptr<const Tmessage>> _message_queue{nullptr, 500};
+	size_t _chunk_size = 0;
+
+private:
+	mutable std::mutex _name_path_mutex;
+	info::NamePath _name_path;
+};