Skip to content

Commit 7034d2a

Browse files
vertexodessavertexodessa
authored
Implement BufferedPacketQueue (#14)
1 parent 06c6e26 commit 7034d2a

File tree

3 files changed

+185
-8
lines changed

3 files changed

+185
-8
lines changed

‎app/videonative/src/main/cpp/BufferedPacketQueue.h

+166
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,166 @@
1+
#pragma once
2+
#include <map>
3+
#define BUFFERED_QUEUE_LOG_TAG "BufferedPacketQueue"
4+
5+
/**
6+
* @class BufferedPacketQueue
7+
* @brief A queue for managing and processing network packets in sequence order.
8+
* This class leverages a buffer to handle out-of-order packets and ensures
9+
* packets are processed sequentially.
10+
*/
11+
class BufferedPacketQueue {
12+
// Using std::map to keep packets sorted by sequence number
13+
template <typename SeqType> class PacketBuffer {
14+
public:
15+
std::map<SeqType, std::vector<uint8_t>> packets;
16+
SeqType lastPacketIdx;
17+
static constexpr std::size_t MAX_BUFFER_SIZE = 20;
18+
bool firstPacket;
19+
20+
// Constructor with logging
21+
PacketBuffer() : lastPacketIdx(0), firstPacket(true) {
22+
__android_log_print(ANDROID_LOG_DEBUG, BUFFERED_QUEUE_LOG_TAG, "PacketBuffer initialized.");
23+
}
24+
25+
/**
26+
* @brief Compares two sequence numbers considering wrap-around.
27+
* @param a First sequence number.
28+
* @param b Second sequence number.
29+
* @return True if sequence a is less than b, accounting for wrap-around.
30+
*/
31+
bool seqLessThan(SeqType a, SeqType b) const {
32+
// Calculate the midpoint based on the sequence number width
33+
const SeqType midpoint = (std::numeric_limits<SeqType>::max() / 2) + 1;
34+
bool result =
35+
((b > a) && (b - a < midpoint)) || ((a > b) && (a - b > midpoint));
36+
// __android_log_print(ANDROID_LOG_VERBOSE, BUFFERED_QUEUE_LOG_TAG, "Comparing
37+
// sequences: a=%u, b=%u, a < b? %s",
38+
// a, b, result ? "true" : "false");
39+
return result;
40+
}
41+
42+
/**
43+
* @brief Processes an incoming packet based on its sequence index.
44+
* @param idx Sequence index of the packet.
45+
* @param data Pointer to packet data.
46+
* @param data_length Size of packet data.
47+
* @param callback Callable to handle processed packets.
48+
*/
49+
template <typename Callback>
50+
void processPacket(SeqType idx, const uint8_t *data,
51+
std::size_t data_length, Callback &callback) {
52+
// __android_log_print(ANDROID_LOG_DEBUG, BUFFERED_QUEUE_LOG_TAG, "Processing
53+
// packet with Sequence=%u", idx);
54+
55+
if (firstPacket) {
56+
// Initialize lastPacketIdx to one before the first packet
57+
lastPacketIdx = idx - 1;
58+
firstPacket = false;
59+
__android_log_print(
60+
ANDROID_LOG_DEBUG, BUFFERED_QUEUE_LOG_TAG,
61+
"First packet received. Initialized lastPacketIdx to %u",
62+
lastPacketIdx);
63+
}
64+
65+
if (idx == lastPacketIdx + 1) {
66+
// Packet is the next expected one
67+
// __android_log_print(ANDROID_LOG_DEBUG, BUFFERED_QUEUE_LOG_TAG, "In-order
68+
// packet detected. Processing immediately.");
69+
callback(data, data_length);
70+
lastPacketIdx = idx;
71+
// __android_log_print(ANDROID_LOG_DEBUG, BUFFERED_QUEUE_LOG_TAG, "Updated
72+
// lastPacketIdx to %u", lastPacketIdx);
73+
74+
// Now check if the buffer has the next packets
75+
while (true) {
76+
auto it = packets.find(lastPacketIdx + 1);
77+
if (it != packets.end()) {
78+
__android_log_print(
79+
ANDROID_LOG_DEBUG, BUFFERED_QUEUE_LOG_TAG,
80+
"Found buffered packet with Sequence=%u. Processing.",
81+
it->first);
82+
callback(it->second.data(), it->second.size());
83+
lastPacketIdx = it->first;
84+
__android_log_print(
85+
ANDROID_LOG_DEBUG, BUFFERED_QUEUE_LOG_TAG,
86+
"Updated lastPacketIdx to %u after processing buffered packet.",
87+
lastPacketIdx);
88+
packets.erase(it);
89+
} else {
90+
//__android_log_print(ANDROID_LOG_DEBUG, BUFFERED_QUEUE_LOG_TAG, "No buffered packet
91+
// found for Sequence=%u.", lastPacketIdx + 1);
92+
break;
93+
}
94+
}
95+
} else if (seqLessThan(lastPacketIdx, idx)) {
96+
// Out-of-order packet
97+
__android_log_print(ANDROID_LOG_DEBUG, BUFFERED_QUEUE_LOG_TAG,
98+
"Out-of-order packet detected. Sequence=%u", idx);
99+
// Avoid duplicate packets
100+
if (packets.find(idx) == packets.end()) {
101+
// Buffer the packet
102+
packets[idx] = std::vector<uint8_t>(data, data + data_length);
103+
__android_log_print(ANDROID_LOG_DEBUG, BUFFERED_QUEUE_LOG_TAG,
104+
"Buffered out-of-order packet. Buffer size: %zu",
105+
packets.size());
106+
107+
// If buffer size exceeds MAX_BUFFER_SIZE, process all buffered
108+
// packets
109+
if (packets.size() >= MAX_BUFFER_SIZE) {
110+
__android_log_print(ANDROID_LOG_WARN, BUFFERED_QUEUE_LOG_TAG,
111+
"Buffer size exceeded MAX_BUFFER_SIZE (%zu). "
112+
"Processing all buffered packets.",
113+
MAX_BUFFER_SIZE);
114+
// Process buffered packets in order
115+
auto it_buffer = packets.begin();
116+
while (it_buffer != packets.end()) {
117+
__android_log_print(ANDROID_LOG_DEBUG, BUFFERED_QUEUE_LOG_TAG,
118+
"Processing buffered packet with Sequence=%u",
119+
it_buffer->first);
120+
callback(it_buffer->second.data(), it_buffer->second.size());
121+
lastPacketIdx = it_buffer->first;
122+
it_buffer = packets.erase(it_buffer);
123+
__android_log_print(ANDROID_LOG_DEBUG, BUFFERED_QUEUE_LOG_TAG,
124+
"Updated lastPacketIdx to %u after "
125+
"processing buffered packet.",
126+
lastPacketIdx);
127+
}
128+
}
129+
} else {
130+
__android_log_print(
131+
ANDROID_LOG_WARN, BUFFERED_QUEUE_LOG_TAG,
132+
"Duplicate packet received with Sequence=%u. Ignoring.", idx);
133+
}
134+
} else {
135+
// Packet is older than lastPacketIdx, possibly a retransmission or
136+
// duplicate
137+
__android_log_print(
138+
ANDROID_LOG_WARN, BUFFERED_QUEUE_LOG_TAG,
139+
"Received old or duplicate packet with Sequence=%u. Ignoring.",
140+
idx);
141+
// Optionally, handle retransmissions or request retransmission here
142+
}
143+
}
144+
};
145+
146+
// Decide whether to use uint8_t or uint16_t based on sequence number type
147+
using SeqType = uint16_t; // Change to uint8_t if sequence numbers are 8-bit
148+
149+
PacketBuffer<SeqType> buffer;
150+
151+
public:
152+
/**
153+
* @brief Process a packet through the queue.
154+
* @param idx Sequence index of the packet.
155+
* @param data Pointer to packet data.
156+
* @param data_length Size of the packet data.
157+
* @param callback Callable to handle processed packets.
158+
*/
159+
template <typename Callback>
160+
void processPacket(SeqType idx, const uint8_t *data, std::size_t data_length,
161+
Callback &callback) {
162+
//__android_log_print(ANDROID_LOG_DEBUG, BUFFERED_QUEUE_LOG_TAG, "BufferedPacketQueue:
163+
// Processing packet with Sequence=%u", idx);
164+
buffer.processPacket(idx, data, data_length, callback);
165+
}
166+
};

‎app/videonative/src/main/cpp/VideoPlayer.cpp

+16-8
Original file line numberDiff line numberDiff line change
@@ -93,15 +93,23 @@ void VideoPlayer::processQueue() {
9393
}
9494

9595
//Not yet parsed bit stream (e.g. raw h264 or rtp data)
96-
void VideoPlayer::onNewRTPData(const uint8_t *data, const std::size_t data_length) {
96+
void VideoPlayer::onNewRTPData(const uint8_t* data, const std::size_t data_length) {
97+
// Parse the RTP packet
9798
const RTP::RTPPacket rtpPacket(data, data_length);
98-
if (rtpPacket.header.payload == RTP_PAYLOAD_TYPE_AUDIO) {
99-
audioDecoder.enqueueAudio(data, data_length);
100-
}
101-
else
102-
{
103-
mParser.parse_rtp_stream(data, data_length);
104-
}
99+
uint16_t idx = rtpPacket.header.getSequence();
100+
101+
// Define the callback based on payload type
102+
auto callback = [&](const uint8_t* packet_data, std::size_t packet_length) {
103+
if (rtpPacket.header.payload == RTP_PAYLOAD_TYPE_AUDIO) {
104+
audioDecoder.enqueueAudio(packet_data, packet_length);
105+
}
106+
else {
107+
mParser.parse_rtp_stream(packet_data, packet_length);
108+
}
109+
};
110+
111+
// Process the packet using the queue
112+
mBufferedPacketQueue.processPacket(idx, data, data_length, callback);
105113
}
106114

107115
void VideoPlayer::onNewNALU(const NALU &nalu) {

‎app/videonative/src/main/cpp/VideoPlayer.h

+3
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
#include <jni.h>
88
#include "VideoDecoder.h"
99
#include "AudioDecoder.h"
10+
#include "BufferedPacketQueue.h"
1011
#include "UdpReceiver.h"
1112
#include "parser/H26XParser.h"
1213
#include "minimp4.h"
@@ -65,6 +66,8 @@ class VideoPlayer {
6566
const std::string GROUND_RECORDING_DIRECTORY;
6667
JavaVM *javaVm = nullptr;
6768
H26XParser mParser;
69+
BufferedPacketQueue mBufferedPacketQueue;
70+
6871

6972
// DVR attributes
7073
int dvr_fd;

0 commit comments

Comments
 (0)