Keep web socket open during calling to improve message delivery.

app/src/main/java/org/thoughtcrime/securesms/messages/IncomingMessageObserver.java

@@ -27,12 +27,15 @@ import org.thoughtcrime.securesms.messages.IncomingMessageProcessor.Processor;
 import org.thoughtcrime.securesms.notifications.NotificationChannels;
 import org.thoughtcrime.securesms.push.SignalServiceNetworkAccess;
 import org.thoughtcrime.securesms.util.AppForegroundObserver;
+import org.thoughtcrime.securesms.util.Util;
 import org.whispersystems.signalservice.api.SignalWebSocket;
 import org.whispersystems.signalservice.api.messages.SignalServiceEnvelope;
 import org.whispersystems.signalservice.api.websocket.WebSocketUnavailableException;
 
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.TimeUnit;
@@ -50,6 +53,7 @@ public class IncomingMessageObserver {
 
   public static final  int  FOREGROUND_ID           = 313399;
   private static final long REQUEST_TIMEOUT_MINUTES = 1;
+  private static final long OLD_REQUEST_WINDOW_MS   = TimeUnit.MINUTES.toMillis(5);
 
   private static final AtomicInteger INSTANCE_COUNT = new AtomicInteger(0);
 
@@ -57,6 +61,7 @@ public class IncomingMessageObserver {
   private final SignalServiceNetworkAccess networkAccess;
   private final List<Runnable>             decryptionDrainedListeners;
   private final BroadcastReceiver          connectionReceiver;
+  private final Map<String, Long>          keepAliveTokens;
 
   private boolean appVisible;
 
@@ -72,6 +77,7 @@ public class IncomingMessageObserver {
     this.context                    = context;
     this.networkAccess              = ApplicationDependencies.getSignalServiceNetworkAccess();
     this.decryptionDrainedListeners = new CopyOnWriteArrayList<>();
+    this.keepAliveTokens            = new HashMap<>();
 
     new MessageRetrievalThread().start();
 
@@ -155,12 +161,18 @@ public class IncomingMessageObserver {
     boolean fcmEnabled = SignalStore.account().isFcmEnabled();
     boolean hasNetwork = NetworkConstraint.isMet(context);
     boolean hasProxy   = SignalStore.proxy().isProxyEnabled();
+    long    oldRequest = System.currentTimeMillis() - OLD_REQUEST_WINDOW_MS;
 
-    Log.d(TAG, String.format("Network: %s, Foreground: %s, FCM: %s, Censored: %s, Registered: %s, Proxy: %s",
-                             hasNetwork, appVisible, fcmEnabled, networkAccess.isCensored(), registered, hasProxy));
+    boolean removedRequests = keepAliveTokens.entrySet().removeIf(e -> e.getValue() < oldRequest);
+    if (removedRequests) {
+      Log.d(TAG, "Removed old keep web socket open requests.");
+    }
+
+    Log.d(TAG, String.format("Network: %s, Foreground: %s, FCM: %s, Stay open requests: [%s], Censored: %s, Registered: %s, Proxy: %s",
+                             hasNetwork, appVisible, fcmEnabled, Util.join(keepAliveTokens.entrySet(), ","), networkAccess.isCensored(), registered, hasProxy));
 
     return registered &&
-           (appVisible || !fcmEnabled) &&
+           (appVisible || !fcmEnabled || Util.hasItems(keepAliveTokens)) &&
            hasNetwork &&
            !networkAccess.isCensored();
   }
@@ -189,6 +201,16 @@ public class IncomingMessageObserver {
     ApplicationDependencies.getSignalWebSocket().disconnect();
   }
 
+  public synchronized void registerKeepAliveToken(String key) {
+    keepAliveTokens.put(key, System.currentTimeMillis());
+    notifyAll();
+  }
+
+  public synchronized void removeKeepAliveToken(String key) {
+    keepAliveTokens.remove(key);
+    notifyAll();
+  }
+
   private class MessageRetrievalThread extends Thread implements Thread.UncaughtExceptionHandler {
 
     MessageRetrievalThread() {

app/src/main/java/org/thoughtcrime/securesms/service/webrtc/WebRtcCallService.java

@@ -14,10 +14,12 @@ import android.os.IBinder;
 import android.telephony.PhoneStateListener;
 import android.telephony.TelephonyManager;
 
+import androidx.annotation.MainThread;
 import androidx.annotation.NonNull;
 import androidx.annotation.Nullable;
 import androidx.core.content.ContextCompat;
 
+import org.signal.core.util.ThreadUtil;
 import org.signal.core.util.logging.Log;
 import org.thoughtcrime.securesms.dependencies.ApplicationDependencies;
 import org.thoughtcrime.securesms.recipients.Recipient;
@@ -31,6 +33,7 @@ import org.thoughtcrime.securesms.webrtc.locks.LockManager;
 
 import java.util.Objects;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 
 /**
  * Provide a foreground service for {@link SignalCallManager} to leverage to run in the background when necessary. Also
@@ -38,7 +41,8 @@ import java.util.Set;
  */
 public final class WebRtcCallService extends Service implements SignalAudioManager.EventListener {
 
-  private static final String TAG = Log.tag(WebRtcCallService.class);
+  private static final String TAG                        = Log.tag(WebRtcCallService.class);
+  private static final String WEBSOCKET_KEEP_ALIVE_TOKEN = WebRtcCallService.class.getName();
 
   private static final String ACTION_UPDATE              = "UPDATE";
   private static final String ACTION_STOP                = "STOP";
@@ -52,7 +56,10 @@ public final class WebRtcCallService extends Service implements SignalAudioManag
   private static final String EXTRA_ENABLED       = "ENABLED";
   private static final String EXTRA_AUDIO_COMMAND = "AUDIO_COMMAND";
 
-  private static final int INVALID_NOTIFICATION_ID = -1;
+  private static final int  INVALID_NOTIFICATION_ID           = -1;
+  private static final long REQUEST_WEBSOCKET_STAY_OPEN_DELAY = TimeUnit.MINUTES.toMillis(1);
+
+  private final WebSocketKeepAliveTask webSocketKeepAliveTask = new WebSocketKeepAliveTask();
 
   private SignalCallManager callManager;
 
@@ -147,15 +154,20 @@ public final class WebRtcCallService extends Service implements SignalAudioManag
     if (!AndroidTelecomUtil.getTelecomSupported()) {
       TelephonyUtil.getManager(this).listen(hangUpRtcOnDeviceCallAnswered, PhoneStateListener.LISTEN_NONE);
     }
+
+    webSocketKeepAliveTask.stop();
   }
 
   @Override
   public int onStartCommand(Intent intent, int flags, int startId) {
     if (intent == null || intent.getAction() == null) {
+      setCallNotification();
+      stop();
       return START_NOT_STICKY;
     }
 
     Log.i(TAG, "action: " + intent.getAction());
+    webSocketKeepAliveTask.start();
 
     switch (intent.getAction()) {
       case ACTION_UPDATE:
@@ -296,6 +308,37 @@ public final class WebRtcCallService extends Service implements SignalAudioManag
     }
   }
 
+  /**
+   * Periodically request the web socket stay open if we are doing anything call related.
+   */
+  private class WebSocketKeepAliveTask implements Runnable {
+    private boolean keepRunning = false;
+
+    @MainThread
+    public void start() {
+      if (!keepRunning) {
+        keepRunning = true;
+        run();
+      }
+    }
+
+    @MainThread
+    public void stop() {
+      keepRunning = false;
+      ThreadUtil.cancelRunnableOnMain(webSocketKeepAliveTask);
+      ApplicationDependencies.getIncomingMessageObserver().removeKeepAliveToken(WEBSOCKET_KEEP_ALIVE_TOKEN);
+    }
+
+    @MainThread
+    @Override
+    public void run() {
+      if (keepRunning) {
+        ApplicationDependencies.getIncomingMessageObserver().registerKeepAliveToken(WEBSOCKET_KEEP_ALIVE_TOKEN);
+        ThreadUtil.runOnMainDelayed(this, REQUEST_WEBSOCKET_STAY_OPEN_DELAY);
+      }
+    }
+  }
+
   private static class NetworkReceiver extends BroadcastReceiver {
     @Override
     public void onReceive(Context context, Intent intent) {

app/src/main/java/org/thoughtcrime/securesms/util/Util.java

@@ -161,6 +161,10 @@ public class Util {
     return collection != null && !collection.isEmpty();
   }
 
+  public static <K, V> boolean hasItems(@Nullable Map<K, V> map) {
+    return map != null && !map.isEmpty();
+  }
+
   public static <K, V> V getOrDefault(@NonNull Map<K, V> map, K key, V defaultValue) {
     return map.containsKey(key) ? map.get(key) : defaultValue;
   }