feat: RTSP multicast, with fallback to TCP (#69)

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());

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,23 +289,31 @@ 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;
                     udp = rx;
                 }
             }
 
-            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<IOException> 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;

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);

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) {