Create a WebsocketDrainedConstraint.

app/src/main/java/org/thoughtcrime/securesms/jobmanager/impl/WebsocketDrainedConstraint.java

@@ -0,0 +1,44 @@
+package org.thoughtcrime.securesms.jobmanager.impl;
+
+import android.app.job.JobInfo;
+
+import androidx.annotation.NonNull;
+import androidx.annotation.RequiresApi;
+
+import org.thoughtcrime.securesms.dependencies.ApplicationDependencies;
+import org.thoughtcrime.securesms.jobmanager.Constraint;
+
+/**
+ * A constraint that is met once we have pulled down all messages from the websocket during initial
+ * load. See {@link org.thoughtcrime.securesms.messages.InitialMessageRetriever}.
+ */
+public final class WebsocketDrainedConstraint implements Constraint {
+
+  public static final String KEY = "WebsocketDrainedConstraint";
+
+  private WebsocketDrainedConstraint() {
+  }
+
+  @Override
+  public boolean isMet() {
+    return ApplicationDependencies.getInitialMessageRetriever().isCaughtUp();
+  }
+
+  @Override
+  public @NonNull String getFactoryKey() {
+    return KEY;
+  }
+
+  @RequiresApi(26)
+  @Override
+  public void applyToJobInfo(@NonNull JobInfo.Builder jobInfoBuilder) {
+  }
+
+  public static final class Factory implements Constraint.Factory<WebsocketDrainedConstraint> {
+
+    @Override
+    public WebsocketDrainedConstraint create() {
+      return new WebsocketDrainedConstraint();
+    }
+  }
+}

app/src/main/java/org/thoughtcrime/securesms/jobmanager/impl/WebsocketDrainedConstraintObserver.java

@@ -0,0 +1,22 @@
+package org.thoughtcrime.securesms.jobmanager.impl;
+
+import androidx.annotation.NonNull;
+
+import org.thoughtcrime.securesms.dependencies.ApplicationDependencies;
+import org.thoughtcrime.securesms.jobmanager.ConstraintObserver;
+
+/**
+ * An observer for {@link WebsocketDrainedConstraint}. Will fire when the
+ * {@link org.thoughtcrime.securesms.messages.InitialMessageRetriever} is caught up.
+ */
+public class WebsocketDrainedConstraintObserver implements ConstraintObserver {
+
+  private static final String REASON = WebsocketDrainedConstraintObserver.class.getSimpleName();
+
+  @Override
+  public void register(@NonNull Notifier notifier) {
+    ApplicationDependencies.getInitialMessageRetriever().addListener(() -> {
+      notifier.onConstraintMet(REASON);
+    });
+  }
+}

app/src/main/java/org/thoughtcrime/securesms/jobs/JobManagerFactories.java

@@ -15,6 +15,8 @@ import org.thoughtcrime.securesms.jobmanager.impl.NetworkConstraintObserver;
 import org.thoughtcrime.securesms.jobmanager.impl.NetworkOrCellServiceConstraint;
 import org.thoughtcrime.securesms.jobmanager.impl.SqlCipherMigrationConstraint;
 import org.thoughtcrime.securesms.jobmanager.impl.SqlCipherMigrationConstraintObserver;
+import org.thoughtcrime.securesms.jobmanager.impl.WebsocketDrainedConstraint;
+import org.thoughtcrime.securesms.jobmanager.impl.WebsocketDrainedConstraintObserver;
 import org.thoughtcrime.securesms.jobmanager.migrations.PushProcessMessageQueueJobMigration;
 import org.thoughtcrime.securesms.jobmanager.migrations.RecipientIdFollowUpJobMigration;
 import org.thoughtcrime.securesms.jobmanager.migrations.RecipientIdFollowUpJobMigration2;
@@ -149,13 +151,15 @@ public final class JobManagerFactories {
       put(NetworkOrCellServiceConstraint.KEY,        new NetworkOrCellServiceConstraint.Factory(application));
       put(NetworkOrCellServiceConstraint.LEGACY_KEY, new NetworkOrCellServiceConstraint.Factory(application));
       put(SqlCipherMigrationConstraint.KEY,          new SqlCipherMigrationConstraint.Factory(application));
+      put(WebsocketDrainedConstraint.KEY,            new WebsocketDrainedConstraint.Factory());
     }};
   }
 
   public static List<ConstraintObserver> getConstraintObservers(@NonNull Application application) {
     return Arrays.asList(CellServiceConstraintObserver.getInstance(application),
                          new NetworkConstraintObserver(application),
-                         new SqlCipherMigrationConstraintObserver());
+                         new SqlCipherMigrationConstraintObserver(),
+                         new WebsocketDrainedConstraintObserver());
   }
 
   public static List<JobMigration> getJobMigrations(@NonNull Application application) {

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

@@ -2,7 +2,6 @@ package org.thoughtcrime.securesms.messages;
 
 import android.content.Context;
 
-import androidx.annotation.AnyThread;
 import androidx.annotation.NonNull;
 import androidx.annotation.WorkerThread;
 
@@ -11,7 +10,6 @@ import org.thoughtcrime.securesms.dependencies.ApplicationDependencies;
 import org.thoughtcrime.securesms.logging.Log;
 import org.thoughtcrime.securesms.util.concurrent.SignalExecutors;
 
-import java.io.IOException;
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;