Skip to content

feat: robust RTP depacketizer — STAP-A/AP, CSRC/extension, loss handling (#10) #47

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
compot-dev merged 3 commits into from
Oct 12, 2026
Merged

feat: robust RTP depacketizer — STAP-A/AP, CSRC/extension, loss handling (#10) #47

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
87c31ec
feat: robust RTP depacketizer for H.264/H.265 video
compot-dev Oct 12, 2026
cc72cde
fix: address review findings for the depacketizer
compot-dev Oct 12, 2026
13bc771
docs: drop fixed #10 from known limitations
compot-dev Oct 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 0 additions & 1 deletion README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,7 +92,6 @@ Shows all configured slots in a 2×2 grid. Each cell has its own connection and
> These are known gaps that we plan to address. Each one has an issue — comments, test results and pull requests are welcome.
>
> - **RTSP over TCP only.** No UDP or multicast — [#7](https://github.com/OpenIPC/decoder/issues/7)
> - **Simplified RTP handling for video:** no STAP-A/AP aggregation, packets with CSRC or header extensions are dropped, no packet-loss recovery — [#10](https://github.com/OpenIPC/decoder/issues/10)
> - **No audio in Quad mode** — [#12](https://github.com/OpenIPC/decoder/issues/12)
> - **Only 4 camera slots** — [#13](https://github.com/OpenIPC/decoder/issues/13)
> - **Audio formats not supported:** MP4A-LATM, Opus, G.726, and AAC frames split across RTP packets — [#14](https://github.com/OpenIPC/decoder/issues/14)
Expand Down
1 change: 0 additions & 1 deletion README.ru.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,7 +92,6 @@
> Это известные пробелы, которые мы планируем закрыть. На каждый заведён issue — комментарии, результаты проверок и pull request приветствуются.
>
> - **Только RTSP поверх TCP.** UDP и multicast не поддерживаются — [#7](https://github.com/OpenIPC/decoder/issues/7)
> - **Упрощённая обработка RTP для видео:** нет агрегации STAP-A/AP, пакеты с CSRC или расширением заголовка отбрасываются, потери пакетов не обрабатываются — [#10](https://github.com/OpenIPC/decoder/issues/10)
> - **В режиме Quad нет звука** — [#12](https://github.com/OpenIPC/decoder/issues/12)
> - **Только 4 слота для камер** — [#13](https://github.com/OpenIPC/decoder/issues/13)
> - **Не поддерживаются форматы звука** MP4A-LATM, Opus, G.726, а также AAC-кадры, разбитые на несколько RTP-пакетов — [#14](https://github.com/OpenIPC/decoder/issues/14)
Expand Down
1 change: 0 additions & 1 deletion README.zh-CN.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,7 +92,6 @@
> 以下是我们计划解决的已知不足，每一项都有对应的 issue。欢迎留言、提交测试结果和 pull request。
>
> - **仅支持 RTSP over TCP**，不支持 UDP 和组播 — [#7](https://github.com/OpenIPC/decoder/issues/7)
> - **视频的 RTP 处理较为简单：** 不支持 STAP-A/AP 聚合包，带 CSRC 或扩展头的包会被丢弃，也不处理丢包 — [#10](https://github.com/OpenIPC/decoder/issues/10)
> - **Quad 模式下没有声音** — [#12](https://github.com/OpenIPC/decoder/issues/12)
> - **仅有 4 个摄像机槽位** — [#13](https://github.com/OpenIPC/decoder/issues/13)
> - **不支持的音频格式：** MP4A-LATM、Opus、G.726，以及跨多个 RTP 包分片的 AAC 帧 — [#14](https://github.com/OpenIPC/decoder/issues/14)
Expand Down
235 changes: 160 additions & 75 deletions android/app/src/main/java/com/openipc/decoder/NalAssembler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
/*
* Copyright (c) OpenIPC https://openipc.org Prosperity Public License 3.0.0
*
* NalAssembler.java — reusable NAL unit reassembler from RTP fragmentation units
* NalAssembler.java — H.264/H.265 NAL units from RTP packets (RFC 6184 / RFC 7798)
*
*/

Expand All @@ -10,109 +10,194 @@
import android.util.Log;

/**
* Reassembles NAL units from RTP fragmentation units (FU-A for H.264, FU for H.265).
* Shared by the main decoder and QuadCell streams. Replaces duplicated buildFrame()/assembleNal().
* Turns RTP video packets into Annex B NAL units (with a 4-byte start code). Handles single
* NAL unit packets, fragmentation units (FU-A, FU) and aggregation packets (STAP-A, AP),
* any RTP header length (CSRC, extension, padding) and packet loss: a fragmented NAL unit
* with a missing piece is dropped instead of being passed on corrupted.
*/
class NalAssembler {
private static final String TAG = "OpenIPCDecoder";

private static final int RTP_FU_H264 = 28;
private static final int RTP_FU_H265 = 49;
private static final int H264_STAP_A = 24;
private static final int H264_FU_A = 28;
private static final int H265_AP = 48;
private static final int H265_FU = 49;
private static final int H265_PACI = 50;

interface Sink {
/** A complete NAL unit; the sink owns the frame (recycle it when done). */
void onNal(Frame nal);
}

private final byte[] nalBuffer;
private int nalSize;
private boolean lastCodec;
private final Runnable onCodecSwitch;
private final FramePool framePool;

private int nalSize;
/** A fragmented NAL unit is being collected in nalBuffer. */
private boolean inFragment;
/** Skipping the rest of a fragmented NAL unit whose start was not seen (counted once). */
private boolean orphan;
private int lastSeq = -1;
private boolean lastCodec;
private int dropped;

NalAssembler(int bufferSize, Runnable onCodecSwitch, FramePool framePool) {
this.nalBuffer = new byte[bufferSize];
this.onCodecSwitch = onCodecSwitch;
this.framePool = framePool;
}

void reset() { nalSize = 0; }
void reset() {
nalSize = 0;
inFragment = false;
orphan = false;
lastSeq = -1;
}

/** Reassemble one RTP packet into (possibly null) NAL frame. */
Frame assemble(Frame frame, boolean codecH265) {
byte[] rx = frame.data();
int rxSize = frame.length();
int cp = 12;
rxSize -= cp;
if (rxSize <= 0) return null;
/** NAL units dropped so far because of packet loss or size (for diagnostics). */
int droppedCount() { return dropped; }

/**
* Process one RTP packet ({@code length} bytes at {@code offset}, header included).
* Returns false if the RTP header is malformed.
*/
boolean assemble(byte[] buf, int offset, int length, boolean codecH265, Sink sink) {
int header = RtspSession.rtpHeaderLength(buf, offset, length);
if (header < 0) return false;
int end = RtspSession.rtpPayloadEnd(buf, offset, length);
if (end < header) return false;
int start = offset + header;
end += offset;

if (lastCodec != codecH265) {
lastCodec = codecH265;
nalSize = 0;
reset();
onCodecSwitch.run();
}

int frag = fragment(rx[cp], codecH265);
if (frag == RTP_FU_H264 || frag == RTP_FU_H265) {
int minPayload = codecH265 ? 3 : 2;
if (rxSize < minPayload) return null;
int seq = ((buf[offset + 2] & 0xFF) << 8) | (buf[offset + 3] & 0xFF);
boolean gap = lastSeq >= 0 && seq != ((lastSeq + 1) & 0xFFFF);
lastSeq = seq;
if (gap && inFragment) {
drop("packet loss inside a fragmented NAL unit");
orphan = true; // its remaining pieces belong to the unit just counted
}

int staBit, endBit;
if (codecH265) {
staBit = rx[cp + 2] & 0x80;
endBit = rx[cp + 2] & 0x40;
} else {
staBit = rx[cp + 1] & 0x80;
endBit = rx[cp + 1] & 0x40;
}
int nalHeader = codecH265 ? 2 : 1;
if (end - start < nalHeader) return true;
int type = fragment(buf[start], codecH265);

if (codecH265 ? type == H265_FU : type == H264_FU_A) {
fragmentUnit(buf, start, end, codecH265, sink);
return true;
}
// Any other packet ends a fragmented NAL unit that did not finish; its remaining
// pieces, if any arrive, belong to the unit counted here
if (inFragment) {
drop("fragmented NAL unit interrupted by another packet");
orphan = true;
} else {
orphan = false;
}
if (codecH265 ? type == H265_AP : type == H264_STAP_A) {
aggregation(buf, start + nalHeader, end, sink);
} else if (codecH265 ? type == H265_PACI : (type >= 25 && type <= 29)) {
// PACI, STAP-B, MTAP and FU-B are not used by cameras in non-interleaved mode
return true;
} else {
emit(buf, start, end - start, sink);
}
return true;
}

private void fragmentUnit(byte[] buf, int start, int end, boolean h265, Sink sink) {
int payloadHeader = h265 ? 2 : 1;
if (end - start <= payloadHeader) return;
int fu = buf[start + payloadHeader] & 0xFF;
boolean first = (fu & 0x80) != 0;
boolean last = (fu & 0x40) != 0;
int pos = start + payloadHeader + 1;

if (staBit != 0) {
nalSize = 0;
if (codecH265) {
nalBuffer[nalSize++] = 0;
nalBuffer[nalSize++] = 0;
nalBuffer[nalSize++] = 0;
nalBuffer[nalSize++] = 1;
// 2-byte NAL header: F + FU type + layer-id MSB, then layer-id/TID as sent
nalBuffer[nalSize++] = (byte)((rx[cp] & 0x81) | ((rx[cp + 2] & 0x3F) << 1));
nalBuffer[nalSize++] = rx[cp + 1];
cp += 3; // payload header + FU header
} else {
nalBuffer[nalSize++] = 0;
nalBuffer[nalSize++] = 0;
nalBuffer[nalSize++] = 0;
nalBuffer[nalSize++] = 1;
nalBuffer[nalSize++] = (byte)((rx[cp] & 0xE0) | (rx[cp + 1] & 0x1F));
cp += 2;
}
rxSize = frame.length() - cp;
if (first) {
if (inFragment) drop("fragmented NAL unit without its end");
orphan = false;
nalSize = 0;
inFragment = true;
nalBuffer[nalSize++] = 0;
nalBuffer[nalSize++] = 0;
nalBuffer[nalSize++] = 0;
nalBuffer[nalSize++] = 1;
if (h265) {
// 2-byte NAL header: F + FU type + layer-id MSB, then layer-id/TID as sent
nalBuffer[nalSize++] = (byte) ((buf[start] & 0x81) | ((fu & 0x3F) << 1));
nalBuffer[nalSize++] = buf[start + 1];
} else {
cp += codecH265 ? 3 : 2;
rxSize = frame.length() - cp;
nalBuffer[nalSize++] = (byte) ((buf[start] & 0xE0) | (fu & 0x1F));
}

int copyLen = Math.min(rxSize, nalBuffer.length - nalSize);
if (copyLen > 0) {
System.arraycopy(rx, cp, nalBuffer, nalSize, copyLen);
nalSize += copyLen;
} else if (!inFragment) {
// The start was lost (or we joined mid-unit): nothing to attach this piece to.
// Count the lost unit once, not once per remaining piece.
if (!orphan) {
orphan = true;
dropped++;
logDrop("fragmented NAL unit without its start");
}
if (last) orphan = false;
return;
Comment thread
qodo-free-for-open-source-projects[bot] marked this conversation as resolved.
}

if (endBit != 0) {
Frame out = framePool.obtain(nalSize);
System.arraycopy(nalBuffer, 0, out.data(), 0, nalSize);
out.setLength(nalSize);
nalSize = 0;
return out;
}
return null;
} else {
int len = end - pos;
if (nalSize + len > nalBuffer.length) {
drop("NAL unit larger than " + nalBuffer.length + " bytes");
return;
}
System.arraycopy(buf, pos, nalBuffer, nalSize, len);
nalSize += len;

if (last) {
Frame out = framePool.obtain(nalSize);
System.arraycopy(nalBuffer, 0, out.data(), 0, nalSize);
out.setLength(nalSize);
nalSize = 0;
int copyLen = rxSize + 4;
if (copyLen > nalBuffer.length || copyLen < 5) return null;
nalBuffer[0] = 0;
nalBuffer[1] = 0;
nalBuffer[2] = 0;
nalBuffer[3] = 1;
System.arraycopy(rx, cp, nalBuffer, 4, rxSize);
Frame out = framePool.obtain(copyLen);
System.arraycopy(nalBuffer, 0, out.data(), 0, copyLen);
out.setLength(copyLen);
return out;
inFragment = false;
sink.onNal(out);
}
}

/** STAP-A / AP: a sequence of (16-bit size, NAL unit). */
private void aggregation(byte[] buf, int pos, int end, Sink sink) {
while (end - pos >= 2) {
int size = ((buf[pos] & 0xFF) << 8) | (buf[pos + 1] & 0xFF);
pos += 2;
if (size == 0 || pos + size > end) break;
emit(buf, pos, size, sink);
pos += size;
Comment thread
qodo-free-for-open-source-projects[bot] marked this conversation as resolved.
}
}

private void emit(byte[] buf, int pos, int size, Sink sink) {
Frame out = framePool.obtain(size + 4);
byte[] d = out.data();
d[0] = 0;
d[1] = 0;
d[2] = 0;
d[3] = 1;
System.arraycopy(buf, pos, d, 4, size);
out.setLength(size + 4);
sink.onNal(out);
}

private void drop(String reason) {
nalSize = 0;
inFragment = false;
dropped++;
logDrop(reason);
}

private void logDrop(String reason) {
if (dropped <= 3 || dropped % 100 == 0) {
Log.w(TAG, "NAL unit dropped (" + dropped + "): " + reason);
}
}

Expand Down
34 changes: 21 additions & 13 deletions android/app/src/main/java/com/openipc/decoder/QuadCell.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,8 @@ class QuadCell implements RtspSession.Callback {

private final BlockingQueue<Frame> nalQueue = new ArrayBlockingQueue<>(30);
private final NalAssembler nalAssembler;
private final NalAssembler.Sink nalSink;
private int queueFullDrops;
private int lastUnknownPayload = -1;

private ExecutorService executor;
Expand All @@ -61,6 +63,15 @@ interface Listener {
this.userAgent = userAgent;
this.listener = listener;
this.session = new RtspSession(tag, this);
this.nalSink = nal -> {
if (!nalQueue.offer(nal)) {
// Rate-limited: one aggregation packet can carry many NAL units
if (queueFullDrops++ % 100 == 0) {
Log.w(TAG, tag + " queue full, NAL dropped (" + queueFullDrops + ")");
}
pool.recycle(nal);
}
};
this.nalAssembler = new NalAssembler(512 * 1024, () -> {
nalQueue.clear();
synchronized (decoderLock) {
Expand Down Expand Up @@ -129,7 +140,14 @@ private void startThreads() {
while (running) {
try {
Frame f = nalQueue.poll(5, TimeUnit.MILLISECONDS);
if (f != null) decode(f);
if (f != null) {
try {
decode(f);
} finally {
// The decoder copied the data: give the buffer back to the pool
cellFramePool.recycle(f);
}
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;
Expand Down Expand Up @@ -189,8 +207,6 @@ public void onSdp(Sdp sdp) {
@Override
public boolean onPacket(int channel, byte[] buf, int offset, int length) {
if (channel != RtspSession.CHANNEL_VIDEO || length < 12) return false;
// CSRC and header extensions are not supported for video yet (#10)
if ((buf[offset] & 0x0F) != 0 || (buf[offset] & 0x10) != 0) return false;
int pt = buf[offset + 1] & 0x7F;
if (pt != videoPt) {
if (pt != lastUnknownPayload) {
Expand All @@ -200,18 +216,10 @@ public boolean onPacket(int channel, byte[] buf, int offset, int length) {
return false;
}

Frame frame = cellFramePool.obtain(length);
System.arraycopy(buf, offset, frame.data(), 0, length);
frame.setLength(length);
Frame output = nalAssembler.assemble(frame, codecH265);
cellFramePool.recycle(frame);
if (output != null && !nalQueue.offer(output)) {
Log.w(TAG, tag + " queue full, dropping frame");
cellFramePool.recycle(output);
}
return true;
return nalAssembler.assemble(buf, offset, length, codecH265, nalSink);
}


private void decode(Frame buffer) {
if (buffer.length() < 5) return;

Expand Down
Loading
Loading