From 9723901ef089773a0097eb3fb37e4c045b65c79a Mon Sep 17 00:00:00 2001 From: Compot Developer Date: Mon, 12 Oct 2026 18:52:04 +0000 Subject: [PATCH 1/3] feat: RTSP over UDP unicast, with fallback to TCP (#67) Every camera slot gets a transport setting: TCP (as before, the default), UDP, or Auto (UDP, falling back to TCP). It is chosen in the URL dialog and can be passed with a shared link (--es transport udp|tcp|auto). - UdpReceiver binds an even RTP port and the RTCP port above it per track, always ephemeral (fixed ports were the bug of #40), and receives every track on the network thread through one selector, so callbacks keep coming from a single thread as over TCP. Only datagrams from the camera (the RTSP peer or the reply's "source") are accepted. - RtpReorder puts packets back in sequence order. In-order packets pass straight through; a gap is waited for at most 40 ms or 64 packets, then skipped; late duplicates are dropped. - Transport parses the SETUP reply, which decides what is received: Majestic answers "RTP/AVP/UDP;...", XM cameras "RTP/AVP;...;source=...; ssrc=...", and a server may answer with interleaved TCP. - Over UDP a second thread reads the RTSP replies (keepalives). Closing a UDP session sends TEARDOWN, since the server otherwise keeps sending until the session times out, and XM cameras allow one client only. - Auto falls back to TCP on a refused UDP SETUP (in the same connection) or when no video datagram arrives within 3 s after PLAY, and stays on TCP until the slot is reconfigured. --- .../java/com/openipc/decoder/Decoder.java | 18 +- .../java/com/openipc/decoder/MenuManager.java | 28 ++- .../java/com/openipc/decoder/QuadCell.java | 7 +- .../java/com/openipc/decoder/RtpReorder.java | 150 ++++++++++++ .../java/com/openipc/decoder/RtspClient.java | 4 +- .../java/com/openipc/decoder/RtspSession.java | 200 ++++++++++++++-- .../java/com/openipc/decoder/Transport.java | 111 +++++++++ .../java/com/openipc/decoder/UdpReceiver.java | 215 ++++++++++++++++++ .../com/openipc/decoder/QuadAudioTest.java | 4 +- .../com/openipc/decoder/RtpReorderTest.java | 115 ++++++++++ .../com/openipc/decoder/RtspSessionTest.java | 169 +++++++++++++- .../com/openipc/decoder/TransportTest.java | 99 ++++++++ 12 files changed, 1082 insertions(+), 38 deletions(-) create mode 100644 android/app/src/main/java/com/openipc/decoder/RtpReorder.java create mode 100644 android/app/src/main/java/com/openipc/decoder/Transport.java create mode 100644 android/app/src/main/java/com/openipc/decoder/UdpReceiver.java create mode 100644 android/app/src/test/java/com/openipc/decoder/RtpReorderTest.java create mode 100644 android/app/src/test/java/com/openipc/decoder/TransportTest.java diff --git a/android/app/src/main/java/com/openipc/decoder/Decoder.java b/android/app/src/main/java/com/openipc/decoder/Decoder.java index 5b59f401..46e3d7d2 100755 --- a/android/app/src/main/java/com/openipc/decoder/Decoder.java +++ b/android/app/src/main/java/com/openipc/decoder/Decoder.java @@ -100,6 +100,8 @@ public class Decoder extends Activity static final int CAM_COUNT = 8; private static final String DEFAULT_URL = "rtsp://root:12345@192.168.1.10:554/stream=0"; private final String[] mHosts = new String[CAM_COUNT]; + /** RTP transport per slot (#67): TCP, UDP or Auto (UDP with fallback to TCP). */ + private final Transport.Mode[] mTransports = new Transport.Mode[CAM_COUNT]; private int mActive; private volatile String mHost; private String mVersion; @@ -299,7 +301,7 @@ public void onToggleQuad() { stopQuad(); boolean configured = mHost != null && !mHost.isEmpty() && !mHost.equals(DEFAULT_URL); if (configured) { - rtspClient.configure(mHost, mUserAgent); + rtspClient.configure(mHost, mUserAgent, mTransports[mActive]); startListener(); rtspClient.start(); } @@ -350,7 +352,7 @@ private boolean leaveUdpView() { public void onExit() { finishAndRemoveTask(); } @Override - public void onShowSettings() { menuManager.showUrlEditor(mActive, mHosts); } + public void onShowSettings() { menuManager.showUrlEditor(mActive, mHosts, mTransports); } @Override public void onWebUI() { menuManager.startBrowser(mHost); } @@ -395,6 +397,7 @@ private void loadSettings() { String url = credentials.open(stored); mHosts[i] = url != null ? url : DEFAULT_URL; mUnreadable[i] = url == null && stored != null ? stored : null; + mTransports[i] = Transport.Mode.parse(pref.getString("transport_" + i, null)); // Plain URLs from older versions are re-saved encrypted right away if (credentials.needsMigration(stored)) { if (migrate == null) migrate = pref.edit(); @@ -408,6 +411,9 @@ private void loadSettings() { if (link != null) { Log.d(TAG, "Link: " + Uri.parse(link).getHost()); mHosts[mActive] = sanitizeUrl(link); + // Optional, e.g. --es transport udp; a link without it keeps the slot's setting + String transport = intent.getStringExtra("transport"); + if (transport != null) mTransports[mActive] = Transport.Mode.parse(transport); // A shared link is meant to be watched: single view, and saved so that it is // still there after the next pause/resume quadEnabled = false; @@ -415,6 +421,7 @@ private void loadSettings() { .putString("host_" + mActive, credentials.sealForSlot( mHosts[mActive], mUnreadable[mActive], DEFAULT_URL)) .putBoolean("quad_enabled", false) + .putString("transport_" + mActive, mTransports[mActive].name()) .apply(); intent.removeExtra(Intent.EXTRA_TEXT); } @@ -446,6 +453,7 @@ private void saveSettings() { for (int i = 0; i < CAM_COUNT; i++) { edit.putString("host_" + i, credentials.sealForSlot(mHosts[i], mUnreadable[i], DEFAULT_URL)); + edit.putString("transport_" + i, mTransports[i].name()); } edit.apply(); @@ -466,7 +474,7 @@ private void saveSettings() { return; } - rtspClient.configure(mHost, mUserAgent); + rtspClient.configure(mHost, mUserAgent, mTransports[mActive]); startListener(); rtspClient.start(); } @@ -709,7 +717,7 @@ private void createQuadCells() { if (quadViews[i] == null) continue; String url = mHosts[i]; if (url == null || url.isEmpty() || url.equals(DEFAULT_URL)) continue; - quadCells[i] = new QuadCell(i, url, quadViews[i], framePool, mUserAgent, + quadCells[i] = new QuadCell(i, url, mTransports[i], quadViews[i], framePool, mUserAgent, index -> runOnUiThread(() -> removeQuadOverlay(index))); quadCells[i].start(); } @@ -1140,7 +1148,7 @@ private void startStreams() { mSurface.setVisibility(View.VISIBLE); boolean configured = mHost != null && !mHost.isEmpty() && !mHost.equals(DEFAULT_URL); if (configured) { - rtspClient.configure(mHost, mUserAgent); + rtspClient.configure(mHost, mUserAgent, mTransports[mActive]); startListener(); rtspClient.start(); } diff --git a/android/app/src/main/java/com/openipc/decoder/MenuManager.java b/android/app/src/main/java/com/openipc/decoder/MenuManager.java index bada8f66..e06f9a93 100644 --- a/android/app/src/main/java/com/openipc/decoder/MenuManager.java +++ b/android/app/src/main/java/com/openipc/decoder/MenuManager.java @@ -32,6 +32,8 @@ import android.widget.HorizontalScrollView; import android.widget.LinearLayout; import android.widget.PopupWindow; +import android.widget.RadioButton; +import android.widget.RadioGroup; import android.widget.TextView; class MenuManager { @@ -342,7 +344,7 @@ private EditText createEdit(String title) { return text; } - void showUrlEditor(int mActive, String[] mHosts) { + void showUrlEditor(int mActive, String[] mHosts, Transport.Mode[] mTransports) { AlertDialog.Builder builder = new AlertDialog.Builder(activity); builder.setTitle("Camera " + (mActive + 1) + " URL"); @@ -351,12 +353,34 @@ void showUrlEditor(int mActive, String[] mHosts) { input.setInputType(InputType.TYPE_CLASS_TEXT | InputType.TYPE_TEXT_VARIATION_URI); input.setSelectAllOnFocus(true); input.setSelection(input.getText().length()); - builder.setView(input); + + // RTP transport of the slot (#67): Auto tries UDP and falls back to TCP + RadioGroup transport = new RadioGroup(activity); + transport.setOrientation(RadioGroup.HORIZONTAL); + Transport.Mode[] modes = Transport.Mode.values(); + String[] labels = {"TCP", "UDP", "Auto"}; + for (int i = 0; i < modes.length; i++) { + RadioButton b = new RadioButton(activity); + b.setId(View.generateViewId()); + b.setText(labels[i]); + b.setTag(modes[i]); + transport.addView(b); + if (modes[i] == mTransports[mActive]) b.setChecked(true); + } + + LinearLayout content = new LinearLayout(activity); + content.setOrientation(LinearLayout.VERTICAL); + content.setPadding(dp(16), dp(8), dp(16), 0); + content.addView(input); + content.addView(transport); + builder.setView(content); final int slot = mActive; builder.setPositiveButton("Save", (dialog, which) -> { String url = input.getText().toString().trim(); mHosts[slot] = Decoder.sanitizeUrl(url); + View checked = transport.findViewById(transport.getCheckedRadioButtonId()); + if (checked != null) mTransports[slot] = (Transport.Mode) checked.getTag(); if (listener != null) listener.onSaveSettings(); // triggers save + reconnect }); builder.setNegativeButton("Cancel", (dialog, which) -> dialog.cancel()); diff --git a/android/app/src/main/java/com/openipc/decoder/QuadCell.java b/android/app/src/main/java/com/openipc/decoder/QuadCell.java index e2d4d28d..45cecc86 100644 --- a/android/app/src/main/java/com/openipc/decoder/QuadCell.java +++ b/android/app/src/main/java/com/openipc/decoder/QuadCell.java @@ -23,6 +23,7 @@ class QuadCell implements RtspSession.Callback { private final int index; private final String host; + private final Transport.Mode transport; private final SurfaceView view; private final String tag; private final FramePool cellFramePool; @@ -66,9 +67,11 @@ void setAudioSink(AudioManager sink, int route) { private final Listener listener; - QuadCell(int index, String host, SurfaceView view, FramePool pool, String userAgent, Listener listener) { + QuadCell(int index, String host, Transport.Mode transport, SurfaceView view, FramePool pool, + String userAgent, Listener listener) { this.index = index; this.host = host; + this.transport = transport; this.view = view; this.tag = "Quad-" + index; this.cellFramePool = pool; @@ -160,7 +163,7 @@ private void startThreads() { Log.i(TAG, tag + ": surface ready, starting session"); // Audio is set up too, so switching the audible cell needs no reconnect (#12) - session.configure(host, userAgent, true); + session.configure(host, userAgent, true, transport); session.start(); } diff --git a/android/app/src/main/java/com/openipc/decoder/RtpReorder.java b/android/app/src/main/java/com/openipc/decoder/RtpReorder.java new file mode 100644 index 00000000..6397a5de --- /dev/null +++ b/android/app/src/main/java/com/openipc/decoder/RtpReorder.java @@ -0,0 +1,150 @@ +/* + * Copyright (c) OpenIPC https://openipc.org Prosperity Public License 3.0.0 + * + * RtpReorder.java — puts RTP packets received over UDP back into sequence order + * + */ + +package com.openipc.decoder; + +/** + * UDP may reorder datagrams; the depacketizer expects them in sequence order and treats a + * gap as loss (#67). In-order packets pass straight through, so a clean network adds no + * delay. A packet that arrives early is held until the missing ones arrive, the hold time + * runs out, or too many are held; then the gap is skipped. Late duplicates are dropped. + * Single-threaded: called from the network thread only. + */ +class RtpReorder { + interface Sink { + void deliver(byte[] buf, int offset, int length); + } + + /** Ring size; a jump of this many packets or more is not waited for. */ + private static final int SLOTS = 128; + /** A sequence number this far behind is a restarted stream, not a late packet. */ + private static final int RESTART_BEHIND = 1000; + + private final int maxHeld; + private final long maxHoldNs; + + private final byte[][] data = new byte[SLOTS][]; + private final int[] length = new int[SLOTS]; + private final long[] heldAtNs = new long[SLOTS]; + private final boolean[] present = new boolean[SLOTS]; + private int held; + private int expected = -1; + private long dropped; + private long skipped; + + RtpReorder(int maxHeld, long maxHoldMs) { + this.maxHeld = Math.min(maxHeld, SLOTS - 1); + this.maxHoldNs = maxHoldMs * 1_000_000L; + } + + void reset() { + for (int i = 0; i < SLOTS; i++) present[i] = false; + held = 0; + expected = -1; + } + + /** Late or duplicate packets dropped so far. */ + long dropped() { return dropped; } + /** Missing packets given up on so far. */ + long skipped() { return skipped; } + int held() { return held; } + + static int seq(byte[] buf, int offset) { + return ((buf[offset + 2] & 0xFF) << 8) | (buf[offset + 3] & 0xFF); + } + + void push(byte[] buf, int offset, int len, long nowNs, Sink sink) { + if (len < 12) return; + int seq = seq(buf, offset); + if (expected < 0) { + expected = seq; + } + int d = (short) (seq - expected); // signed distance, wraps at 16 bits + if (d == 0) { + sink.deliver(buf, offset, len); + expected = (seq + 1) & 0xFFFF; + drain(sink); + } else if (d < 0) { + if (d > -RESTART_BEHIND) { + dropped++; + return; + } + restart(seq, buf, offset, len, sink); + } else if (d >= SLOTS) { + // Too far ahead to wait for: hand over what is held, then continue from here + flushAll(sink); + restart(seq, buf, offset, len, sink); + } else { + int i = seq & (SLOTS - 1); + if (present[i]) { + dropped++; + return; + } + if (data[i] == null || data[i].length < len) data[i] = new byte[Math.max(len, 2048)]; + System.arraycopy(buf, offset, data[i], 0, len); + length[i] = len; + heldAtNs[i] = nowNs; + present[i] = true; + held++; + if (held > maxHeld) skipGap(sink); + } + } + + /** Give up on a gap whose oldest held packet has waited too long. */ + void poll(long nowNs, Sink sink) { + while (held > 0 && nowNs - oldestHeldNs() >= maxHoldNs) skipGap(sink); + } + + /** Time until {@link #poll} has something to do, in ms; -1 when nothing is held. */ + long msUntilDeadline(long nowNs) { + if (held == 0) return -1; + long ns = oldestHeldNs() + maxHoldNs - nowNs; + return ns <= 0 ? 0 : Math.max(1, ns / 1_000_000); + } + + private long oldestHeldNs() { + long oldest = Long.MAX_VALUE; + for (int i = 0; i < SLOTS; i++) { + if (present[i] && heldAtNs[i] < oldest) oldest = heldAtNs[i]; + } + return oldest; + } + + private void restart(int seq, byte[] buf, int offset, int len, Sink sink) { + reset(); + sink.deliver(buf, offset, len); + expected = (seq + 1) & 0xFFFF; + } + + /** Deliver the held packets that now follow on. */ + private void drain(Sink sink) { + while (held > 0) { + int i = expected & (SLOTS - 1); + if (!present[i]) return; + present[i] = false; + held--; + sink.deliver(data[i], 0, length[i]); + expected = (expected + 1) & 0xFFFF; + } + } + + /** Skip to the first held packet. */ + private void skipGap(Sink sink) { + for (int k = 1; k < SLOTS; k++) { + if (present[(expected + k) & (SLOTS - 1)]) { + skipped += k; + expected = (expected + k) & 0xFFFF; + drain(sink); + return; + } + } + } + + private void flushAll(Sink sink) { + while (held > 0) skipGap(sink); + } +} diff --git a/android/app/src/main/java/com/openipc/decoder/RtspClient.java b/android/app/src/main/java/com/openipc/decoder/RtspClient.java index 2bc3b6d5..c0cbc69c 100644 --- a/android/app/src/main/java/com/openipc/decoder/RtspClient.java +++ b/android/app/src/main/java/com/openipc/decoder/RtspClient.java @@ -64,8 +64,8 @@ interface Listener { void setAudioManager(AudioManager am) { this.audioManager = am; } void setAudioRoute(int route) { this.audioRoute = route; } - void configure(String host, String userAgent) { - session.configure(host, userAgent, true); + void configure(String host, String userAgent, Transport.Mode transport) { + session.configure(host, userAgent, true, transport); } boolean isActive() { return session.isActive(); } 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 78e7345a..b44701e7 100644 --- a/android/app/src/main/java/com/openipc/decoder/RtspSession.java +++ b/android/app/src/main/java/com/openipc/decoder/RtspSession.java @@ -1,7 +1,7 @@ /* * Copyright (c) OpenIPC https://openipc.org Prosperity Public License 3.0.0 * - * RtspSession.java — one RTSP-over-TCP session with automatic reconnect + * RtspSession.java — one RTSP session with automatic reconnect * */ @@ -14,6 +14,7 @@ import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; +import java.net.InetAddress; import java.net.InetSocketAddress; import java.net.Socket; import java.nio.charset.StandardCharsets; @@ -22,11 +23,13 @@ import java.util.List; import java.util.Locale; import java.util.Map; +import java.util.concurrent.atomic.AtomicReference; /** - * Handshake (DESCRIBE, SETUP, PLAY), SDP parsing, the interleaved receive loop, 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. + * 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. */ class RtspSession { private static final String TAG = "OpenIPCDecoder"; @@ -42,6 +45,8 @@ class RtspSession { /** Lower bound for the keepalive interval (half the session timeout). */ private static final int MIN_KEEPALIVE_MS = 250; static volatile int maxRetryMs = 30000; + /** Auto transport: how long to wait for the first UDP video packet before using TCP. */ + static volatile long udpFirstPacketMs = 3000; interface Callback { /** "disconnected", "buffering" (after PLAY) or "connected" (first video packet). */ @@ -63,11 +68,18 @@ interface Callback { private volatile String url; private volatile String userAgent = ""; private volatile boolean withAudio; + private volatile Transport.Mode mode = Transport.Mode.TCP; + /** Auto transport: UDP failed for this URL, stay on TCP until reconfigured. */ + private volatile boolean udpFailed; // Incremented by start()/stop(); threads and callbacks of an older generation do nothing private volatile int generation; private final Object lock = new Object(); private Socket socket; + /** UDP receiver of the current connection, else null. */ + private UdpReceiver udp; + /** The UDP receive loop is running: closing ends it, the network thread closes the rest. */ + private boolean udpPlaying; private Thread networkThread; private Thread watchdogThread; @@ -85,9 +97,15 @@ interface Callback { /** @param userAgent full header line, e.g. "User-Agent: x\r\n" */ void configure(String url, String userAgent, boolean withAudio) { + configure(url, userAgent, withAudio, Transport.Mode.TCP); + } + + void configure(String url, String userAgent, boolean withAudio, Transport.Mode mode) { this.url = url; this.userAgent = userAgent != null ? userAgent : ""; this.withAudio = withAudio; + this.mode = mode != null ? mode : Transport.Mode.TCP; + this.udpFailed = false; } boolean isActive() { return active; } @@ -118,7 +136,15 @@ void stop() { } } + /** + * End the current connection. While playing over UDP only the receiver is closed: the + * network thread then sends TEARDOWN, so the server stops sending to our ports. + */ private void closeSocketLocked() { + if (udp != null) { + udp.close(); + if (udpPlaying) return; + } if (socket != null) { try { socket.close(); @@ -210,6 +236,8 @@ private boolean connect(int gen) throws IOException { socket = s; } Conn c = null; + UdpReceiver rx = null; + String session = null; try { s.connect(new InetSocketAddress(u.host, u.port), CONNECT_TIMEOUT_MS); s.setTcpNoDelay(true); @@ -232,19 +260,48 @@ private boolean connect(int gen) throws IOException { throw new IOException("No H.264/H.265 video track in SDP"); } - r = exchange(c, "SETUP", sdp.videoUrl, - "Transport: RTP/AVP/TCP;unicast;interleaved=0-1\r\n"); + Transport.Mode m = mode; + boolean auto = m == Transport.Mode.AUTO; + if (m == Transport.Mode.UDP || (auto && !udpFailed)) { + rx = new UdpReceiver(tag); + synchronized (lock) { + if (gen != generation) return false; + udp = rx; + } + } + + String transport = rx != null + ? Transport.udpRequest(rx.addTrack(CHANNEL_VIDEO)) + : Transport.tcpRequest(CHANNEL_VIDEO); + r = exchange(c, "SETUP", sdp.videoUrl, "Transport: " + transport + "\r\n"); + if (rx != null && auto && !r.ok()) { + // 461 Unsupported Transport and the like: the server does not do UDP + Log.i(TAG, tag + ": UDP SETUP failed (" + r.code + "), using TCP"); + udpFailed = true; + rx = dropUdp(rx); + r = exchange(c, "SETUP", sdp.videoUrl, + "Transport: " + Transport.tcpRequest(CHANNEL_VIDEO) + "\r\n"); + } r.require("SETUP video"); String sessionHeader = r.header("session"); if (sessionHeader == null) throw new IOException("No Session header in SETUP reply"); - String session = sessionHeader.split(";")[0].trim(); + session = sessionHeader.split(";")[0].trim(); long keepaliveMs = keepaliveInterval(sessionHeader); + Transport reply = Transport.parse(r.header("transport")); + if (rx != null && reply.tcp) { + // Asked for UDP, answered with interleaved TCP: take what the server chose + Log.i(TAG, tag + ": server chose TCP instead of UDP"); + rx = dropUdp(rx); + } boolean audio = withAudio && sdp.audioUrl != null && !AudioConfig.NONE.equals(sdp.audio.codec); if (audio) { + transport = rx != null + ? Transport.udpRequest(rx.addTrack(CHANNEL_AUDIO)) + : Transport.tcpRequest(CHANNEL_AUDIO); r = exchange(c, "SETUP", sdp.audioUrl, - "Transport: RTP/AVP/TCP;unicast;interleaved=2-3\r\n" + + "Transport: " + transport + "\r\n" + "Session: " + session + "\r\n"); if (!r.ok()) { // Not fatal: play video without audio @@ -254,6 +311,12 @@ private boolean connect(int gen) throws IOException { } if (!audio) sdp.audio = new AudioConfig(); + if (rx != null) { + // Only the camera may send to our ports; "source" may be its own view of + // its address (NAT), so the address we connected to is accepted as well + rx.setSources(s.getInetAddress(), sourceAddress(reply.source)); + } + if (gen != generation) return false; callback.onSdp(sdp); @@ -268,42 +331,68 @@ private boolean connect(int gen) throws IOException { bitrateMarkNs = System.nanoTime(); active = true; keepalive = new Keepalive(c, u.requestUrl, session, keepaliveMs); + udpPlaying = rx != null; } + Log.i(TAG, tag + ": playing over " + (rx != null ? "UDP" : "TCP")); status(gen, "buffering"); s.setSoTimeout(0); // the watchdog handles silence - receive(gen, c); + if (rx != null) { + receiveUdp(gen, c, rx, auto); + } else { + receive(gen, c); + } return true; } finally { synchronized (lock) { if (socket == s) socket = null; + if (udp == rx) udp = null; + udpPlaying = false; Keepalive k = keepalive; if (k != null && k.conn == c) keepalive = null; } + if (rx != null) { + rx.close(); + if (session != null && !s.isClosed()) { + // Over UDP the server keeps sending until told to stop or until the + // session times out; cameras that allow one client need it gone (#67) + try { + send(c, "TEARDOWN", u.requestUrl, "Session: " + session + "\r\n"); + } catch (IOException ignored) {} + } + } try { s.close(); } catch (IOException ignored) {} } } + private UdpReceiver dropUdp(UdpReceiver rx) { + rx.close(); + synchronized (lock) { + if (udp == rx) udp = null; + } + return null; + } + + /** The "source" of a Transport reply as an address; null unless it is a literal IP. */ + static InetAddress sourceAddress(String source) { + if (source == null || !source.matches("[0-9.]+|[0-9A-Fa-f:.]*:[0-9A-Fa-f:.]*")) { + return null; // never resolve a host name here + } + try { + return InetAddress.getByName(source); + } catch (IOException e) { + return null; + } + } + /** Receive loop; keepalive replies are recognised between interleaved packets. */ private void receive(int gen, Conn c) throws IOException { byte[] buf = new byte[65535 + 4]; boolean first = true; while (active && gen == generation) { if (isResponseNext(c.in)) { - Response r = readResponse(c.in); - Keepalive k = keepalive; - if (r.code == 401) { - if (!c.auth.onUnauthorized(r.authenticate)) { - throw new IOException("Keepalive rejected: authentication failed"); - } - if (k != null) k.lastSentMs = 0; // repeat it now with the new nonce - } else if (r.code == 454) { - throw new IOException("Session expired on the server"); - } else if (!r.ok() && k != null && "GET_PARAMETER".equals(k.pending)) { - Log.i(TAG, tag + ": GET_PARAMETER rejected (" + r.code + "), using OPTIONS"); - k.useOptions = true; - } + handleReply(c, readResponse(c.in)); continue; } @@ -324,6 +413,73 @@ private void receive(int gen, Conn c) throws IOException { } } + /** + * Receive over UDP on this thread while a second thread reads the RTSP replies + * (keepalives). With {@code auto}, no video within udpFirstPacketMs switches the + * URL to TCP for the next connection. + */ + private void receiveUdp(int gen, Conn c, UdpReceiver rx, boolean auto) throws IOException { + AtomicReference controlError = new AtomicReference<>(); + Thread control = new Thread(() -> { + try { + while (gen == generation) { + Response r = readResponse(c.in); + handleReply(c, r); + } + } catch (IOException e) { + controlError.set(e); + } + // Ends the receive loop; harmless when it already ended + rx.close(); + }, tag + "-ctl"); + control.setDaemon(true); + control.start(); + + final boolean[] first = {true}; + long playMs = System.nanoTime() / 1_000_000; + rx.run(() -> { + if (!active || gen != generation) return false; + if (auto && first[0] + && System.nanoTime() / 1_000_000 - playMs > udpFirstPacketMs) { + Log.w(TAG, tag + ": no video over UDP, using TCP"); + udpFailed = true; + return false; + } + return true; + }, (channel, buf, offset, length) -> { + if (gen != generation) return; + streamBytes += length; + boolean usable = callback.onPacket(channel, buf, offset, length); + if (channel == CHANNEL_VIDEO && usable) { + lastPacketMs = System.nanoTime() / 1_000_000; + if (first[0]) { + first[0] = false; + status(gen, "connected"); + } + } + }); + IOException e = controlError.get(); + if (e != null && gen == generation && active) { + throw new IOException("RTSP connection lost: " + e.getMessage(), e); + } + } + + /** A reply that arrived while playing: keepalive answers and session errors. */ + private void handleReply(Conn c, Response r) throws IOException { + Keepalive k = keepalive; + if (r.code == 401) { + if (!c.auth.onUnauthorized(r.authenticate)) { + throw new IOException("Keepalive rejected: authentication failed"); + } + if (k != null) k.lastSentMs = 0; // repeat it now with the new nonce + } else if (r.code == 454) { + throw new IOException("Session expired on the server"); + } else if (!r.ok() && k != null && "GET_PARAMETER".equals(k.pending)) { + Log.i(TAG, tag + ": GET_PARAMETER rejected (" + r.code + "), using OPTIONS"); + k.useOptions = true; + } + } + private static final class Keepalive { final Conn conn; final String url; diff --git a/android/app/src/main/java/com/openipc/decoder/Transport.java b/android/app/src/main/java/com/openipc/decoder/Transport.java new file mode 100644 index 00000000..beb41418 --- /dev/null +++ b/android/app/src/main/java/com/openipc/decoder/Transport.java @@ -0,0 +1,111 @@ +/* + * Copyright (c) OpenIPC https://openipc.org Prosperity Public License 3.0.0 + * + * Transport.java — RTP transport choice of a camera slot and the RTSP Transport header + * + */ + +package com.openipc.decoder; + +import java.util.Locale; + +/** + * The Transport header of a SETUP reply (RFC 2326 12.39). Servers word it differently: + * Majestic answers "RTP/AVP/UDP;unicast;client_port=a-b;server_port=c-d", XM cameras + * "RTP/AVP;unicast;mode=PLAY;source=ip;client_port=a-b;server_port=c-d;ssrc=x", and both + * downgrade a multicast request to unicast, so the reply decides what is received (#67). + */ +class Transport { + /** Slot setting: RTP over the RTSP connection, over UDP, or UDP with fallback to TCP. */ + enum Mode { + TCP, UDP, AUTO; + + /** Stored or shared value ("tcp", "udp", "auto"); unknown values mean TCP. */ + static Mode parse(String s) { + if (s != null) { + for (Mode m : values()) { + if (m.name().equalsIgnoreCase(s.trim())) return m; + } + } + return TCP; + } + } + + /** RTP over the RTSP connection ("RTP/AVP/TCP" or an interleaved= parameter). */ + boolean tcp; + boolean multicast; + int[] interleaved; + int[] clientPort; + int[] serverPort; + /** Multicast group, or null. */ + String destination; + /** Multicast ports. */ + int[] port; + int ttl = -1; + /** Address the server sends from, or null when not given. */ + String source; + + static Transport parse(String header) { + Transport t = new Transport(); + if (header == null) return t; + // A server may list several transports; the reply carries the chosen one first + String first = header.split(",")[0]; + String[] parts = first.split(";"); + String spec = parts[0].trim().toUpperCase(Locale.ROOT); + t.tcp = spec.endsWith("/TCP"); + for (int i = 1; i < parts.length; i++) { + String p = parts[i].trim(); + int eq = p.indexOf('='); + String name = (eq < 0 ? p : p.substring(0, eq)).trim().toLowerCase(Locale.ROOT); + String value = eq < 0 ? "" : p.substring(eq + 1).trim(); + switch (name) { + case "multicast": t.multicast = true; break; + case "unicast": t.multicast = false; break; + case "interleaved": + t.interleaved = range(value); + if (t.interleaved != null) t.tcp = true; + break; + case "client_port": t.clientPort = range(value); break; + case "server_port": t.serverPort = range(value); break; + case "port": t.port = range(value); break; + case "destination": t.destination = unquote(value); break; + case "source": t.source = unquote(value); break; + case "ttl": + try { + t.ttl = Integer.parseInt(value); + } catch (NumberFormatException ignored) {} + break; + default: break; + } + } + return t; + } + + /** "a-b" or "a" as {a, b} (b = a + 1 when absent), or null if not a port pair. */ + static int[] range(String value) { + String[] ab = value.split("-", 2); + try { + int a = Integer.parseInt(ab[0].trim()); + int b = ab.length > 1 ? Integer.parseInt(ab[1].trim()) : a + 1; + if (a < 0 || a > 65535 || b < 0 || b > 65535) return null; + return new int[] {a, b}; + } catch (NumberFormatException e) { + return null; + } + } + + private static String unquote(String v) { + return v.length() >= 2 && v.startsWith("\"") && v.endsWith("\"") + ? v.substring(1, v.length() - 1) : v; + } + + /** Request header value for interleaved RTP on channels {@code ch} and {@code ch + 1}. */ + static String tcpRequest(int ch) { + return "RTP/AVP/TCP;unicast;interleaved=" + ch + "-" + (ch + 1); + } + + /** Request header value for UDP unicast to local ports {@code rtp} and {@code rtp + 1}. */ + static String udpRequest(int rtp) { + return "RTP/AVP;unicast;client_port=" + rtp + "-" + (rtp + 1); + } +} diff --git a/android/app/src/main/java/com/openipc/decoder/UdpReceiver.java b/android/app/src/main/java/com/openipc/decoder/UdpReceiver.java new file mode 100644 index 00000000..8afcc97c --- /dev/null +++ b/android/app/src/main/java/com/openipc/decoder/UdpReceiver.java @@ -0,0 +1,215 @@ +/* + * Copyright (c) OpenIPC https://openipc.org Prosperity Public License 3.0.0 + * + * UdpReceiver.java — RTP over UDP: local port pairs and the receive loop + * + */ + +package com.openipc.decoder; + +import android.util.Log; + +import java.io.Closeable; +import java.io.IOException; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.SocketAddress; +import java.net.StandardSocketOptions; +import java.nio.ByteBuffer; +import java.nio.channels.ClosedSelectorException; +import java.nio.channels.DatagramChannel; +import java.nio.channels.SelectionKey; +import java.nio.channels.Selector; +import java.util.ArrayList; +import java.util.List; +import java.util.function.BooleanSupplier; + +/** + * One RTP/RTCP port pair per track, bound to ephemeral ports (never fixed ones, see #40), + * and one loop that receives every track on the calling thread, so packets reach the + * callback from a single thread as they do over TCP (#67). RTCP is read and discarded. + */ +class UdpReceiver implements Closeable { + private static final String TAG = "OpenIPCDecoder"; + + interface Handler { + void onPacket(int channel, byte[] buf, int offset, int length); + } + + /** Hold time of the reorder buffer; only spent when a packet is missing. */ + static final int REORDER_MS = 40; + private static final int REORDER_PACKETS = 64; + private static final int VIDEO_RCVBUF = 2 * 1024 * 1024; + private static final int AUDIO_RCVBUF = 256 * 1024; + private static final int BIND_ATTEMPTS = 32; + + private static final class Track { + final int channel; + final DatagramChannel rtp; + final DatagramChannel rtcp; + final RtpReorder reorder = new RtpReorder(REORDER_PACKETS, REORDER_MS); + /** Delivers to the handler of the running loop; one per track, not per packet. */ + RtpReorder.Sink sink; + + Track(int channel, DatagramChannel rtp, DatagramChannel rtcp) { + this.channel = channel; + this.rtp = rtp; + this.rtcp = rtcp; + } + } + + private final String tag; + private final Selector selector; + private final List tracks = new ArrayList<>(); + /** Accepted senders; empty means any. */ + private final List sources = new ArrayList<>(); + private volatile boolean closed; + private boolean running; + private boolean released; + private long rejected; + + UdpReceiver(String tag) throws IOException { + this.tag = tag; + this.selector = Selector.open(); + } + + /** + * Bind an even RTP port and the odd RTCP port above it for {@code channel} + * (RtspSession.CHANNEL_VIDEO or CHANNEL_AUDIO). Returns the RTP port. + */ + synchronized int addTrack(int channel) throws IOException { + if (closed) throw new IOException("Receiver closed"); + for (int attempt = 0; attempt < BIND_ATTEMPTS; attempt++) { + DatagramChannel rtp = DatagramChannel.open(); + DatagramChannel rtcp = null; + try { + rtp.bind(new InetSocketAddress(0)); + int port = ((InetSocketAddress) rtp.getLocalAddress()).getPort(); + if ((port & 1) != 0 || port >= 65535) { + rtp.close(); + continue; + } + rtcp = DatagramChannel.open(); + rtcp.bind(new InetSocketAddress(port + 1)); + int size = channel == RtspSession.CHANNEL_VIDEO ? VIDEO_RCVBUF : AUDIO_RCVBUF; + rtp.setOption(StandardSocketOptions.SO_RCVBUF, size); + rtp.configureBlocking(false); + rtcp.configureBlocking(false); + Track t = new Track(channel, rtp, rtcp); + rtp.register(selector, SelectionKey.OP_READ, t); + rtcp.register(selector, SelectionKey.OP_READ, null); + tracks.add(t); + return port; + } catch (IOException e) { + // Port above taken by someone else: try another pair + rtp.close(); + if (rtcp != null) rtcp.close(); + } + } + throw new IOException("No free UDP port pair"); + } + + /** Accept datagrams only from these addresses (nulls are ignored). */ + synchronized void setSources(InetAddress... addresses) { + sources.clear(); + for (InetAddress a : addresses) { + if (a != null && !sources.contains(a)) sources.add(a); + } + } + + /** + * Receive until {@code keepGoing} turns false or {@link #close} is called. The + * condition is checked at least every 250 ms. + */ + void run(BooleanSupplier keepGoing, Handler handler) throws IOException { + List accepted; + synchronized (this) { + if (closed) return; + running = true; + accepted = new ArrayList<>(sources); + } + for (Track t : tracks) { + t.sink = (b, off, len) -> handler.onPacket(t.channel, b, off, len); + } + ByteBuffer buf = ByteBuffer.allocate(65536); + byte[] array = buf.array(); + try { + while (!closed && keepGoing.getAsBoolean()) { + long now = System.nanoTime(); + long timeout = 250; + for (Track t : tracks) { + long ms = t.reorder.msUntilDeadline(now); + if (ms >= 0) timeout = Math.max(1, Math.min(timeout, ms)); + } + selector.select(timeout); + for (SelectionKey key : selector.selectedKeys()) { + if (!key.isValid()) continue; + DatagramChannel ch = (DatagramChannel) key.channel(); + Track t = (Track) key.attachment(); + SocketAddress from; + while ((from = receive(ch, buf)) != null) { + if (t == null) continue; // RTCP: not used + if (!accepted.isEmpty() && !accepted.contains( + ((InetSocketAddress) from).getAddress())) { + if (rejected++ % 100 == 0) { + Log.w(TAG, tag + ": UDP from an unexpected sender ignored (" + + rejected + ")"); + } + continue; + } + t.reorder.push(array, 0, buf.position(), System.nanoTime(), t.sink); + } + } + selector.selectedKeys().clear(); + now = System.nanoTime(); + for (Track t : tracks) { + t.reorder.poll(now, t.sink); + } + } + } catch (ClosedSelectorException e) { + // close() from another thread + } finally { + for (Track t : tracks) { + if (t.reorder.dropped() > 0 || t.reorder.skipped() > 0) { + Log.i(TAG, tag + ": UDP channel " + t.channel + ": " + t.reorder.skipped() + + " packets lost, " + t.reorder.dropped() + " late or duplicate"); + } + } + release(); + } + } + + private static SocketAddress receive(DatagramChannel ch, ByteBuffer buf) throws IOException { + buf.clear(); + return ch.receive(buf); + } + + /** Ends {@link #run} from any thread; releases the ports when it is not running. */ + @Override + public void close() { + boolean releaseNow; + synchronized (this) { + closed = true; + releaseNow = !running; + } + selector.wakeup(); + if (releaseNow) release(); + } + + private synchronized void release() { + if (released) return; + released = true; + closed = true; + for (Track t : tracks) { + closeQuietly(t.rtp); + closeQuietly(t.rtcp); + } + closeQuietly(selector); + } + + private static void closeQuietly(Closeable c) { + try { + c.close(); + } catch (IOException ignored) {} + } +} diff --git a/android/app/src/test/java/com/openipc/decoder/QuadAudioTest.java b/android/app/src/test/java/com/openipc/decoder/QuadAudioTest.java index 8ed19e60..40862782 100644 --- a/android/app/src/test/java/com/openipc/decoder/QuadAudioTest.java +++ b/android/app/src/test/java/com/openipc/decoder/QuadAudioTest.java @@ -40,7 +40,7 @@ private static byte[] rtp(int pt) { @Test public void onlyTheSelectedCellPlaysAudio() { - QuadCell cell = new QuadCell(1, "rtsp://cam/s", null, new FramePool(4), "", index -> {}); + QuadCell cell = new QuadCell(1, "rtsp://cam/s", Transport.Mode.TCP, null, new FramePool(4), "", index -> {}); CountingAudio audio = new CountingAudio(); byte[] a = rtp(98); @@ -64,7 +64,7 @@ public void onlyTheSelectedCellPlaysAudio() { @Test public void audioConfigGoesToTheSinkWithTheCellsRoute() { - QuadCell cell = new QuadCell(2, "rtsp://cam/s", null, new FramePool(4), "", index -> {}); + QuadCell cell = new QuadCell(2, "rtsp://cam/s", Transport.Mode.TCP, null, new FramePool(4), "", index -> {}); CountingAudio audio = new CountingAudio(); Sdp sdp = Sdp.parse("m=video 0 RTP/AVP 96\r\nm=audio 0 RTP/AVP 0\r\n", "rtsp://cam/s"); diff --git a/android/app/src/test/java/com/openipc/decoder/RtpReorderTest.java b/android/app/src/test/java/com/openipc/decoder/RtpReorderTest.java new file mode 100644 index 00000000..19d2216b --- /dev/null +++ b/android/app/src/test/java/com/openipc/decoder/RtpReorderTest.java @@ -0,0 +1,115 @@ +/* + * Copyright (c) OpenIPC https://openipc.org Prosperity Public License 3.0.0 + * + * RtpReorderTest.java — sequence ordering of RTP packets received over UDP + * + */ + +package com.openipc.decoder; + +import static org.junit.Assert.assertEquals; + +import org.junit.Test; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +public class RtpReorderTest { + private static final long MS = 1_000_000L; + + private final List out = new ArrayList<>(); + private final RtpReorder.Sink sink = (buf, offset, length) -> { + assertEquals(14, length); + out.add(RtpReorder.seq(buf, offset)); + }; + + private static byte[] packet(int seq) { + byte[] p = new byte[14]; + p[0] = (byte) 0x80; + p[1] = 96; + p[2] = (byte) (seq >> 8); + p[3] = (byte) seq; + return p; + } + + private void push(RtpReorder r, long nowNs, int... seqs) { + for (int s : seqs) { + byte[] p = packet(s); + r.push(p, 0, p.length, nowNs, sink); + // The caller's buffer is reused for the next datagram + Arrays.fill(p, (byte) 0); + } + } + + @Test + public void inOrderPassesStraightThrough() { + RtpReorder r = new RtpReorder(64, 40); + push(r, 0, 10, 11, 12); + assertEquals(Arrays.asList(10, 11, 12), out); + assertEquals(-1, r.msUntilDeadline(0)); + } + + @Test + public void swappedPacketsAreReordered() { + RtpReorder r = new RtpReorder(64, 40); + push(r, 0, 1, 3, 2, 4); + assertEquals(Arrays.asList(1, 2, 3, 4), out); + assertEquals(0, r.skipped()); + } + + @Test + public void lostPacketIsSkippedAfterTheHoldTime() { + RtpReorder r = new RtpReorder(64, 40); + push(r, 0, 1, 3, 4); + assertEquals(Arrays.asList(1), out); + assertEquals(40, r.msUntilDeadline(0)); + r.poll(39 * MS, sink); + assertEquals(Arrays.asList(1), out); + r.poll(40 * MS, sink); + assertEquals(Arrays.asList(1, 3, 4), out); + assertEquals(1, r.skipped()); + // The missing one arriving late is dropped + push(r, 41 * MS, 2); + assertEquals(Arrays.asList(1, 3, 4), out); + assertEquals(1, r.dropped()); + } + + @Test + public void tooManyHeldSkipsTheGapAtOnce() { + RtpReorder r = new RtpReorder(3, 1000); + push(r, 0, 1, 3, 4, 5); + assertEquals(Arrays.asList(1), out); + push(r, 0, 6); + assertEquals(Arrays.asList(1, 3, 4, 5, 6), out); + } + + @Test + public void duplicatesAreDropped() { + RtpReorder r = new RtpReorder(64, 40); + push(r, 0, 1, 2, 2, 4, 4, 3); + assertEquals(Arrays.asList(1, 2, 3, 4), out); + assertEquals(2, r.dropped()); + } + + @Test + public void sequenceWrapsAround() { + RtpReorder r = new RtpReorder(64, 40); + push(r, 0, 65534, 0, 65535, 1); + assertEquals(Arrays.asList(65534, 65535, 0, 1), out); + } + + @Test + public void bigJumpRestartsWithoutWaiting() { + RtpReorder r = new RtpReorder(64, 40); + push(r, 0, 100, 102, 5000, 5001); + assertEquals(Arrays.asList(100, 102, 5000, 5001), out); + } + + @Test + public void streamRestartFarBehindIsFollowed() { + RtpReorder r = new RtpReorder(64, 40); + push(r, 0, 30000, 30001, 7, 8); + assertEquals(Arrays.asList(30000, 30001, 7, 8), out); + } +} diff --git a/android/app/src/test/java/com/openipc/decoder/RtspSessionTest.java b/android/app/src/test/java/com/openipc/decoder/RtspSessionTest.java index 3e8b359f..05c99b4e 100644 --- a/android/app/src/test/java/com/openipc/decoder/RtspSessionTest.java +++ b/android/app/src/test/java/com/openipc/decoder/RtspSessionTest.java @@ -19,7 +19,10 @@ import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; +import java.net.DatagramPacket; +import java.net.DatagramSocket; import java.net.InetAddress; +import java.net.InetSocketAddress; import java.net.ServerSocket; import java.net.Socket; import java.nio.charset.StandardCharsets; @@ -38,6 +41,14 @@ public class RtspSessionTest { private static final String SDP_VIDEO_ONLY = "v=0\r\nm=video 0 RTP/AVP 96\r\na=rtpmap:96 H264/90000\r\na=control:trackID=0\r\n"; + /** RTP packet as sent over UDP: no interleaved header, a real sequence number. */ + static byte[] udpPacket(int pt, int seq) { + byte[] p = FakeServer.packet(0, pt); + p[6] = (byte) (seq >> 8); + p[7] = (byte) seq; + return java.util.Arrays.copyOfRange(p, 4, p.length); + } + /** Scripted RTSP server: answers each connection with the configured behaviour. */ private static final class FakeServer implements AutoCloseable { final ServerSocket server; @@ -65,6 +76,12 @@ private static final class FakeServer implements AutoCloseable { volatile boolean streamContinuously; final AtomicInteger unauthorized = new AtomicInteger(); volatile int audioPackets = 1; + /** Reply 461 Unsupported Transport to a UDP SETUP. */ + volatile boolean udpUnsupported; + /** Accept a UDP SETUP but never send a datagram. */ + volatile boolean udpSilent; + /** Send the datagrams from this local address instead of 127.0.0.1. */ + volatile String udpFrom = "127.0.0.1"; FakeServer() throws IOException { server = new ServerSocket(0, 50, InetAddress.getLoopbackAddress()); @@ -97,6 +114,8 @@ private void serve(Socket s, int n) { if (n <= failFirst) return; InputStream in = new BufferedInputStream(socket.getInputStream()); OutputStream out = socket.getOutputStream(); + // Client RTP ports of this connection's UDP SETUPs: video, audio + int[] udpPorts = {-1, -1}; while (true) { String line = RtspSession.readLine(in); if (line == null) return; @@ -129,10 +148,24 @@ private void serve(Socket s, int n) { + "\r\n" + sdp; break; case "SETUP": - int code = requestLine.contains("trackID=1") ? audioSetupStatus : 200; + boolean audioTrack = requestLine.contains("trackID=1"); + int code = audioTrack ? audioSetupStatus : 200; + Transport asked = Transport.parse(header(headers, "Transport")); + String transport = ""; + if (asked.clientPort != null) { + if (udpUnsupported) code = 461; + if (code == 200) udpPorts[audioTrack ? 1 : 0] = asked.clientPort[0]; + transport = "Transport: RTP/AVP/UDP;unicast;client_port=" + + asked.clientPort[0] + "-" + asked.clientPort[1] + + ";server_port=6970-6971\r\n"; + } else if (asked.interleaved != null) { + transport = "Transport: RTP/AVP/TCP;unicast;interleaved=" + + asked.interleaved[0] + "-" + asked.interleaved[1] + + "\r\n"; + } reply = "RTSP/1.0 " + code + (code == 200 ? " OK" : " Not Found") - + "\r\nCSeq: " + cseq + "\r\nSession: ABC123;timeout=" - + sessionTimeout + "\r\n\r\n"; + + "\r\nCSeq: " + cseq + "\r\n" + (code == 200 ? transport : "") + + "Session: ABC123;timeout=" + sessionTimeout + "\r\n\r\n"; break; case "GET_PARAMETER": if (keepaliveUnauthorized) { @@ -149,6 +182,10 @@ private void serve(Socket s, int n) { break; } write(out, reply); + if (method.equals("PLAY") && udpPorts[0] > 0) { + if (!udpSilent) sendUdp(udpPorts); + continue; + } if (method.equals("PLAY") && streamContinuously) { Thread writer = new Thread(() -> { try { @@ -179,6 +216,36 @@ private void serve(Socket s, int n) { } } + private void sendUdp(int[] ports) { + Thread sender = new Thread(() -> { + try (DatagramSocket ds = new DatagramSocket( + new InetSocketAddress(InetAddress.getByName(udpFrom), 0))) { + InetAddress to = InetAddress.getLoopbackAddress(); + int seq = 100; + for (int i = 0; i < videoPackets; i++) { + byte[] p = udpPacket(96, seq++); + ds.send(new DatagramPacket(p, p.length, to, ports[0])); + } + for (int i = 0; i < audioPackets && ports[1] > 0; i++) { + byte[] p = udpPacket(98, 500 + i); + ds.send(new DatagramPacket(p, p.length, to, ports[1])); + } + } catch (IOException ignored) { + } + }); + sender.setDaemon(true); + sender.start(); + } + + private static String header(CharSequence headers, String name) { + for (String h : headers.toString().split("\n")) { + if (h.regionMatches(true, 0, name + ":", 0, name.length() + 1)) { + return h.substring(name.length() + 1).trim(); + } + } + return null; + } + private static void write(OutputStream out, String reply) throws IOException { synchronized (out) { out.write(reply.getBytes(StandardCharsets.UTF_8)); @@ -280,6 +347,7 @@ public void tearDown() throws IOException { RtspSession.minRetryMs = 1000; RtspSession.maxRetryMs = 30000; RtspSession.watchdogMs = 8000; + RtspSession.udpFirstPacketMs = 3000; } private void runUntilVideo(String url, boolean withAudio) throws InterruptedException { @@ -512,4 +580,99 @@ public void rejectedKeepaliveEndsTheSession() throws Exception { .filter(x -> x.startsWith("GET_PARAMETER")).count(); assertTrue(perFirstConnection <= 2 * server.connections.get()); } + + // ====================== UDP (#67) ====================== + + private void runUdp(Transport.Mode mode, boolean withAudio) { + session.configure(server.url(null), "User-Agent: test\r\n", withAudio, mode); + session.start(); + } + + private List setups() { + List out = new ArrayList<>(); + for (String r : server.snapshot()) if (r.startsWith("SETUP")) out.add(r); + return out; + } + + @Test + public void udpHandshakeAndPackets() throws Exception { + runUdp(Transport.Mode.UDP, true); + assertTrue("video over UDP", rec.videoLatch.await(5, TimeUnit.SECONDS)); + long deadline = System.currentTimeMillis() + 2000; + while (rec.audio.get() == 0 && System.currentTimeMillis() < deadline) Thread.sleep(10); + assertEquals(1, rec.audio.get()); + + List s = setups(); + assertEquals(2, s.size()); + for (String r : s) { + Transport t = Transport.parse(FakeServer.header(r, "Transport")); + assertFalse(t.tcp); + assertEquals("RTP port is even", 0, t.clientPort[0] % 2); + assertEquals(t.clientPort[0] + 1, t.clientPort[1]); + } + assertTrue(rec.statuses.contains("connected")); + assertEquals(1, server.connections.get()); + } + + @Test + public void autoFallsBackToTcpWhenUdpIsRefused() throws Exception { + server.udpUnsupported = true; + runUdp(Transport.Mode.AUTO, false); + assertTrue("video over TCP", rec.videoLatch.await(5, TimeUnit.SECONDS)); + List s = setups(); + assertTrue(s.get(0).contains("client_port=")); + assertTrue(s.get(1).contains("interleaved=0-1")); + // Same connection: the refusal is answered by a second SETUP, not a reconnect + assertEquals(1, server.connections.get()); + } + + @Test + public void udpModeDoesNotFallBack() throws Exception { + server.udpUnsupported = true; + runUdp(Transport.Mode.UDP, false); + assertFalse(rec.videoLatch.await(500, TimeUnit.MILLISECONDS)); + for (String r : setups()) assertFalse(r.contains("interleaved")); + assertTrue("keeps retrying", server.connections.get() > 1); + } + + @Test + public void autoFallsBackToTcpWhenNoDatagramArrives() throws Exception { + server.udpSilent = true; + RtspSession.udpFirstPacketMs = 300; + runUdp(Transport.Mode.AUTO, false); + assertTrue("video over TCP", rec.videoLatch.await(5, TimeUnit.SECONDS)); + List m = server.methods(); + assertTrue("UDP session torn down", m.contains("TEARDOWN " + base() + " RTSP/1.0")); + List s = setups(); + assertTrue(s.get(0).contains("client_port=")); + assertTrue(s.get(s.size() - 1).contains("interleaved=0-1")); + assertEquals(2, server.connections.get()); + } + + @Test + public void stopTearsDownTheUdpSession() throws Exception { + runUdp(Transport.Mode.UDP, false); + assertTrue(rec.videoLatch.await(5, TimeUnit.SECONDS)); + session.stop(); + long deadline = System.currentTimeMillis() + 2000; + while (!server.methods().contains("TEARDOWN " + base() + " RTSP/1.0") + && System.currentTimeMillis() < deadline) { + Thread.sleep(10); + } + assertTrue(server.methods().contains("TEARDOWN " + base() + " RTSP/1.0")); + } + + @Test + public void datagramsFromAnotherAddressAreIgnored() throws Exception { + // 127.0.0.2 is a different loopback address on Linux (the CI runner) + org.junit.Assume.assumeTrue(System.getProperty("os.name").startsWith("Linux")); + server.udpFrom = "127.0.0.2"; + runUdp(Transport.Mode.UDP, false); + assertFalse(rec.videoLatch.await(500, TimeUnit.MILLISECONDS)); + assertEquals(0, rec.video.get()); + } + + private String base() { + return "rtsp://127.0.0.1:" + server.server.getLocalPort() + "/stream=0"; + } } diff --git a/android/app/src/test/java/com/openipc/decoder/TransportTest.java b/android/app/src/test/java/com/openipc/decoder/TransportTest.java new file mode 100644 index 00000000..72aeb130 --- /dev/null +++ b/android/app/src/test/java/com/openipc/decoder/TransportTest.java @@ -0,0 +1,99 @@ +/* + * Copyright (c) OpenIPC https://openipc.org Prosperity Public License 3.0.0 + * + * TransportTest.java — RTSP Transport header parsing and the slot transport setting + * + */ + +package com.openipc.decoder; + +import static org.junit.Assert.assertArrayEquals; +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 org.junit.Test; + +public class TransportTest { + @Test + public void majesticUdpReply() { + Transport t = Transport.parse( + "RTP/AVP/UDP;unicast;client_port=50000-50001;server_port=46666-46667"); + assertFalse(t.tcp); + assertFalse(t.multicast); + assertArrayEquals(new int[] {50000, 50001}, t.clientPort); + assertArrayEquals(new int[] {46666, 46667}, t.serverPort); + assertNull(t.source); + } + + @Test + public void xmUdpReply() { + Transport t = Transport.parse("RTP/AVP;unicast;mode=PLAY;source=172.17.32.53;" + + "client_port=50000-50001;server_port=40008-40009;ssrc=00001BCC"); + assertFalse(t.tcp); + assertFalse(t.multicast); + assertEquals("172.17.32.53", t.source); + assertArrayEquals(new int[] {40008, 40009}, t.serverPort); + } + + @Test + public void interleavedReplyIsTcp() { + Transport t = Transport.parse("RTP/AVP/TCP;unicast;interleaved=0-1"); + assertTrue(t.tcp); + assertArrayEquals(new int[] {0, 1}, t.interleaved); + // Interleaved without the /TCP profile still means the RTSP connection + assertTrue(Transport.parse("RTP/AVP;unicast;interleaved=2-3").tcp); + } + + @Test + public void multicastReply() { + Transport t = Transport.parse( + "RTP/AVP;multicast;destination=\"239.1.2.3\";port=5000-5001;ttl=4"); + assertTrue(t.multicast); + assertEquals("239.1.2.3", t.destination); + assertArrayEquals(new int[] {5000, 5001}, t.port); + assertEquals(4, t.ttl); + } + + @Test + public void onlyTheFirstOfSeveralTransportsCounts() { + Transport t = Transport.parse( + "RTP/AVP;unicast;client_port=4000-4001,RTP/AVP/TCP;interleaved=0-1"); + assertFalse(t.tcp); + assertArrayEquals(new int[] {4000, 4001}, t.clientPort); + } + + @Test + public void malformedValuesAreIgnored() { + Transport t = Transport.parse("RTP/AVP;client_port=abc;server_port=70000-70001;ttl=x"); + assertNull(t.clientPort); + assertNull(t.serverPort); + assertEquals(-1, t.ttl); + assertFalse(Transport.parse(null).tcp); + assertArrayEquals(new int[] {6970, 6971}, Transport.range("6970")); + } + + @Test + public void requests() { + assertEquals("RTP/AVP/TCP;unicast;interleaved=2-3", Transport.tcpRequest(2)); + assertEquals("RTP/AVP;unicast;client_port=50000-50001", Transport.udpRequest(50000)); + } + + @Test + public void modeFromSettings() { + assertEquals(Transport.Mode.UDP, Transport.Mode.parse("udp")); + assertEquals(Transport.Mode.AUTO, Transport.Mode.parse(" Auto ")); + assertEquals(Transport.Mode.TCP, Transport.Mode.parse("TCP")); + assertEquals(Transport.Mode.TCP, Transport.Mode.parse(null)); + assertEquals(Transport.Mode.TCP, Transport.Mode.parse("quic")); + } + + @Test + public void sourceIsOnlyALiteralAddress() { + assertEquals("172.17.32.53", RtspSession.sourceAddress("172.17.32.53").getHostAddress()); + assertTrue(RtspSession.sourceAddress("::1").isLoopbackAddress()); + assertNull(RtspSession.sourceAddress("camera.example")); + assertNull(RtspSession.sourceAddress(null)); + } +} From 5ffe66c89bbe9ef7fee0e0bd72a7190da73ef0d9 Mon Sep 17 00:00:00 2001 From: Compot Developer Date: Mon, 12 Oct 2026 18:53:33 +0000 Subject: [PATCH 2/3] fix: UDP receiver within API 21 (lint NewApi) --- .../java/com/openipc/decoder/UdpReceiver.java | 20 +++++++++++-------- 1 file changed, 12 insertions(+), 8 deletions(-) diff --git a/android/app/src/main/java/com/openipc/decoder/UdpReceiver.java b/android/app/src/main/java/com/openipc/decoder/UdpReceiver.java index 8afcc97c..54529cc8 100644 --- a/android/app/src/main/java/com/openipc/decoder/UdpReceiver.java +++ b/android/app/src/main/java/com/openipc/decoder/UdpReceiver.java @@ -14,7 +14,6 @@ import java.net.InetAddress; import java.net.InetSocketAddress; import java.net.SocketAddress; -import java.net.StandardSocketOptions; import java.nio.ByteBuffer; import java.nio.channels.ClosedSelectorException; import java.nio.channels.DatagramChannel; @@ -22,7 +21,6 @@ import java.nio.channels.Selector; import java.util.ArrayList; import java.util.List; -import java.util.function.BooleanSupplier; /** * One RTP/RTCP port pair per track, bound to ephemeral ports (never fixed ones, see #40), @@ -36,6 +34,11 @@ interface Handler { void onPacket(int channel, byte[] buf, int offset, int length); } + /** java.util.function.BooleanSupplier needs API 24. */ + interface Condition { + boolean holds(); + } + /** Hold time of the reorder buffer; only spent when a packet is missing. */ static final int REORDER_MS = 40; private static final int REORDER_PACKETS = 64; @@ -83,16 +86,17 @@ synchronized int addTrack(int channel) throws IOException { DatagramChannel rtp = DatagramChannel.open(); DatagramChannel rtcp = null; try { - rtp.bind(new InetSocketAddress(0)); - int port = ((InetSocketAddress) rtp.getLocalAddress()).getPort(); + // Through socket(): DatagramChannel.bind() and setOption() need API 24 + rtp.socket().bind(new InetSocketAddress(0)); + int port = rtp.socket().getLocalPort(); if ((port & 1) != 0 || port >= 65535) { rtp.close(); continue; } rtcp = DatagramChannel.open(); - rtcp.bind(new InetSocketAddress(port + 1)); + rtcp.socket().bind(new InetSocketAddress(port + 1)); int size = channel == RtspSession.CHANNEL_VIDEO ? VIDEO_RCVBUF : AUDIO_RCVBUF; - rtp.setOption(StandardSocketOptions.SO_RCVBUF, size); + rtp.socket().setReceiveBufferSize(size); rtp.configureBlocking(false); rtcp.configureBlocking(false); Track t = new Track(channel, rtp, rtcp); @@ -121,7 +125,7 @@ synchronized void setSources(InetAddress... addresses) { * Receive until {@code keepGoing} turns false or {@link #close} is called. The * condition is checked at least every 250 ms. */ - void run(BooleanSupplier keepGoing, Handler handler) throws IOException { + void run(Condition keepGoing, Handler handler) throws IOException { List accepted; synchronized (this) { if (closed) return; @@ -134,7 +138,7 @@ void run(BooleanSupplier keepGoing, Handler handler) throws IOException { ByteBuffer buf = ByteBuffer.allocate(65536); byte[] array = buf.array(); try { - while (!closed && keepGoing.getAsBoolean()) { + while (!closed && keepGoing.holds()) { long now = System.nanoTime(); long timeout = 250; for (Track t : tracks) { From 0378d9c5521a0405464c20f34fe446780f79959c Mon Sep 17 00:00:00 2001 From: Compot Developer Date: Mon, 12 Oct 2026 19:06:15 +0000 Subject: [PATCH 3/3] fix: UDP transport review findings (#67) - A multicast reply, or one naming other client ports, is a UDP failure: Auto falls back to TCP, UDP reports the error, instead of waiting for datagrams that cannot arrive. - Audio answered with a different transport than UDP video is not set up; the audio reply's source is accepted as a sender too. - A finished connection clears the UDP state only if it still owns it, so a quick restart keeps the new session's TEARDOWN. - Auto remembers a fallback per URL for 10 minutes across sessions, so a resume or a rebuilt grid does not try UDP again at once. --- .../java/com/openipc/decoder/RtspSession.java | 66 +++++++++++++++---- .../com/openipc/decoder/RtspSessionTest.java | 47 ++++++++++++- .../com/openipc/decoder/TransportTest.java | 13 ++++ 3 files changed, 113 insertions(+), 13 deletions(-) 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 b44701e7..a4477320 100644 --- a/android/app/src/main/java/com/openipc/decoder/RtspSession.java +++ b/android/app/src/main/java/com/openipc/decoder/RtspSession.java @@ -23,6 +23,7 @@ import java.util.List; import java.util.Locale; import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicReference; /** @@ -69,8 +70,12 @@ interface Callback { private volatile String userAgent = ""; private volatile boolean withAudio; private volatile Transport.Mode mode = Transport.Mode.TCP; - /** Auto transport: UDP failed for this URL, stay on TCP until reconfigured. */ - private volatile boolean udpFailed; + /** + * Auto transport: URLs whose UDP failed, with the time, kept for the process so that + * a resume or a rebuilt grid does not try UDP again at once (#67). + */ + static final Map udpFailedAt = new ConcurrentHashMap<>(); + static volatile long udpRetryMs = 10 * 60 * 1000; // Incremented by start()/stop(); threads and callbacks of an older generation do nothing private volatile int generation; @@ -105,7 +110,18 @@ void configure(String url, String userAgent, boolean withAudio, Transport.Mode m this.userAgent = userAgent != null ? userAgent : ""; this.withAudio = withAudio; this.mode = mode != null ? mode : Transport.Mode.TCP; - this.udpFailed = false; + } + + private static boolean udpFailed(String url) { + Long at = udpFailedAt.get(url); + if (at == null) return false; + if (System.nanoTime() / 1_000_000 - at < udpRetryMs) return true; + udpFailedAt.remove(url); + return false; + } + + private void markUdpFailed() { + udpFailedAt.put(url, System.nanoTime() / 1_000_000); } boolean isActive() { return active; } @@ -262,7 +278,7 @@ private boolean connect(int gen) throws IOException { Transport.Mode m = mode; boolean auto = m == Transport.Mode.AUTO; - if (m == Transport.Mode.UDP || (auto && !udpFailed)) { + if (m == Transport.Mode.UDP || (auto && !udpFailed(url))) { rx = new UdpReceiver(tag); synchronized (lock) { if (gen != generation) return false; @@ -270,14 +286,15 @@ private boolean connect(int gen) throws IOException { } } + int videoPort = rx != null ? rx.addTrack(CHANNEL_VIDEO) : -1; String transport = rx != null - ? Transport.udpRequest(rx.addTrack(CHANNEL_VIDEO)) + ? Transport.udpRequest(videoPort) : Transport.tcpRequest(CHANNEL_VIDEO); r = exchange(c, "SETUP", sdp.videoUrl, "Transport: " + transport + "\r\n"); if (rx != null && auto && !r.ok()) { // 461 Unsupported Transport and the like: the server does not do UDP Log.i(TAG, tag + ": UDP SETUP failed (" + r.code + "), using TCP"); - udpFailed = true; + markUdpFailed(); rx = dropUdp(rx); r = exchange(c, "SETUP", sdp.videoUrl, "Transport: " + Transport.tcpRequest(CHANNEL_VIDEO) + "\r\n"); @@ -292,21 +309,36 @@ private boolean connect(int gen) throws IOException { // Asked for UDP, answered with interleaved TCP: take what the server chose Log.i(TAG, tag + ": server chose TCP instead of UDP"); rx = dropUdp(rx); + } else if (rx != null && !udpReplyUsable(reply, videoPort)) { + // Multicast (#69) or other ports than ours: nothing would reach the receiver + if (auto) markUdpFailed(); + throw new IOException("Unusable UDP transport in SETUP reply: " + + r.header("transport")); } + InetAddress audioSource = null; boolean audio = withAudio && sdp.audioUrl != null && !AudioConfig.NONE.equals(sdp.audio.codec); if (audio) { + int audioPort = rx != null ? rx.addTrack(CHANNEL_AUDIO) : -1; transport = rx != null - ? Transport.udpRequest(rx.addTrack(CHANNEL_AUDIO)) + ? Transport.udpRequest(audioPort) : Transport.tcpRequest(CHANNEL_AUDIO); r = exchange(c, "SETUP", sdp.audioUrl, "Transport: " + transport + "\r\n" + "Session: " + session + "\r\n"); + Transport audioReply = Transport.parse(r.header("transport")); if (!r.ok()) { // Not fatal: play video without audio Log.w(TAG, tag + ": audio SETUP failed (" + r.code + "), no audio"); audio = false; + } else if (rx != null && (audioReply.tcp + || !udpReplyUsable(audioReply, audioPort))) { + // Video over UDP, audio elsewhere: the UDP loop cannot play it + Log.w(TAG, tag + ": audio not over UDP like video, no audio"); + audio = false; + } else if (rx != null) { + audioSource = sourceAddress(audioReply.source); } } if (!audio) sdp.audio = new AudioConfig(); @@ -314,7 +346,7 @@ private boolean connect(int gen) throws IOException { if (rx != null) { // Only the camera may send to our ports; "source" may be its own view of // its address (NAT), so the address we connected to is accepted as well - rx.setSources(s.getInetAddress(), sourceAddress(reply.source)); + rx.setSources(s.getInetAddress(), sourceAddress(reply.source), audioSource); } if (gen != generation) return false; @@ -345,8 +377,11 @@ private boolean connect(int gen) throws IOException { } finally { synchronized (lock) { if (socket == s) socket = null; - if (udp == rx) udp = null; - udpPlaying = false; + // A newer connection may own these by now + if (udp == rx) { + udp = null; + udpPlaying = false; + } Keepalive k = keepalive; if (k != null && k.conn == c) keepalive = null; } @@ -366,6 +401,15 @@ private boolean connect(int gen) throws IOException { } } + /** + * A UDP reply we can receive: unicast, and the client ports, when given, are ours. + * A missing Transport header is taken as agreement. + */ + static boolean udpReplyUsable(Transport reply, int rtpPort) { + if (reply.multicast) return false; + return reply.clientPort == null || reply.clientPort[0] == rtpPort; + } + private UdpReceiver dropUdp(UdpReceiver rx) { rx.close(); synchronized (lock) { @@ -442,7 +486,7 @@ private void receiveUdp(int gen, Conn c, UdpReceiver rx, boolean auto) throws IO if (auto && first[0] && System.nanoTime() / 1_000_000 - playMs > udpFirstPacketMs) { Log.w(TAG, tag + ": no video over UDP, using TCP"); - udpFailed = true; + markUdpFailed(); return false; } return true; diff --git a/android/app/src/test/java/com/openipc/decoder/RtspSessionTest.java b/android/app/src/test/java/com/openipc/decoder/RtspSessionTest.java index 05c99b4e..d706576a 100644 --- a/android/app/src/test/java/com/openipc/decoder/RtspSessionTest.java +++ b/android/app/src/test/java/com/openipc/decoder/RtspSessionTest.java @@ -82,6 +82,8 @@ private static final class FakeServer implements AutoCloseable { volatile boolean udpSilent; /** Send the datagrams from this local address instead of 127.0.0.1. */ volatile String udpFrom = "127.0.0.1"; + /** Answer a UDP SETUP with this Transport value instead of the client's ports. */ + volatile String udpReply; FakeServer() throws IOException { server = new ServerSocket(0, 50, InetAddress.getLoopbackAddress()); @@ -155,9 +157,10 @@ private void serve(Socket s, int n) { if (asked.clientPort != null) { if (udpUnsupported) code = 461; if (code == 200) udpPorts[audioTrack ? 1 : 0] = asked.clientPort[0]; - transport = "Transport: RTP/AVP/UDP;unicast;client_port=" + transport = "Transport: " + (udpReply != null ? udpReply + : "RTP/AVP/UDP;unicast;client_port=" + asked.clientPort[0] + "-" + asked.clientPort[1] - + ";server_port=6970-6971\r\n"; + + ";server_port=6970-6971") + "\r\n"; } else if (asked.interleaved != null) { transport = "Transport: RTP/AVP/TCP;unicast;interleaved=" + asked.interleaved[0] + "-" + asked.interleaved[1] @@ -338,6 +341,7 @@ public void setUp() throws IOException { RtspSession.watchdogMs = 8000; server = new FakeServer(); session = new RtspSession("test", rec); + RtspSession.udpFailedAt.clear(); } @After @@ -675,4 +679,43 @@ public void datagramsFromAnotherAddressAreIgnored() throws Exception { private String base() { return "rtsp://127.0.0.1:" + server.server.getLocalPort() + "/stream=0"; } + + @Test + public void autoRemembersTheFallbackForTheUrl() throws Exception { + server.udpUnsupported = true; + runUdp(Transport.Mode.AUTO, false); + assertTrue(rec.videoLatch.await(5, TimeUnit.SECONDS)); + session.stop(); + + // A resumed view or a rebuilt grid cell goes straight to TCP + server.requests.clear(); + RtspSession again = new RtspSession("again", new Recorder()); + again.configure(server.url(null), "User-Agent: test\r\n", false, Transport.Mode.AUTO); + again.start(); + try { + long deadline = System.currentTimeMillis() + 3000; + while (setups().isEmpty() && System.currentTimeMillis() < deadline) Thread.sleep(10); + assertTrue(setups().get(0).contains("interleaved=0-1")); + } finally { + again.stop(); + } + } + + @Test + public void autoFallsBackOnAMulticastReply() throws Exception { + server.udpReply = "RTP/AVP;multicast;destination=239.1.2.3;port=5000-5001;ttl=4"; + runUdp(Transport.Mode.AUTO, false); + assertTrue("video over TCP", rec.videoLatch.await(5, TimeUnit.SECONDS)); + List s = setups(); + assertTrue(s.get(s.size() - 1).contains("interleaved=0-1")); + } + + @Test + public void autoFallsBackWhenTheReplyNamesOtherPorts() throws Exception { + server.udpReply = "RTP/AVP/UDP;unicast;client_port=1000-1001;server_port=6970-6971"; + runUdp(Transport.Mode.AUTO, false); + assertTrue("video over TCP", rec.videoLatch.await(5, TimeUnit.SECONDS)); + List s = setups(); + assertTrue(s.get(s.size() - 1).contains("interleaved=0-1")); + } } diff --git a/android/app/src/test/java/com/openipc/decoder/TransportTest.java b/android/app/src/test/java/com/openipc/decoder/TransportTest.java index 72aeb130..60adadb2 100644 --- a/android/app/src/test/java/com/openipc/decoder/TransportTest.java +++ b/android/app/src/test/java/com/openipc/decoder/TransportTest.java @@ -96,4 +96,17 @@ public void sourceIsOnlyALiteralAddress() { assertNull(RtspSession.sourceAddress("camera.example")); assertNull(RtspSession.sourceAddress(null)); } + + @Test + public void usableUdpReplies() { + assertTrue(RtspSession.udpReplyUsable(Transport.parse( + "RTP/AVP/UDP;unicast;client_port=50000-50001;server_port=1-2"), 50000)); + // No ports or no header at all: taken as agreement + assertTrue(RtspSession.udpReplyUsable(Transport.parse("RTP/AVP;unicast"), 50000)); + assertTrue(RtspSession.udpReplyUsable(Transport.parse(null), 50000)); + assertFalse(RtspSession.udpReplyUsable(Transport.parse( + "RTP/AVP;unicast;client_port=4000-4001"), 50000)); + assertFalse(RtspSession.udpReplyUsable(Transport.parse( + "RTP/AVP;multicast;destination=239.1.2.3;port=5000-5001"), 50000)); + } }