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 e06f9a93..6560a2a0 100644 --- a/android/app/src/main/java/com/openipc/decoder/MenuManager.java +++ b/android/app/src/main/java/com/openipc/decoder/MenuManager.java @@ -354,11 +354,11 @@ void showUrlEditor(int mActive, String[] mHosts, Transport.Mode[] mTransports) { input.setSelectAllOnFocus(true); input.setSelection(input.getText().length()); - // RTP transport of the slot (#67): Auto tries UDP and falls back to TCP + // RTP transport of the slot (#67, #69): Auto and Multicast fall back to TCP RadioGroup transport = new RadioGroup(activity); transport.setOrientation(RadioGroup.HORIZONTAL); Transport.Mode[] modes = Transport.Mode.values(); - String[] labels = {"TCP", "UDP", "Auto"}; + String[] labels = {"TCP", "UDP", "Auto", "Multicast"}; for (int i = 0; i < modes.length; i++) { RadioButton b = new RadioButton(activity); b.setId(View.generateViewId()); 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 552f1658..67596809 100644 --- a/android/app/src/main/java/com/openipc/decoder/RtspSession.java +++ b/android/app/src/main/java/com/openipc/decoder/RtspSession.java @@ -117,16 +117,18 @@ void configure(String url, String userAgent, boolean withAudio, Transport.Mode m this.mode = mode != null ? mode : Transport.Mode.TCP; } - private static boolean udpFailed(String url) { - Long at = udpFailedAt.get(url); + /** Keyed by mode too: a failed multicast says nothing about unicast UDP. */ + private static boolean udpFailed(Transport.Mode mode, String url) { + String key = mode + " " + url; + Long at = udpFailedAt.get(key); if (at == null) return false; if (System.nanoTime() / 1_000_000 - at < udpRetryMs) return true; - udpFailedAt.remove(url); + udpFailedAt.remove(key); return false; } - private void markUdpFailed() { - udpFailedAt.put(url, System.nanoTime() / 1_000_000); + private void markUdpFailed(Transport.Mode mode) { + udpFailedAt.put(mode + " " + url, System.nanoTime() / 1_000_000); } boolean isActive() { return active; } @@ -287,7 +289,14 @@ private boolean connect(int gen) throws IOException { Transport.Mode m = mode; boolean auto = m == Transport.Mode.AUTO; - if (m == Transport.Mode.UDP || (auto && !udpFailed(url))) { + boolean mc = m == Transport.Mode.MULTICAST && !udpFailed(m, url); + if (mc && !UdpReceiver.multicastSupported()) { + Log.i(TAG, tag + ": multicast needs Android 8, using TCP"); + mc = false; + } + // Auto and Multicast fall back to TCP; UDP does not + boolean fallback = auto || m == Transport.Mode.MULTICAST; + if (m == Transport.Mode.UDP || (auto && !udpFailed(m, url)) || mc) { rx = new UdpReceiver(tag); synchronized (lock) { if (gen != generation) return false; @@ -295,15 +304,16 @@ private boolean connect(int gen) throws IOException { } } - int videoPort = rx != null ? rx.addTrack(CHANNEL_VIDEO) : -1; - String transport = rx != null - ? Transport.udpRequest(videoPort) - : Transport.tcpRequest(CHANNEL_VIDEO); + int videoPort = rx != null && !mc ? rx.addTrack(CHANNEL_VIDEO) : -1; + String transport = rx == null ? Transport.tcpRequest(CHANNEL_VIDEO) + : mc ? Transport.multicastRequest() : Transport.udpRequest(videoPort); 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"); - markUdpFailed(); + if (rx != null && fallback && !r.ok()) { + // 461 Unsupported Transport and the like: Majestic answers a multicast + // request with 400, XM cameras with 451 + Log.i(TAG, tag + ": " + (mc ? "multicast" : "UDP") + " SETUP failed (" + + r.code + "), using TCP"); + markUdpFailed(m); rx = dropUdp(rx); r = exchange(c, "SETUP", sdp.videoUrl, "Transport: " + Transport.tcpRequest(CHANNEL_VIDEO) + "\r\n"); @@ -318,9 +328,17 @@ 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 && mc) { + // A server may also answer with unicast (both Majestic and XM do when the + // request carries client_port): not what was asked, use TCP next time + if (!joinGroup(rx, CHANNEL_VIDEO, reply, s.getLocalAddress())) { + markUdpFailed(m); + throw new IOException("No usable multicast group in SETUP reply: " + + r.header("transport")); + } } else if (rx != null && !udpReplyUsable(reply, videoPort)) { - // Multicast (#69) or other ports than ours: nothing would reach the receiver - if (auto) markUdpFailed(); + // Multicast or other ports than ours: nothing would reach the receiver + if (auto) markUdpFailed(m); throw new IOException("Unusable UDP transport in SETUP reply: " + r.header("transport")); } @@ -329,10 +347,9 @@ private boolean connect(int gen) throws IOException { 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(audioPort) - : Transport.tcpRequest(CHANNEL_AUDIO); + int audioPort = rx != null && !mc ? rx.addTrack(CHANNEL_AUDIO) : -1; + transport = rx == null ? Transport.tcpRequest(CHANNEL_AUDIO) + : mc ? Transport.multicastRequest() : Transport.udpRequest(audioPort); r = exchange(c, "SETUP", sdp.audioUrl, "Transport: " + transport + "\r\n" + "Session: " + session + "\r\n"); @@ -341,6 +358,13 @@ private boolean connect(int gen) throws IOException { // Not fatal: play video without audio Log.w(TAG, tag + ": audio SETUP failed (" + r.code + "), no audio"); audio = false; + } else if (rx != null && mc) { + if (!joinGroup(rx, CHANNEL_AUDIO, audioReply, s.getLocalAddress())) { + Log.w(TAG, tag + ": no usable multicast group for audio, no audio"); + audio = false; + } else { + audioSource = sourceAddress(audioReply.source); + } } else if (rx != null && (audioReply.tcp || !udpReplyUsable(audioReply, audioPort))) { // Video over UDP, audio elsewhere: the UDP loop cannot play it @@ -355,7 +379,13 @@ 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), audioSource); + if (mc) { + // The group already selects the stream, and a server may send it from + // another interface than the RTSP one: only a named source restricts it + rx.setSources(sourceAddress(reply.source), audioSource); + } else { + rx.setSources(s.getInetAddress(), sourceAddress(reply.source), audioSource); + } } if (gen != generation) return false; @@ -374,11 +404,11 @@ private boolean connect(int gen) throws IOException { keepalive = new Keepalive(c, u.requestUrl, session, keepaliveMs); udpPlaying = rx != null; } - Log.i(TAG, tag + ": playing over " + (rx != null ? "UDP" : "TCP")); + Log.i(TAG, tag + ": playing over " + (rx == null ? "TCP" : mc ? "multicast" : "UDP")); status(gen, "buffering"); s.setSoTimeout(0); // the watchdog handles silence if (rx != null) { - receiveUdp(gen, c, rx, auto); + receiveUdp(gen, c, rx, fallback, m); } else { receive(gen, c); } @@ -484,6 +514,27 @@ private boolean listen(int gen, RtpListener.Url u) throws IOException { } } + /** + * Join the group of a multicast SETUP reply (#69): a literal multicast destination and + * a port are required. False when the reply is not usable or the join fails. + */ + private boolean joinGroup(UdpReceiver rx, int channel, Transport reply, InetAddress local) { + InetAddress group = sourceAddress(reply.destination); + if (!reply.multicast || group == null || !group.isMulticastAddress() + || reply.port == null || reply.port[0] <= 0) { + return false; + } + try { + rx.addMulticastTrack(channel, group, reply.port[0], local); + Log.i(TAG, tag + ": joined " + group.getHostAddress() + ":" + reply.port[0] + + (reply.ttl >= 0 ? ", ttl " + reply.ttl : "")); + return true; + } catch (IOException e) { + Log.w(TAG, tag + ": cannot join " + group.getHostAddress() + ": " + e.getMessage()); + return false; + } + } + /** * A UDP reply we can receive: unicast, and the client ports, when given, are ours. * A missing Transport header is taken as agreement. @@ -545,7 +596,8 @@ private void receive(int gen, Conn c) throws IOException { * (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 { + private void receiveUdp(int gen, Conn c, UdpReceiver rx, boolean auto, Transport.Mode m) + throws IOException { AtomicReference controlError = new AtomicReference<>(); Thread control = new Thread(() -> { try { @@ -568,8 +620,9 @@ private void receiveUdp(int gen, Conn c, UdpReceiver rx, boolean auto) throws IO 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"); - markUdpFailed(); + Log.w(TAG, tag + ": no video over " + (m == Transport.Mode.MULTICAST + ? "multicast" : "UDP") + ", using TCP"); + markUdpFailed(m); return false; } return true; diff --git a/android/app/src/main/java/com/openipc/decoder/Transport.java b/android/app/src/main/java/com/openipc/decoder/Transport.java index beb41418..64f44ef0 100644 --- a/android/app/src/main/java/com/openipc/decoder/Transport.java +++ b/android/app/src/main/java/com/openipc/decoder/Transport.java @@ -16,9 +16,12 @@ * 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. */ + /** + * Slot setting: RTP over the RTSP connection, over UDP, UDP with fallback to TCP, or + * RTSP multicast with fallback to TCP (#69). + */ enum Mode { - TCP, UDP, AUTO; + TCP, UDP, AUTO, MULTICAST; /** Stored or shared value ("tcp", "udp", "auto"); unknown values mean TCP. */ static Mode parse(String s) { @@ -104,6 +107,11 @@ static String tcpRequest(int ch) { return "RTP/AVP/TCP;unicast;interleaved=" + ch + "-" + (ch + 1); } + /** Request header value for multicast; the server picks group, ports and TTL. */ + static String multicastRequest() { + return "RTP/AVP;multicast"; + } + /** 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 index 54529cc8..86b40747 100644 --- a/android/app/src/main/java/com/openipc/decoder/UdpReceiver.java +++ b/android/app/src/main/java/com/openipc/decoder/UdpReceiver.java @@ -7,13 +7,19 @@ package com.openipc.decoder; +import android.annotation.TargetApi; +import android.os.Build; import android.util.Log; import java.io.Closeable; import java.io.IOException; +import java.net.Inet6Address; import java.net.InetAddress; import java.net.InetSocketAddress; +import java.net.NetworkInterface; import java.net.SocketAddress; +import java.net.StandardProtocolFamily; +import java.net.StandardSocketOptions; import java.nio.ByteBuffer; import java.nio.channels.ClosedSelectorException; import java.nio.channels.DatagramChannel; @@ -49,6 +55,7 @@ interface Condition { private static final class Track { final int channel; final DatagramChannel rtp; + /** Null for a multicast track: its RTCP is not received. */ 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. */ @@ -69,6 +76,7 @@ private static final class Track { private volatile boolean closed; private boolean running; private boolean released; + private boolean multicastLocked; private long rejected; UdpReceiver(String tag) throws IOException { @@ -113,6 +121,51 @@ synchronized int addTrack(int channel) throws IOException { throw new IOException("No free UDP port pair"); } + /** Joining a group from a DatagramChannel needs API 26 (Android 8). Tests override it. */ + static volatile int sdkInt = Build.VERSION.SDK_INT; + + static boolean multicastSupported() { + return sdkInt >= 26; + } + + /** + * Join {@code group} on {@code port} for {@code channel}, on the interface that owns + * {@code local} (the local address of the RTSP connection, #69). Only when + * {@link #multicastSupported()}. + */ + @TargetApi(26) + synchronized void addMulticastTrack(int channel, InetAddress group, int port, + InetAddress local) throws IOException { + if (closed) throw new IOException("Receiver closed"); + if (!group.isMulticastAddress()) throw new IOException("Not a multicast group: " + group); + if (port <= 0 || port > 65535) throw new IOException("Invalid multicast port: " + port); + NetworkInterface nif = NetworkInterface.getByInetAddress(local); + if (nif == null) throw new IOException("No network interface for " + local); + DatagramChannel ch = DatagramChannel.open(group instanceof Inet6Address + ? StandardProtocolFamily.INET6 : StandardProtocolFamily.INET); + try { + // Other receivers of the group on this device bind the same port. 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 + ch.setOption(StandardSocketOptions.SO_REUSEADDR, true); + ch.bind(new InetSocketAddress(group, port)); + ch.setOption(StandardSocketOptions.SO_RCVBUF, + channel == RtspSession.CHANNEL_VIDEO ? VIDEO_RCVBUF : AUDIO_RCVBUF); + ch.join(group, nif); + ch.configureBlocking(false); + Track t = new Track(channel, ch, null); + ch.register(selector, SelectionKey.OP_READ, t); + tracks.add(t); + } catch (IOException e) { + ch.close(); + throw e; + } + if (!multicastLocked) { + multicastLocked = true; + RtpListener.MulticastLock.acquire(); + } + } + /** Accept datagrams only from these addresses (nulls are ignored). */ synchronized void setSources(InetAddress... addresses) { sources.clear(); @@ -206,9 +259,13 @@ private synchronized void release() { closed = true; for (Track t : tracks) { closeQuietly(t.rtp); - closeQuietly(t.rtcp); + if (t.rtcp != null) closeQuietly(t.rtcp); } closeQuietly(selector); + if (multicastLocked) { + multicastLocked = false; + RtpListener.MulticastLock.release(); + } } private static void closeQuietly(Closeable c) { 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 d706576a..3ee7e058 100644 --- a/android/app/src/test/java/com/openipc/decoder/RtspSessionTest.java +++ b/android/app/src/test/java/com/openipc/decoder/RtspSessionTest.java @@ -84,6 +84,11 @@ private static final class FakeServer implements AutoCloseable { volatile String udpFrom = "127.0.0.1"; /** Answer a UDP SETUP with this Transport value instead of the client's ports. */ volatile String udpReply; + /** + * Answer a multicast SETUP with this Transport value; null answers 400 as Majestic + * does. A "destination=group;port=n" reply streams to that group after PLAY. + */ + volatile String multicastReply; FakeServer() throws IOException { server = new ServerSocket(0, 50, InetAddress.getLoopbackAddress()); @@ -118,6 +123,8 @@ private void serve(Socket s, int n) { OutputStream out = socket.getOutputStream(); // Client RTP ports of this connection's UDP SETUPs: video, audio int[] udpPorts = {-1, -1}; + // Multicast group and port of this connection's SETUP, if any + String[] mcast = {null, null}; while (true) { String line = RtspSession.readLine(in); if (line == null) return; @@ -154,7 +161,18 @@ private void serve(Socket s, int n) { int code = audioTrack ? audioSetupStatus : 200; Transport asked = Transport.parse(header(headers, "Transport")); String transport = ""; - if (asked.clientPort != null) { + if (asked.multicast && asked.clientPort == null) { + if (multicastReply == null) { + code = 400; + } else { + Transport t = Transport.parse(multicastReply); + if (t.multicast && t.port != null) { + mcast[0] = t.destination; + mcast[1] = String.valueOf(t.port[0]); + } + transport = "Transport: " + multicastReply + "\r\n"; + } + } else if (asked.clientPort != null) { if (udpUnsupported) code = 461; if (code == 200) udpPorts[audioTrack ? 1 : 0] = asked.clientPort[0]; transport = "Transport: " + (udpReply != null ? udpReply @@ -185,6 +203,10 @@ private void serve(Socket s, int n) { break; } write(out, reply); + if (method.equals("PLAY") && mcast[0] != null) { + sendMulticast(mcast[0], Integer.parseInt(mcast[1])); + continue; + } if (method.equals("PLAY") && udpPorts[0] > 0) { if (!udpSilent) sendUdp(udpPorts); continue; @@ -240,6 +262,25 @@ private void sendUdp(int[] ports) { sender.start(); } + private void sendMulticast(String group, int port) { + Thread sender = new Thread(() -> { + try (java.net.MulticastSocket ms = new java.net.MulticastSocket()) { + ms.setNetworkInterface(java.net.NetworkInterface.getByInetAddress( + InetAddress.getLoopbackAddress())); + ms.setTimeToLive(1); + InetAddress to = InetAddress.getByName(group); + int seq = 100; + for (int i = 0; i < videoPackets; i++) { + byte[] p = udpPacket(96, seq++); + ms.send(new DatagramPacket(p, p.length, to, port)); + } + } 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)) { @@ -342,6 +383,7 @@ public void setUp() throws IOException { server = new FakeServer(); session = new RtspSession("test", rec); RtspSession.udpFailedAt.clear(); + UdpReceiver.sdkInt = 29; } @After @@ -352,6 +394,7 @@ public void tearDown() throws IOException { RtspSession.maxRetryMs = 30000; RtspSession.watchdogMs = 8000; RtspSession.udpFirstPacketMs = 3000; + UdpReceiver.sdkInt = android.os.Build.VERSION.SDK_INT; } private void runUntilVideo(String url, boolean withAudio) throws InterruptedException { @@ -718,4 +761,86 @@ public void autoFallsBackWhenTheReplyNamesOtherPorts() throws Exception { List s = setups(); assertTrue(s.get(s.size() - 1).contains("interleaved=0-1")); } + + // ====================== Multicast (#69) ====================== + + @Test + public void multicastRefusedLikeMajesticFallsBackToTcp() throws Exception { + runUdp(Transport.Mode.MULTICAST, false); + assertTrue("video over TCP", rec.videoLatch.await(5, TimeUnit.SECONDS)); + List s = setups(); + assertTrue(s.get(0).contains("Transport: RTP/AVP;multicast")); + assertTrue(s.get(1).contains("interleaved=0-1")); + assertEquals(1, server.connections.get()); + } + + @Test + public void multicastAnsweredWithUnicastFallsBackToTcp() throws Exception { + server.multicastReply = "RTP/AVP/UDP;unicast;client_port=50020-50021;server_port=1-2"; + runUdp(Transport.Mode.MULTICAST, false); + assertTrue("video over TCP", rec.videoLatch.await(5, TimeUnit.SECONDS)); + List s = setups(); + assertTrue(s.get(s.size() - 1).contains("interleaved=0-1")); + assertEquals(2, server.connections.get()); + } + + @Test + public void multicastReplyWithoutAPortFallsBackToTcp() throws Exception { + server.multicastReply = "RTP/AVP;multicast;destination=239.255.42.18;port=0-1;ttl=1"; + runUdp(Transport.Mode.MULTICAST, 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 multicastNeedsAndroid8() throws Exception { + UdpReceiver.sdkInt = 25; + runUdp(Transport.Mode.MULTICAST, false); + assertTrue("video over TCP", rec.videoLatch.await(5, TimeUnit.SECONDS)); + assertTrue(setups().get(0).contains("interleaved=0-1")); + } + + @Test + public void multicastGroupIsJoinedAndReceived() throws Exception { + String group = "239.255.42.17"; + int port = freeUdpPort(); + boolean works = loopbackMulticastWorks(group, port); + System.out.println("loopback multicast " + (works ? "works" : "does not work")); + org.junit.Assume.assumeTrue("loopback multicast works here", works); + server.multicastReply = "RTP/AVP;multicast;destination=" + group + ";port=" + + port + "-" + (port + 1) + ";ttl=1"; + runUdp(Transport.Mode.MULTICAST, false); + assertTrue("video over multicast", rec.videoLatch.await(5, TimeUnit.SECONDS)); + for (String r : setups()) assertFalse(r.contains("interleaved")); + assertEquals(1, server.connections.get()); + } + + private static int freeUdpPort() throws IOException { + try (DatagramSocket s = new DatagramSocket(0)) { + return s.getLocalPort() & ~1; + } + } + + /** Whether this machine delivers a group joined on the loopback interface. */ + private static boolean loopbackMulticastWorks(String group, int port) { + try (java.nio.channels.DatagramChannel ch = java.nio.channels.DatagramChannel.open( + java.net.StandardProtocolFamily.INET); + java.net.MulticastSocket tx = new java.net.MulticastSocket()) { + java.net.NetworkInterface lo = java.net.NetworkInterface.getByInetAddress( + InetAddress.getLoopbackAddress()); + ch.setOption(java.net.StandardSocketOptions.SO_REUSEADDR, true); + ch.bind(new InetSocketAddress(port)); + ch.join(InetAddress.getByName(group), lo); + ch.socket().setSoTimeout(500); + tx.setNetworkInterface(lo); + byte[] p = {1, 2, 3}; + tx.send(new DatagramPacket(p, p.length, InetAddress.getByName(group), port)); + byte[] in = new byte[16]; + ch.socket().receive(new DatagramPacket(in, in.length)); + return true; + } catch (Exception e) { + return false; + } + } } 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 60adadb2..7671a5dc 100644 --- a/android/app/src/test/java/com/openipc/decoder/TransportTest.java +++ b/android/app/src/test/java/com/openipc/decoder/TransportTest.java @@ -78,6 +78,7 @@ public void malformedValuesAreIgnored() { 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)); + assertEquals("RTP/AVP;multicast", Transport.multicastRequest()); } @Test @@ -87,6 +88,7 @@ public void modeFromSettings() { assertEquals(Transport.Mode.TCP, Transport.Mode.parse("TCP")); assertEquals(Transport.Mode.TCP, Transport.Mode.parse(null)); assertEquals(Transport.Mode.TCP, Transport.Mode.parse("quic")); + assertEquals(Transport.Mode.MULTICAST, Transport.Mode.parse("multicast")); } @Test