Add listener for data channel messages

For now only the same data channel messages that were already handled
are taken into account, but at a later point the missing messages
("speaking" and "stoppedSpeaking") could be added too.

Note that the thread used to handle the data channel messages has
changed; the EventBus subscriber mode was "MAIN", but as the messages
were posted from a DataChannel observer, which run in a worker thread
rather than in the main thread, the subscriber was executed in the main
thread rather than in the same thread as the poster. Due to this the
actions performed by the handler now must be explicitly run in the main
thread.

Signed-off-by: Daniel Calviño Sánchez <danxuliu@gmail.com>

app/src/main/java/com/nextcloud/talk/activities/CallActivity.java

@@ -268,6 +268,8 @@ public class CallActivity extends CallBaseActivity {
     private Map<String, SignalingMessageReceiver.CallParticipantMessageListener> callParticipantMessageListeners =
         new HashMap<>();
 
+    private Map<String, PeerConnectionWrapper.DataChannelMessageListener> dataChannelMessageListeners = new HashMap<>();
+
     private SignalingMessageReceiver.ParticipantListMessageListener participantListMessageListener = new SignalingMessageReceiver.ParticipantListMessageListener() {
 
         @Override
@@ -2007,6 +2009,12 @@ public class CallActivity extends CallBaseActivity {
                     new CallActivityCallParticipantMessageListener(sessionId);
                 callParticipantMessageListeners.put(sessionId, callParticipantMessageListener);
                 signalingMessageReceiver.addListener(callParticipantMessageListener, sessionId);
+
+                // DataChannel messages are sent only in video peers; (sender) screen peers do not even open them.
+                PeerConnectionWrapper.DataChannelMessageListener dataChannelMessageListener =
+                    new CallActivityDataChannelMessageListener(sessionId);
+                dataChannelMessageListeners.put(sessionId, dataChannelMessageListener);
+                peerConnectionWrapper.addListener(dataChannelMessageListener);
             }
 
             if (!publisher && !hasExternalSignalingServer && offerAnswerNickProviders.get(sessionId) == null) {
@@ -2040,6 +2048,10 @@ public class CallActivity extends CallBaseActivity {
         if (!(peerConnectionWrappers = getPeerConnectionWrapperListForSessionId(sessionId)).isEmpty()) {
             for (PeerConnectionWrapper peerConnectionWrapper : peerConnectionWrappers) {
                 if (peerConnectionWrapper.getSessionId().equals(sessionId)) {
+                    if (!justScreen && VIDEO_STREAM_TYPE_VIDEO.equals(peerConnectionWrapper.getVideoStreamType())) {
+                        PeerConnectionWrapper.DataChannelMessageListener dataChannelMessageListener = dataChannelMessageListeners.remove(sessionId);
+                        peerConnectionWrapper.removeListener(dataChannelMessageListener);
+                    }
                     String videoStreamType = peerConnectionWrapper.getVideoStreamType();
                     if (VIDEO_STREAM_TYPE_SCREEN.equals(videoStreamType) || !justScreen) {
                         runOnUiThread(() -> removeMediaStream(sessionId, videoStreamType));
@@ -2163,24 +2175,6 @@ public class CallActivity extends CallBaseActivity {
                     toggleMedia(enableVideo, true);
                 }
             }
-        } else if (peerConnectionEvent.getPeerConnectionEventType() ==
-            PeerConnectionEvent.PeerConnectionEventType.NICK_CHANGE) {
-            if (participantDisplayItems.get(participantDisplayItemId) != null) {
-                participantDisplayItems.get(participantDisplayItemId).setNick(peerConnectionEvent.getNick());
-                participantsAdapter.notifyDataSetChanged();
-            }
-        } else if (peerConnectionEvent.getPeerConnectionEventType() ==
-            PeerConnectionEvent.PeerConnectionEventType.VIDEO_CHANGE) {
-            if (participantDisplayItems.get(participantDisplayItemId) != null) {
-                participantDisplayItems.get(participantDisplayItemId).setStreamEnabled(peerConnectionEvent.getChangeValue());
-                participantsAdapter.notifyDataSetChanged();
-            }
-        } else if (peerConnectionEvent.getPeerConnectionEventType() ==
-            PeerConnectionEvent.PeerConnectionEventType.AUDIO_CHANGE) {
-            if (participantDisplayItems.get(participantDisplayItemId) != null) {
-                participantDisplayItems.get(participantDisplayItemId).setAudioEnabled(peerConnectionEvent.getChangeValue());
-                participantsAdapter.notifyDataSetChanged();
-            }
         } else if (peerConnectionEvent.getPeerConnectionEventType() ==
             PeerConnectionEvent.PeerConnectionEventType.PUBLISHER_FAILED) {
             setCallState(CallStatus.PUBLISHER_FAILED);
@@ -2631,6 +2625,66 @@ public class CallActivity extends CallBaseActivity {
         }
     }
 
+    private class CallActivityDataChannelMessageListener implements PeerConnectionWrapper.DataChannelMessageListener {
+
+        private final String participantDisplayItemId;
+
+        private CallActivityDataChannelMessageListener(String sessionId) {
+            // DataChannel messages are sent only in video peers, so the listener only acts on the "video" items.
+            this.participantDisplayItemId = sessionId + "-video";
+        }
+
+        @Override
+        public void onAudioOn() {
+            runOnUiThread(() -> {
+                if (participantDisplayItems.get(participantDisplayItemId) != null) {
+                    participantDisplayItems.get(participantDisplayItemId).setAudioEnabled(true);
+                    participantsAdapter.notifyDataSetChanged();
+                }
+            });
+        }
+
+        @Override
+        public void onAudioOff() {
+            runOnUiThread(() -> {
+                if (participantDisplayItems.get(participantDisplayItemId) != null) {
+                    participantDisplayItems.get(participantDisplayItemId).setAudioEnabled(false);
+                    participantsAdapter.notifyDataSetChanged();
+                }
+            });
+        }
+
+        @Override
+        public void onVideoOn() {
+            runOnUiThread(() -> {
+                if (participantDisplayItems.get(participantDisplayItemId) != null) {
+                    participantDisplayItems.get(participantDisplayItemId).setStreamEnabled(true);
+                    participantsAdapter.notifyDataSetChanged();
+                }
+            });
+        }
+
+        @Override
+        public void onVideoOff() {
+            runOnUiThread(() -> {
+                if (participantDisplayItems.get(participantDisplayItemId) != null) {
+                    participantDisplayItems.get(participantDisplayItemId).setStreamEnabled(false);
+                    participantsAdapter.notifyDataSetChanged();
+                }
+            });
+        }
+
+        @Override
+        public void onNickChanged(String nick) {
+            runOnUiThread(() -> {
+                if (participantDisplayItems.get(participantDisplayItemId) != null) {
+                    participantDisplayItems.get(participantDisplayItemId).setNick(nick);
+                    participantsAdapter.notifyDataSetChanged();
+                }
+            });
+        }
+    }
+
     private class InternalSignalingMessageSender implements SignalingMessageSender {
 
         @Override

app/src/main/java/com/nextcloud/talk/events/PeerConnectionEvent.java

@@ -120,6 +120,6 @@ public class PeerConnectionEvent {
     }
 
     public enum PeerConnectionEventType {
-        PEER_CONNECTED, PEER_DISCONNECTED, PEER_CLOSED, SENSOR_FAR, SENSOR_NEAR, NICK_CHANGE, AUDIO_CHANGE, VIDEO_CHANGE, PUBLISHER_FAILED
+        PEER_CONNECTED, PEER_DISCONNECTED, PEER_CLOSED, SENSOR_FAR, SENSOR_NEAR, PUBLISHER_FAILED
     }
 }

app/src/main/java/com/nextcloud/talk/webrtc/DataChannelMessageNotifier.java

@@ -0,0 +1,77 @@
+/*
+ * Nextcloud Talk application
+ *
+ * @author Daniel Calviño Sánchez
+ * Copyright (C) 2022 Daniel Calviño Sánchez <danxuliu@gmail.com>
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package com.nextcloud.talk.webrtc;
+
+import java.util.ArrayList;
+import java.util.LinkedHashSet;
+import java.util.Set;
+
+/**
+ * Helper class to register and notify DataChannelMessageListeners.
+ *
+ * This class is only meant for internal use by PeerConnectionWrapper; listeners must register themselves against
+ * a PeerConnectionWrapper rather than against a DataChannelMessageNotifier.
+ */
+public class DataChannelMessageNotifier {
+
+    private final Set<PeerConnectionWrapper.DataChannelMessageListener> dataChannelMessageListeners = new LinkedHashSet<>();
+
+    public synchronized void addListener(PeerConnectionWrapper.DataChannelMessageListener listener) {
+        if (listener == null) {
+            throw new IllegalArgumentException("DataChannelMessageListener can not be null");
+        }
+
+        dataChannelMessageListeners.add(listener);
+    }
+
+    public synchronized void removeListener(PeerConnectionWrapper.DataChannelMessageListener listener) {
+        dataChannelMessageListeners.remove(listener);
+    }
+
+    public synchronized void notifyAudioOn() {
+        for (PeerConnectionWrapper.DataChannelMessageListener listener : new ArrayList<>(dataChannelMessageListeners)) {
+            listener.onAudioOn();
+        }
+    }
+
+    public synchronized void notifyAudioOff() {
+        for (PeerConnectionWrapper.DataChannelMessageListener listener : new ArrayList<>(dataChannelMessageListeners)) {
+            listener.onAudioOff();
+        }
+    }
+
+    public synchronized void notifyVideoOn() {
+        for (PeerConnectionWrapper.DataChannelMessageListener listener : new ArrayList<>(dataChannelMessageListeners)) {
+            listener.onVideoOn();
+        }
+    }
+
+    public synchronized void notifyVideoOff() {
+        for (PeerConnectionWrapper.DataChannelMessageListener listener : new ArrayList<>(dataChannelMessageListeners)) {
+            listener.onVideoOff();
+        }
+    }
+
+    public synchronized void notifyNickChanged(String nick) {
+        for (PeerConnectionWrapper.DataChannelMessageListener listener : new ArrayList<>(dataChannelMessageListeners)) {
+            listener.onNickChanged(nick);
+        }
+    }
+}

app/src/main/java/com/nextcloud/talk/webrtc/PeerConnectionWrapper.java

@@ -65,12 +65,26 @@ import javax.inject.Inject;
 import androidx.annotation.Nullable;
 import autodagger.AutoInjector;
 
-import static java.lang.Boolean.FALSE;
-import static java.lang.Boolean.TRUE;
-
 @AutoInjector(NextcloudTalkApplication.class)
 public class PeerConnectionWrapper {
 
+    /**
+     * Listener for data channel messages.
+     *
+     * The messages are bound to a specific peer connection, so each listener is expected to handle messages only for
+     * a single peer connection.
+     *
+     * All methods are called on the so called "signaling" thread of WebRTC, which is an internal thread created by the
+     * WebRTC library and NOT the same thread where signaling messages are received.
+     */
+    public interface DataChannelMessageListener {
+        void onAudioOn();
+        void onAudioOff();
+        void onVideoOn();
+        void onVideoOff();
+        void onNickChanged(String nick);
+    }
+
     private static final String TAG = PeerConnectionWrapper.class.getCanonicalName();
 
     private final SignalingMessageReceiver signalingMessageReceiver;
@@ -78,6 +92,8 @@ public class PeerConnectionWrapper {
 
     private final SignalingMessageSender signalingMessageSender;
 
+    private final DataChannelMessageNotifier dataChannelMessageNotifier = new DataChannelMessageNotifier();
+
     private List<IceCandidate> iceCandidates = new ArrayList<>();
     private PeerConnection peerConnection;
     private String sessionId;
@@ -156,6 +172,21 @@ public class PeerConnectionWrapper {
         }
     }
 
+    /**
+     * Adds a listener for data channel messages.
+     *
+     * A listener is expected to be added only once. If the same listener is added again it will be notified just once.
+     *
+     * @param listener the DataChannelMessageListener
+     */
+    public void addListener(DataChannelMessageListener listener) {
+        dataChannelMessageNotifier.addListener(listener);
+    }
+
+    public void removeListener(DataChannelMessageListener listener) {
+        dataChannelMessageNotifier.removeListener(listener);
+    }
+
     public String getVideoStreamType() {
         return videoStreamType;
     }
@@ -339,21 +370,16 @@ public class PeerConnectionWrapper {
                 }
 
                 if (nick != null) {
-                    EventBus.getDefault().post(new PeerConnectionEvent(PeerConnectionEvent.PeerConnectionEventType
+                    dataChannelMessageNotifier.notifyNickChanged(nick);
-                            .NICK_CHANGE, sessionId, nick, null, videoStreamType));
                 }
             } else if ("audioOn".equals(dataChannelMessage.getType())) {
-                EventBus.getDefault().post(new PeerConnectionEvent(PeerConnectionEvent.PeerConnectionEventType
+                dataChannelMessageNotifier.notifyAudioOn();
-                        .AUDIO_CHANGE, sessionId, null, TRUE, videoStreamType));
             } else if ("audioOff".equals(dataChannelMessage.getType())) {
-                EventBus.getDefault().post(new PeerConnectionEvent(PeerConnectionEvent.PeerConnectionEventType
+                dataChannelMessageNotifier.notifyAudioOff();
-                        .AUDIO_CHANGE, sessionId, null, FALSE, videoStreamType));
             } else if ("videoOn".equals(dataChannelMessage.getType())) {
-                EventBus.getDefault().post(new PeerConnectionEvent(PeerConnectionEvent.PeerConnectionEventType
+                dataChannelMessageNotifier.notifyVideoOn();
-                        .VIDEO_CHANGE, sessionId, null, TRUE, videoStreamType));
             } else if ("videoOff".equals(dataChannelMessage.getType())) {
-                EventBus.getDefault().post(new PeerConnectionEvent(PeerConnectionEvent.PeerConnectionEventType
+                dataChannelMessageNotifier.notifyVideoOff();
-                        .VIDEO_CHANGE, sessionId, null, FALSE, videoStreamType));
             }
         }
     }