From dfb8f05d8ec0a19dd47a63147c2f973b90efe813 Mon Sep 17 00:00:00 2001 From: Compot Developer Date: Mon, 12 Oct 2026 18:58:37 +0000 Subject: [PATCH 1/4] feat: receive plain RTP from Majestic outgoing over UDP, unicast and multicast (#68) A slot URL of the form udp://[source@][address]:port[?codec=h264|h265&pt=N] (rtp:// works too, the syntax is VLC's) receives the stream a camera pushes, with no RTSP: Majestic "outgoing" destinations send RTP with payload type 96 (H.264) or 97 (H.265) and repeat the parameter sets before every IDR, so the stream can be joined at any time without an SDP. - udp://@:5600 listens on a port; udp://@239.1.2.3:5600 joins the group; udp://192.168.1.10@239.1.2.3:5600 accepts that sender only. - RtspSession runs it in place of the RTSP handshake, so the watchdog, reconnection and status are the same as for RTSP slots. - The codec follows the payload type and a switch between 96 and 97 on the camera is followed; codec= and pt= force them for other senders. - RtpListener uses a MulticastSocket (DatagramChannel.join needs API 24) and the reorder buffer of #67. Unicast ports are bound exclusively, so a second slot on the same port fails instead of receiving nothing. - Wi-Fi drivers drop multicast unless an app holds a MulticastLock: it is held while any slot receives a group (CHANGE_WIFI_MULTICAST_STATE). --- android/app/src/main/AndroidManifest.xml | 2 + .../main/java/com/openipc/decoder/App.java | 5 + .../java/com/openipc/decoder/RtpListener.java | 266 ++++++++++++++++++ .../java/com/openipc/decoder/RtspSession.java | 67 ++++- .../com/openipc/decoder/RtpListenerTest.java | 238 ++++++++++++++++ 5 files changed, 577 insertions(+), 1 deletion(-) create mode 100644 android/app/src/main/java/com/openipc/decoder/RtpListener.java create mode 100644 android/app/src/test/java/com/openipc/decoder/RtpListenerTest.java diff --git a/android/app/src/main/AndroidManifest.xml b/android/app/src/main/AndroidManifest.xml index 521986a3..b8703dc1 100755 --- a/android/app/src/main/AndroidManifest.xml +++ b/android/app/src/main/AndroidManifest.xml @@ -2,6 +2,8 @@ + + { Log.e("OpenIPCDecoder", "Uncaught exception", throwable); diff --git a/android/app/src/main/java/com/openipc/decoder/RtpListener.java b/android/app/src/main/java/com/openipc/decoder/RtpListener.java new file mode 100644 index 00000000..decaa6bd --- /dev/null +++ b/android/app/src/main/java/com/openipc/decoder/RtpListener.java @@ -0,0 +1,266 @@ +/* + * Copyright (c) OpenIPC https://openipc.org Prosperity Public License 3.0.0 + * + * RtpListener.java — plain RTP pushed to us over UDP, unicast or multicast, without RTSP + * + */ + +package com.openipc.decoder; + +import android.content.Context; +import android.net.wifi.WifiManager; +import android.util.Log; + +import java.io.Closeable; +import java.io.IOException; +import java.net.DatagramPacket; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.MulticastSocket; +import java.net.SocketTimeoutException; +import java.util.Locale; + +/** + * Receives the stream that Majestic sends to "outgoing" destinations (udp://host:port): + * RTP with payload type 96 (H.264) or 97 (H.265) and the parameter sets repeated in band + * before every IDR, so it can be joined at any time without an SDP (#68). + * A MulticastSocket is used for unicast too: joining a group from a DatagramChannel needs + * API 24. + */ +class RtpListener implements Closeable { + private static final String TAG = "OpenIPCDecoder"; + private static final int RCVBUF = 2 * 1024 * 1024; + + interface Handler { + void onPacket(byte[] buf, int offset, int length); + } + + /** + * udp://[source@][address]:port[?codec=h264|h265][&pt=N], as VLC writes it; rtp:// is + * accepted too. An empty or local address listens on that port, a multicast address + * joins the group, and a source accepts datagrams from that sender only. + */ + static final class Url { + /** Multicast group or local address to bind; null for every local address. */ + InetAddress address; + int port; + /** Only sender accepted, or null. */ + InetAddress source; + /** Codec forced by the URL: null (by payload type), "H264" or "H265". */ + String codec; + /** Video payload type forced by the URL, or -1. */ + int pt = -1; + + boolean multicast() { return address != null && address.isMulticastAddress(); } + + static boolean matches(String url) { + if (url == null) return false; + String u = url.trim().toLowerCase(Locale.ROOT); + return u.startsWith("udp://") || u.startsWith("rtp://"); + } + + static Url parse(String url) throws IOException { + if (!matches(url)) throw new IOException("Not a udp:// or rtp:// URL"); + String s = url.trim(); + s = s.substring(s.indexOf("://") + 3); + String query = ""; + int q = s.indexOf('?'); + if (q >= 0) { + query = s.substring(q + 1); + s = s.substring(0, q); + } + if (s.endsWith("/")) s = s.substring(0, s.length() - 1); + + Url u = new Url(); + int at = s.lastIndexOf('@'); + if (at >= 0) { + String src = s.substring(0, at); + if (!src.isEmpty()) u.source = literal(stripBrackets(src), "source"); + s = s.substring(at + 1); + } + String host; + String port; + if (s.startsWith("[")) { + int close = s.indexOf(']'); + if (close < 0 || close + 1 >= s.length() || s.charAt(close + 1) != ':') { + throw new IOException("Invalid address in " + s); + } + host = s.substring(1, close); + port = s.substring(close + 2); + } else { + int colon = s.lastIndexOf(':'); + if (colon < 0) throw new IOException("A port is required, e.g. udp://@:5600"); + host = s.substring(0, colon); + port = s.substring(colon + 1); + } + try { + u.port = Integer.parseInt(port); + } catch (NumberFormatException e) { + u.port = -1; + } + if (u.port <= 0 || u.port > 65535) throw new IOException("Invalid port: " + port); + if (!host.isEmpty()) u.address = literal(host, "address"); + + for (String param : query.split("&")) { + int eq = param.indexOf('='); + if (eq < 0) continue; + String name = param.substring(0, eq).toLowerCase(Locale.ROOT); + String value = param.substring(eq + 1).toLowerCase(Locale.ROOT); + if (name.equals("codec")) { + if (value.equals("h264") || value.equals("avc")) { + u.codec = "H264"; + } else if (value.equals("h265") || value.equals("hevc")) { + u.codec = "H265"; + } else { + throw new IOException("Unknown codec: " + value); + } + } else if (name.equals("pt")) { + try { + u.pt = Integer.parseInt(value); + } catch (NumberFormatException ignored) {} + if (u.pt < 0 || u.pt > 127) throw new IOException("Invalid pt: " + value); + } + } + return u; + } + + private static String stripBrackets(String s) { + return s.startsWith("[") && s.endsWith("]") ? s.substring(1, s.length() - 1) : s; + } + + /** Numeric addresses only: no name lookup on a slot URL. */ + private static InetAddress literal(String s, String what) throws IOException { + if (!s.matches("[0-9.]+|[0-9A-Fa-f:.]*:[0-9A-Fa-f:.]*")) { + throw new IOException("The " + what + " must be an IP address: " + s); + } + return InetAddress.getByName(s); + } + + /** Codec of a stream with payload type {@code pt}: Majestic sends 96 = H.264, 97 = H.265. */ + boolean h265(int pt) { + if (codec != null) return codec.equals("H265"); + return pt == Sdp.PT_H265; + } + } + + private final String tag; + private final Url url; + private final MulticastSocket socket; + private final RtpReorder reorder = + new RtpReorder(64, UdpReceiver.REORDER_MS); + private volatile boolean closed; + private boolean joined; + private long rejected; + + RtpListener(String tag, Url url) throws IOException { + this.tag = tag; + this.url = url; + MulticastSocket s = new MulticastSocket(null); + try { + if (url.multicast()) { + // Several receivers of one group on this device each get a copy + s.setReuseAddress(true); + s.bind(new InetSocketAddress(url.port)); + s.joinGroup(url.address); + joined = true; + } else { + // Unicast: a second slot on the same port fails here instead of starving + s.setReuseAddress(false); + s.bind(url.address != null + ? new InetSocketAddress(url.address, url.port) + : new InetSocketAddress(url.port)); + } + s.setReceiveBufferSize(RCVBUF); + s.setSoTimeout(250); + } catch (IOException e) { + s.close(); + throw e; + } + this.socket = s; + if (joined) MulticastLock.acquire(); + } + + /** Receive until {@code keepGoing} turns false or {@link #close} is called. */ + void run(UdpReceiver.Condition keepGoing, Handler handler) throws IOException { + byte[] buf = new byte[65536]; + DatagramPacket p = new DatagramPacket(buf, buf.length); + RtpReorder.Sink sink = handler::onPacket; + InetAddress source = url.source; + try { + while (!closed && keepGoing.holds()) { + long ms = reorder.msUntilDeadline(System.nanoTime()); + socket.setSoTimeout(ms < 0 ? 250 : (int) Math.max(1, Math.min(250, ms))); + try { + p.setLength(buf.length); + socket.receive(p); + if (source != null && !source.equals(p.getAddress())) { + if (rejected++ % 100 == 0) { + Log.w(TAG, tag + ": UDP from an unexpected sender ignored (" + + rejected + ")"); + } + continue; + } + reorder.push(buf, p.getOffset(), p.getLength(), System.nanoTime(), sink); + } catch (SocketTimeoutException ignored) { + // Check the condition and the reorder deadline + } + reorder.poll(System.nanoTime(), sink); + } + } catch (IOException e) { + if (!closed) throw e; + } finally { + if (reorder.dropped() > 0 || reorder.skipped() > 0) { + Log.i(TAG, tag + ": UDP " + reorder.skipped() + " packets lost, " + + reorder.dropped() + " late or duplicate"); + } + close(); + } + } + + /** From any thread; a blocked receive ends with the socket. */ + @Override + public synchronized void close() { + if (closed) return; + closed = true; + if (joined) { + try { + socket.leaveGroup(url.address); + } catch (IOException ignored) {} + MulticastLock.release(); + } + socket.close(); + } + + /** + * Wi-Fi drivers drop multicast unless an app holds a MulticastLock. Counted, so it is + * held while any slot receives a group; a no-op without an app context (unit tests). + */ + static final class MulticastLock { + private static WifiManager.MulticastLock lock; + private static int users; + + static synchronized void acquire() { + if (users++ > 0) return; + Context ctx = App.context; + if (ctx == null) return; + WifiManager wifi = (WifiManager) ctx.getSystemService(Context.WIFI_SERVICE); + if (wifi == null) return; + try { + lock = wifi.createMulticastLock("OpenIPCDecoder"); + lock.setReferenceCounted(false); + lock.acquire(); + } catch (RuntimeException e) { + Log.w(TAG, "No Wi-Fi multicast lock: " + e.getMessage()); + lock = null; + } + } + + static synchronized void release() { + if (users == 0 || --users > 0) return; + if (lock != null) { + lock.release(); + lock = null; + } + } + } +} diff --git a/android/app/src/main/java/com/openipc/decoder/RtspSession.java b/android/app/src/main/java/com/openipc/decoder/RtspSession.java index a4477320..5ad349f0 100644 --- a/android/app/src/main/java/com/openipc/decoder/RtspSession.java +++ b/android/app/src/main/java/com/openipc/decoder/RtspSession.java @@ -30,7 +30,8 @@ * Handshake (DESCRIBE, SETUP, PLAY), SDP parsing, the receive loop (RTP interleaved in the * RTSP connection, or over UDP, #67), a watchdog and reconnection with capped back-off. Used * by the single view (RtspClient) and by every Quad cell. Packets are delivered on the - * network thread; channel 0 is video, channel 2 audio. + * network thread; channel 0 is video, channel 2 audio. A udp:// or rtp:// URL is not RTSP: + * the camera pushes plain RTP to us (Majestic "outgoing", #68), see {@link RtpListener}. */ class RtspSession { private static final String TAG = "OpenIPCDecoder"; @@ -85,6 +86,8 @@ interface Callback { private UdpReceiver udp; /** The UDP receive loop is running: closing ends it, the network thread closes the rest. */ private boolean udpPlaying; + /** Listener of a udp:// slot, else null. */ + private RtpListener listener; private Thread networkThread; private Thread watchdogThread; @@ -157,6 +160,7 @@ void stop() { * network thread then sends TEARDOWN, so the server stops sending to our ports. */ private void closeSocketLocked() { + if (listener != null) listener.close(); if (udp != null) { udp.close(); if (udpPlaying) return; @@ -245,6 +249,7 @@ private void watchdogLoop(int gen) { /** One session from connect to disconnect. Returns true if PLAY succeeded. */ private boolean connect(int gen) throws IOException { + if (RtpListener.Url.matches(url)) return listen(gen, RtpListener.Url.parse(url)); Url u = Url.parse(url); Socket s = new Socket(); synchronized (lock) { @@ -401,6 +406,66 @@ private boolean connect(int gen) throws IOException { } } + /** + * Receive RTP pushed to us (#68). The codec follows the payload type unless the URL + * forces one: 96 is H.264 and 97 H.265, and a switch between them on the camera is + * followed. Returns true once listening, like a PLAY that succeeded. + */ + private boolean listen(int gen, RtpListener.Url u) throws IOException { + RtpListener l = new RtpListener(tag, u); + synchronized (lock) { + if (gen != generation) { + l.close(); + return false; + } + listener = l; + lastPacketMs = System.nanoTime() / 1_000_000; + streamBytes = 0; + bitrateMarkNs = System.nanoTime(); + active = true; + } + try { + Log.i(TAG, tag + ": listening for RTP on UDP port " + u.port + + (u.multicast() ? ", group " + u.address.getHostAddress() : "")); + status(gen, "buffering"); + final Sdp[] current = new Sdp[1]; + final boolean[] first = {true}; + l.run(() -> active && gen == generation, (buf, offset, length) -> { + if (gen != generation || length < 12) return; + streamBytes += length; + int pt = buf[offset + 1] & 0x7F; + Sdp sdp = current[0]; + boolean video = u.pt >= 0 ? pt == u.pt + : u.codec != null ? sdp == null || pt == sdp.videoPt + : pt == Sdp.PT_H264 || pt == Sdp.PT_H265; + if (video && (sdp == null || sdp.videoPt != pt)) { + sdp = new Sdp(); + sdp.videoPt = pt; + sdp.videoH265 = u.h265(pt); + current[0] = sdp; + Log.i(TAG, tag + ": RTP payload type " + pt + + (sdp.videoH265 ? ", H.265" : ", H.264")); + callback.onSdp(sdp); + } + if (sdp == null) return; // nothing to decode yet + boolean usable = callback.onPacket(CHANNEL_VIDEO, buf, offset, length); + if (usable) { + lastPacketMs = System.nanoTime() / 1_000_000; + if (first[0]) { + first[0] = false; + status(gen, "connected"); + } + } + }); + return true; + } finally { + synchronized (lock) { + if (listener == l) listener = null; + } + l.close(); + } + } + /** * A UDP reply we can receive: unicast, and the client ports, when given, are ours. * A missing Transport header is taken as agreement. diff --git a/android/app/src/test/java/com/openipc/decoder/RtpListenerTest.java b/android/app/src/test/java/com/openipc/decoder/RtpListenerTest.java new file mode 100644 index 00000000..44e4a39c --- /dev/null +++ b/android/app/src/test/java/com/openipc/decoder/RtpListenerTest.java @@ -0,0 +1,238 @@ +/* + * Copyright (c) OpenIPC https://openipc.org Prosperity Public License 3.0.0 + * + * RtpListenerTest.java — udp:// slots: URL parsing and plain RTP pushed by the camera + * + */ + +package com.openipc.decoder; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import java.io.IOException; +import java.net.DatagramPacket; +import java.net.DatagramSocket; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +public class RtpListenerTest { + // ====================== URL ====================== + + @Test + public void listenOnAPort() throws Exception { + RtpListener.Url u = RtpListener.Url.parse("udp://@:5600"); + assertNull(u.address); + assertEquals(5600, u.port); + assertFalse(u.multicast()); + assertNull(u.source); + assertNull(u.codec); + assertEquals(-1, u.pt); + // Without the "@" too, and with rtp:// + assertEquals(5600, RtpListener.Url.parse("udp://:5600").port); + assertEquals(5000, RtpListener.Url.parse("rtp://@:5000/").port); + } + + @Test + public void multicastGroupAndSource() throws Exception { + RtpListener.Url u = RtpListener.Url.parse("udp://192.168.1.10@239.1.2.3:5600"); + assertTrue(u.multicast()); + assertEquals("239.1.2.3", u.address.getHostAddress()); + assertEquals("192.168.1.10", u.source.getHostAddress()); + + u = RtpListener.Url.parse("udp://@[ff02::1234]:5600"); + assertTrue(u.multicast()); + assertEquals(5600, u.port); + } + + @Test + public void localAddressIsNotMulticast() throws Exception { + RtpListener.Url u = RtpListener.Url.parse("udp://@127.0.0.1:5600"); + assertFalse(u.multicast()); + assertTrue(u.address.isLoopbackAddress()); + } + + @Test + public void codecAndPayloadType() throws Exception { + RtpListener.Url u = RtpListener.Url.parse("udp://@:5600?codec=hevc&pt=100"); + assertEquals("H265", u.codec); + assertEquals(100, u.pt); + assertTrue(u.h265(100)); + // By payload type, as Majestic sends them + RtpListener.Url plain = RtpListener.Url.parse("udp://@:5600"); + assertTrue(plain.h265(97)); + assertFalse(plain.h265(96)); + } + + @Test + public void badUrlsAreRefused() { + String[] bad = { + "udp://@239.1.2.3", "udp://@:0", "udp://@:70000", "udp://@:abc", + "udp://@camera.local:5600", "udp://@:5600?codec=mjpeg", "udp://@:5600?pt=200", + "rtsp://cam/stream", "udp://@[ff02::1:5600", + }; + for (String url : bad) { + try { + RtpListener.Url.parse(url); + fail("accepted " + url); + } catch (IOException expected) { + } + } + assertTrue(RtpListener.Url.matches(" UDP://@:5600")); + assertFalse(RtpListener.Url.matches("rtsp://cam")); + assertFalse(RtpListener.Url.matches(null)); + } + + // ====================== Session ====================== + + private static final class Recorder implements RtspSession.Callback { + final List statuses = Collections.synchronizedList(new ArrayList<>()); + final List sdps = Collections.synchronizedList(new ArrayList<>()); + final List pts = Collections.synchronizedList(new ArrayList<>()); + volatile CountDownLatch latch = new CountDownLatch(3); + + @Override public void onStatus(String status) { statuses.add(status); } + @Override public void onSdp(Sdp sdp) { sdps.add(sdp); } + @Override public boolean onPacket(int channel, byte[] buf, int offset, int length) { + assertEquals(RtspSession.CHANNEL_VIDEO, channel); + pts.add(buf[offset + 1] & 0x7F); + latch.countDown(); + return true; + } + } + + private final Recorder rec = new Recorder(); + private RtspSession session; + private DatagramSocket sender; + private int seq = 1000; + + @Before + public void setUp() throws IOException { + RtspSession.minRetryMs = 5; + RtspSession.maxRetryMs = 20; + session = new RtspSession("test", rec); + sender = new DatagramSocket(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0)); + } + + @After + public void tearDown() { + session.stop(); + sender.close(); + RtspSession.minRetryMs = 1000; + RtspSession.maxRetryMs = 30000; + } + + private static int freePort() throws IOException { + try (DatagramSocket s = new DatagramSocket(0)) { + return s.getLocalPort(); + } + } + + private void send(int port, int pt, int count) throws IOException { + for (int i = 0; i < count; i++) { + byte[] p = RtspSessionTest.udpPacket(pt, seq++); + sender.send(new DatagramPacket(p, p.length, InetAddress.getLoopbackAddress(), port)); + } + } + + /** Start listening and wait until the port is bound ("buffering"). */ + private void listen(String url) throws InterruptedException { + session.configure(url, "", true, Transport.Mode.TCP); + session.start(); + long deadline = System.currentTimeMillis() + 3000; + while (!rec.statuses.contains("buffering") && System.currentTimeMillis() < deadline) { + Thread.sleep(5); + } + assertTrue("listening", rec.statuses.contains("buffering")); + } + + @Test + public void receivesH264AndFollowsASwitchToH265() throws Exception { + int port = freePort(); + listen("udp://@:" + port); + send(port, 96, 3); + assertTrue(rec.latch.await(3, TimeUnit.SECONDS)); + assertEquals(1, rec.sdps.size()); + assertEquals(96, rec.sdps.get(0).videoPt); + assertFalse(rec.sdps.get(0).videoH265); + assertTrue(rec.statuses.contains("connected")); + + // The camera switched to H.265: a new session description, then the packets + rec.latch = new CountDownLatch(2); + send(port, 97, 2); + assertTrue(rec.latch.await(3, TimeUnit.SECONDS)); + assertEquals(2, rec.sdps.size()); + assertTrue(rec.sdps.get(1).videoH265); + } + + @Test + public void otherPayloadTypesAreNotVideo() throws Exception { + int port = freePort(); + listen("udp://@:" + port); + send(port, 98, 3); // audio or anything else: no codec, nothing delivered + send(port, 96, 3); + assertTrue(rec.latch.await(3, TimeUnit.SECONDS)); + assertEquals(1, rec.sdps.size()); + for (int pt : rec.pts) assertEquals(96, pt); + } + + @Test + public void forcedCodecAndPayloadType() throws Exception { + int port = freePort(); + listen("udp://@:" + port + "?codec=h265&pt=110"); + send(port, 96, 2); // not the forced type: ignored + send(port, 110, 3); + assertTrue(rec.latch.await(3, TimeUnit.SECONDS)); + assertEquals(1, rec.sdps.size()); + assertEquals(110, rec.sdps.get(0).videoPt); + assertTrue(rec.sdps.get(0).videoH265); + } + + @Test + public void wrongSourceIsIgnored() throws Exception { + int port = freePort(); + listen("udp://192.0.2.1@:" + port); + send(port, 96, 3); + assertFalse(rec.latch.await(500, TimeUnit.MILLISECONDS)); + assertTrue(rec.sdps.isEmpty()); + } + + @Test + public void portInUseIsRetriedNotShared() throws Exception { + try (DatagramSocket busy = new DatagramSocket(0)) { + int port = busy.getLocalPort(); + session.configure("udp://@:" + port, "", true, Transport.Mode.TCP); + session.start(); + Thread.sleep(300); + assertFalse(rec.statuses.contains("buffering")); + } + } + + @Test + public void stopReleasesThePort() throws Exception { + int port = freePort(); + listen("udp://@:" + port); + session.stop(); + long deadline = System.currentTimeMillis() + 2000; + while (System.currentTimeMillis() < deadline) { + try (DatagramSocket again = new DatagramSocket(port)) { + return; // bound: released + } catch (IOException stillBound) { + Thread.sleep(20); + } + } + fail("port still in use after stop()"); + } +} From f2ac215371c206a70b90bcea7ceb882c765f34f6 Mon Sep 17 00:00:00 2001 From: Compot Developer Date: Mon, 12 Oct 2026 19:21:39 +0000 Subject: [PATCH 2/4] fix: plain RTP review findings (#68) - One reorder buffer per payload type: audio on the same port, with its own sequence numbers, no longer pushes video packets out as late ones. - Video is payload type 96 or 97 unless pt= names another; codec= alone only overrides the codec, so audio arriving first is not taken for video. - A quiet sender is reported (disconnected, then connected again with the next packet) and the port stays bound: a rebind only lost packets and, being counted as a played session, never backed off. - stop() and a starting listener clear the keepalive of an older RTSP connection, so the watchdog cannot send on it and close the listener. - android.hardware.wifi is not required: the multicast permission must not hide the app on Ethernet-only TV boxes (lint LeanbackUsesWifi). --- android/app/src/main/AndroidManifest.xml | 2 + .../java/com/openipc/decoder/RtpListener.java | 41 ++++++++---- .../java/com/openipc/decoder/RtspSession.java | 30 +++++++-- .../com/openipc/decoder/RtpListenerTest.java | 63 +++++++++++++++++-- 4 files changed, 115 insertions(+), 21 deletions(-) diff --git a/android/app/src/main/AndroidManifest.xml b/android/app/src/main/AndroidManifest.xml index b8703dc1..33e1180b 100755 --- a/android/app/src/main/AndroidManifest.xml +++ b/android/app/src/main/AndroidManifest.xml @@ -11,6 +11,8 @@ + + = 0) ms = Math.min(ms, d); + } + socket.setSoTimeout((int) Math.max(1, ms)); try { p.setLength(buf.length); socket.receive(p); @@ -200,18 +209,28 @@ void run(UdpReceiver.Condition keepGoing, Handler handler) throws IOException { } continue; } - reorder.push(buf, p.getOffset(), p.getLength(), System.nanoTime(), sink); + if (p.getLength() < 12) continue; + int pt = buf[p.getOffset() + 1] & 0x7F; + RtpReorder r = reorders[pt]; + if (r == null) r = reorders[pt] = new RtpReorder(64, UdpReceiver.REORDER_MS); + r.push(buf, p.getOffset(), p.getLength(), System.nanoTime(), sink); } catch (SocketTimeoutException ignored) { - // Check the condition and the reorder deadline + // Check the condition and the reorder deadlines + } + now = System.nanoTime(); + for (RtpReorder r : reorders) { + if (r != null) r.poll(now, sink); } - reorder.poll(System.nanoTime(), sink); } } catch (IOException e) { if (!closed) throw e; } finally { - if (reorder.dropped() > 0 || reorder.skipped() > 0) { - Log.i(TAG, tag + ": UDP " + reorder.skipped() + " packets lost, " - + reorder.dropped() + " late or duplicate"); + for (int pt = 0; pt < reorders.length; pt++) { + RtpReorder r = reorders[pt]; + if (r != null && (r.dropped() > 0 || r.skipped() > 0)) { + Log.i(TAG, tag + ": UDP pt " + pt + ": " + r.skipped() + " packets lost, " + + r.dropped() + " late or duplicate"); + } } close(); } diff --git a/android/app/src/main/java/com/openipc/decoder/RtspSession.java b/android/app/src/main/java/com/openipc/decoder/RtspSession.java index 5ad349f0..abc6457d 100644 --- a/android/app/src/main/java/com/openipc/decoder/RtspSession.java +++ b/android/app/src/main/java/com/openipc/decoder/RtspSession.java @@ -88,6 +88,8 @@ interface Callback { private boolean udpPlaying; /** Listener of a udp:// slot, else null. */ private RtpListener listener; + /** A udp:// slot is listening: silence is reported, never "fixed" by a rebind. */ + private volatile boolean listening; private Thread networkThread; private Thread watchdogThread; @@ -147,6 +149,7 @@ void stop() { synchronized (lock) { generation++; active = false; + keepalive = null; closeSocketLocked(); if (networkThread != null) networkThread.interrupt(); if (watchdogThread != null) watchdogThread.interrupt(); @@ -232,7 +235,8 @@ private void watchdogLoop(int gen) { } } } - if (active && System.nanoTime() / 1_000_000 - lastPacketMs > watchdogMs) { + if (active && !listening + && System.nanoTime() / 1_000_000 - lastPacketMs > watchdogMs) { Log.w(TAG, tag + ": stream is inactive, reconnecting"); active = false; synchronized (lock) { @@ -419,6 +423,9 @@ private boolean listen(int gen, RtpListener.Url u) throws IOException { return false; } listener = l; + listening = true; + // No RTSP here: nothing of an older connection may be sent by the watchdog + keepalive = null; lastPacketMs = System.nanoTime() / 1_000_000; streamBytes = 0; bitrateMarkNs = System.nanoTime(); @@ -430,14 +437,22 @@ private boolean listen(int gen, RtpListener.Url u) throws IOException { status(gen, "buffering"); final Sdp[] current = new Sdp[1]; final boolean[] first = {true}; - l.run(() -> active && gen == generation, (buf, offset, length) -> { + l.run(() -> { + if (!active || gen != generation) return false; + // The sender went quiet: say so, keep listening (a rebind would only lose + // packets), and report "connected" again with the next usable one + if (!first[0] && System.nanoTime() / 1_000_000 - lastPacketMs > watchdogMs) { + first[0] = true; + status(gen, "disconnected"); + } + return true; + }, (buf, offset, length) -> { if (gen != generation || length < 12) return; streamBytes += length; int pt = buf[offset + 1] & 0x7F; Sdp sdp = current[0]; - boolean video = u.pt >= 0 ? pt == u.pt - : u.codec != null ? sdp == null || pt == sdp.videoPt - : pt == Sdp.PT_H264 || pt == Sdp.PT_H265; + // Video is the forced type, or 96/97; anything else (audio) is ignored + boolean video = u.pt >= 0 ? pt == u.pt : pt == Sdp.PT_H264 || pt == Sdp.PT_H265; if (video && (sdp == null || sdp.videoPt != pt)) { sdp = new Sdp(); sdp.videoPt = pt; @@ -460,7 +475,10 @@ private boolean listen(int gen, RtpListener.Url u) throws IOException { return true; } finally { synchronized (lock) { - if (listener == l) listener = null; + if (listener == l) { + listener = null; + listening = false; + } } l.close(); } diff --git a/android/app/src/test/java/com/openipc/decoder/RtpListenerTest.java b/android/app/src/test/java/com/openipc/decoder/RtpListenerTest.java index 44e4a39c..c0f8b53b 100644 --- a/android/app/src/test/java/com/openipc/decoder/RtpListenerTest.java +++ b/android/app/src/test/java/com/openipc/decoder/RtpListenerTest.java @@ -132,6 +132,7 @@ public void tearDown() { sender.close(); RtspSession.minRetryMs = 1000; RtspSession.maxRetryMs = 30000; + RtspSession.watchdogMs = 8000; } private static int freePort() throws IOException { @@ -141,10 +142,12 @@ private static int freePort() throws IOException { } private void send(int port, int pt, int count) throws IOException { - for (int i = 0; i < count; i++) { - byte[] p = RtspSessionTest.udpPacket(pt, seq++); - sender.send(new DatagramPacket(p, p.length, InetAddress.getLoopbackAddress(), port)); - } + for (int i = 0; i < count; i++) sendSeq(port, pt, seq++); + } + + private void sendSeq(int port, int pt, int s) throws IOException { + byte[] p = RtspSessionTest.udpPacket(pt, s); + sender.send(new DatagramPacket(p, p.length, InetAddress.getLoopbackAddress(), port)); } /** Start listening and wait until the port is bound ("buffering"). */ @@ -235,4 +238,56 @@ public void stopReleasesThePort() throws Exception { } fail("port still in use after stop()"); } + + @Test + public void audioOnTheSamePortDoesNotPushVideoOut() throws Exception { + int port = freePort(); + rec.latch = new CountDownLatch(6); + listen("udp://@:" + port); + // Audio with its own, higher sequence numbers between the video packets + for (int i = 0; i < 6; i++) { + sendSeq(port, 96, 100 + i); + sendSeq(port, 98, 40000 + i); + } + assertTrue(rec.latch.await(3, TimeUnit.SECONDS)); + Thread.sleep(100); + assertEquals(6, rec.pts.size()); + } + + @Test + public void codecWithoutPtStillTakesOnly96Or97() throws Exception { + int port = freePort(); + listen("udp://@:" + port + "?codec=h265"); + send(port, 98, 2); // audio first: not video + send(port, 96, 3); + assertTrue(rec.latch.await(3, TimeUnit.SECONDS)); + assertEquals(1, rec.sdps.size()); + assertEquals(96, rec.sdps.get(0).videoPt); + assertTrue("codec forced by the URL", rec.sdps.get(0).videoH265); + } + + @Test + public void silenceIsReportedWithoutRebinding() throws Exception { + RtspSession.watchdogMs = 200; + int port = freePort(); + listen("udp://@:" + port); + send(port, 96, 3); + assertTrue(rec.latch.await(3, TimeUnit.SECONDS)); + long deadline = System.currentTimeMillis() + 3000; + while (!rec.statuses.contains("disconnected") && System.currentTimeMillis() < deadline) { + Thread.sleep(10); + } + assertTrue(rec.statuses.contains("disconnected")); + // Still bound to the port: the next packet plays at once + rec.latch = new CountDownLatch(1); + send(port, 96, 1); + assertTrue(rec.latch.await(3, TimeUnit.SECONDS)); + assertEquals(1, Collections.frequency(rec.statuses, "buffering")); + deadline = System.currentTimeMillis() + 1000; + while (Collections.frequency(rec.statuses, "connected") < 2 + && System.currentTimeMillis() < deadline) { + Thread.sleep(10); + } + assertEquals(2, Collections.frequency(rec.statuses, "connected")); + } } From 5da838c3ac55c0e22196a69f70ea6c228ab9dbfe Mon Sep 17 00:00:00 2001 From: Compot Developer Date: Mon, 12 Oct 2026 19:23:52 +0000 Subject: [PATCH 3/4] fix: plain RTP passes only video packets on (#68) --- android/app/src/main/java/com/openipc/decoder/RtspSession.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/android/app/src/main/java/com/openipc/decoder/RtspSession.java b/android/app/src/main/java/com/openipc/decoder/RtspSession.java index abc6457d..552f1658 100644 --- a/android/app/src/main/java/com/openipc/decoder/RtspSession.java +++ b/android/app/src/main/java/com/openipc/decoder/RtspSession.java @@ -462,7 +462,7 @@ private boolean listen(int gen, RtpListener.Url u) throws IOException { + (sdp.videoH265 ? ", H.265" : ", H.264")); callback.onSdp(sdp); } - if (sdp == null) return; // nothing to decode yet + if (sdp == null || pt != sdp.videoPt) return; // not video, or none yet boolean usable = callback.onPacket(CHANNEL_VIDEO, buf, offset, length); if (usable) { lastPacketMs = System.nanoTime() / 1_000_000; From a00d6f2f5f001916fb37dcedb7f3b97f94e8df5c Mon Sep 17 00:00:00 2001 From: Compot Developer Date: Mon, 12 Oct 2026 19:36:18 +0000 Subject: [PATCH 4/4] fix: a udp:// multicast slot binds to its group, not the wildcard (#68) On Linux a socket bound to the wildcard address receives every group joined on that port on the device, so two slots on different groups with the same port would mix their streams (found by the review of #73). --- .../app/src/main/java/com/openipc/decoder/RtpListener.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/android/app/src/main/java/com/openipc/decoder/RtpListener.java b/android/app/src/main/java/com/openipc/decoder/RtpListener.java index 0a0f095e..e92ed1ff 100644 --- a/android/app/src/main/java/com/openipc/decoder/RtpListener.java +++ b/android/app/src/main/java/com/openipc/decoder/RtpListener.java @@ -162,9 +162,11 @@ boolean h265(int pt) { MulticastSocket s = new MulticastSocket(null); try { if (url.multicast()) { - // Several receivers of one group on this device each get a copy + // Several receivers of one group on this device each get a copy. Bound to + // the group, not the wildcard: on Linux a wildcard socket also receives every + // other group joined on this port, e.g. another channel of the same NVR s.setReuseAddress(true); - s.bind(new InetSocketAddress(url.port)); + s.bind(new InetSocketAddress(url.address, url.port)); s.joinGroup(url.address); joined = true; } else {