Detect and recycle dangling websockets.

dpapavas · Sep 27, 2018 · b4af5da · b4af5da
1 parent 1af9304
commit b4af5da
Showing 1 changed file with 108 additions and 5 deletions.
diff --git a/src/org/thoughtcrime/securesms/service/MessageRetrievalService.java b/src/org/thoughtcrime/securesms/service/MessageRetrievalService.java
@@ -1,8 +1,15 @@
 package org.thoughtcrime.securesms.service;
 
 import android.app.Service;
+import android.content.BroadcastReceiver;
 import android.content.Context;
 import android.content.Intent;
+import android.content.IntentFilter;
+import android.net.ConnectivityManager;
+import android.net.Network;
+import android.net.NetworkRequest;
+import android.net.NetworkInfo;
+import android.os.Build;
 import android.os.IBinder;
 import android.support.annotation.Nullable;
 import android.support.v4.app.NotificationCompat;
@@ -22,6 +29,7 @@
 import org.whispersystems.signalservice.api.SignalServiceMessagePipe;
 import org.whispersystems.signalservice.api.SignalServiceMessageReceiver;
 
+import java.io.IOException;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
@@ -44,6 +52,8 @@ public class MessageRetrievalService extends Service implements InjectableType,
 
   private NetworkRequirement         networkRequirement;
   private NetworkRequirementProvider networkRequirementProvider;
+  private BroadcastReceiver connectivityChangeReceiver;
+  private ConnectivityManager.NetworkCallback connectivityChangeCallback;
 
   @Inject
   public SignalServiceMessageReceiver receiver;
@@ -67,6 +77,7 @@ public void onCreate() {
     retrievalThread = new MessageRetrievalThread();
     retrievalThread.start();
 
+    monitorNetworkIfNecessary();
     setForegroundIfNecessary();
   }
 
@@ -88,6 +99,18 @@ public void onDestroy() {
       retrievalThread.stopThread();
     }
 
+    if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.M) {
+      if (connectivityChangeCallback != null) {
+        final ConnectivityManager connectivityManager = (ConnectivityManager)getSystemService(Context.CONNECTIVITY_SERVICE);
+
+        connectivityManager.unregisterNetworkCallback(connectivityChangeCallback);
+      }
+    } else {
+      if (connectivityChangeReceiver != null) {
+        unregisterReceiver(connectivityChangeReceiver);
+      }
+    }
+
     sendBroadcast(new Intent("org.thoughtcrime.securesms.RESTART"));
   }
 
@@ -103,6 +126,80 @@ public IBinder onBind(Intent intent) {
     return null;
   }
 
+  private void monitorNetworkIfNecessary() {
+    if (TextSecurePreferences.isGcmDisabled(this)) {
+      if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.M) {
+        final ConnectivityManager connectivityManager = (ConnectivityManager)getSystemService(Context.CONNECTIVITY_SERVICE);
+
+        connectivityChangeCallback = new ConnectivityManager.NetworkCallback() {
+          private Network current;
+
+          private void update(Network network) {
+            final Network previous = current;
+            current = network;
+
+            Log.d(TAG, "Currently active network: " + network);
+
+            if (previous != null && (current == null || !current.equals(previous))) {
+              Log.d(TAG,
+                    "Active network changed (" + previous + " -> " + current +
+                    "); interrupting the retrieval thread to recycle the pipe.");
+
+              retrievalThread.interrupt();
+            }
+          }
+
+          @Override
+          public void onAvailable(Network network) {
+            final ConnectivityManager connectivityManager = (ConnectivityManager)getSystemService(Context.CONNECTIVITY_SERVICE);
+
+            update(connectivityManager.getActiveNetwork());
+          }
+
+          @Override
+          public void onLost(Network network) {
+            final ConnectivityManager connectivityManager = (ConnectivityManager)getSystemService(Context.CONNECTIVITY_SERVICE);
+
+            update(connectivityManager.getActiveNetwork());
+          }
+        };
+
+        connectivityManager.registerNetworkCallback(new NetworkRequest.Builder().build(),
+                                                    connectivityChangeCallback);
+      } else {
+        connectivityChangeReceiver = new BroadcastReceiver() {
+          private int current = -1;
+
+          @Override
+          public void onReceive(Context context, Intent intent) {
+            final ConnectivityManager connectivityManager = (ConnectivityManager)getSystemService(Context.CONNECTIVITY_SERVICE);
+
+            final NetworkInfo info = connectivityManager.getActiveNetworkInfo();
+            final int previous = current;
+
+            if (info == null) {
+              current = -1;
+            } else if (info.isConnected()) {
+              current = info.getType();
+            }
+
+            Log.d(TAG, "Currently active network: " + current);
+
+            if (previous != -1 && previous != current) {
+              Log.d(TAG,
+                    "Active network changed (" + previous + " -> " + current +
+                    "); interrupting the retrieval thread to recycle the pipe.");
+              retrievalThread.interrupt();
+            }
+          }
+        };
+
+        registerReceiver(connectivityChangeReceiver,
+                         new IntentFilter(ConnectivityManager.CONNECTIVITY_ACTION));
+      }
+    }
+  }
+
   private void setForegroundIfNecessary() {
     if (TextSecurePreferences.isGcmDisabled(this)) {
       NotificationCompat.Builder builder = new NotificationCompat.Builder(this, NotificationChannels.OTHER);
@@ -153,10 +250,12 @@ private synchronized boolean isConnectionNecessary() {
   }
 
   private synchronized void waitForConnectionNecessary() {
-    try {
-      while (!isConnectionNecessary()) wait();
-    } catch (InterruptedException e) {
-      throw new AssertionError(e);
+    while (!isConnectionNecessary()) {
+      try {
+        wait();
+      } catch (InterruptedException e) {
+        Log.d(TAG, "Retrieval thread interrupted while not connected; ignoring.");
+      }
     }
   }
 
@@ -205,7 +304,7 @@ public void run() {
         SignalServiceMessagePipe localPipe = pipe;
 
         try {
-          while (isConnectionNecessary() && !stopThread.get()) {
+          while (isConnectionNecessary() && !stopThread.get() && !interrupted()) {
             try {
               Log.i(TAG, "Reading message...");
               localPipe.read(REQUEST_TIMEOUT_MINUTES, TimeUnit.MINUTES,
@@ -223,6 +322,10 @@ public void run() {
               Log.w(TAG, e);
             }
           }
+        } catch (InterruptedException e) {
+          Log.d(TAG, "Retrieval thread interrupted.");
+        } catch (IOException e) {
+          Log.d(TAG, "Message pipe failed: " + e.getMessage());
         } catch (Throwable e) {
           Log.w(TAG, e);
         } finally {