Fix some ordering problems with conversation data loading.

app/src/main/java/org/thoughtcrime/securesms/conversation/ConversationDataSource.java

@@ -12,6 +12,7 @@ import org.thoughtcrime.securesms.database.DatabaseFactory;
 import org.thoughtcrime.securesms.database.MmsSmsDatabase;
 import org.thoughtcrime.securesms.database.model.MessageRecord;
 import org.thoughtcrime.securesms.logging.Log;
+import org.thoughtcrime.securesms.util.Util;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -25,10 +26,12 @@ class ConversationDataSource extends PositionalDataSource<MessageRecord> {
 
   private final Context             context;
   private final long                threadId;
+  private final DataUpdatedCallback dataUpdateCallback;
 
-  private ConversationDataSource(@NonNull Context context, long threadId) {
+  private ConversationDataSource(@NonNull Context context, long threadId, @NonNull DataUpdatedCallback dataUpdateCallback) {
     this.context            = context;
     this.threadId           = threadId;
+    this.dataUpdateCallback = dataUpdateCallback;
 
     ContentObserver contentObserver = new ContentObserver(null) {
       @Override
@@ -56,6 +59,8 @@ class ConversationDataSource extends PositionalDataSource<MessageRecord> {
     }
 
     callback.onResult(records, params.requestedStartPosition, db.getConversationCount(threadId));
+    Util.runOnMain(dataUpdateCallback::onDataUpdated);
+
     Log.d(TAG, "[Initial Load] " + (System.currentTimeMillis() - start) + " ms" + (isInvalid() ? " -- invalidated" : ""));
   }
 
@@ -74,22 +79,30 @@ class ConversationDataSource extends PositionalDataSource<MessageRecord> {
     }
 
     callback.onResult(records);
+    Util.runOnMain(dataUpdateCallback::onDataUpdated);
+
     Log.d(TAG, "[Update] " + (System.currentTimeMillis() - start) + " ms" + (isInvalid() ? " -- invalidated" : ""));
   }
 
+  interface DataUpdatedCallback {
+    void onDataUpdated();
+  }
+
   static class Factory extends DataSource.Factory<Integer, MessageRecord> {
 
     private final Context             context;
     private final long                threadId;
+    private final DataUpdatedCallback callback;
 
-    Factory(Context context, long threadId) {
+    Factory(Context context, long threadId, @NonNull DataUpdatedCallback callback) {
       this.context  = context;
       this.threadId = threadId;
+      this.callback = callback;
     }
 
     @Override
     public @NonNull DataSource<Integer, MessageRecord> create() {
-      return new ConversationDataSource(context, threadId);
+      return new ConversationDataSource(context, threadId, callback);
     }
   }
 }

app/src/main/java/org/thoughtcrime/securesms/conversation/ConversationFragment.java

@@ -864,6 +864,8 @@ public class ConversationFragment extends Fragment {
   }
 
   private void presentConversationMetadata(@NonNull ConversationData conversation) {
+    Log.d(TAG, "presentConversationMetadata()");
+
     ConversationAdapter adapter = getListAdapter();
     if (adapter == null) {
       return;
@@ -946,9 +948,18 @@ public class ConversationFragment extends Fragment {
   private void moveToMessagePosition(int position, @Nullable Runnable onMessageNotFound) {
     if (position >= 0) {
       list.scrollToPosition(position);
+
+      if (getListAdapter() == null || getListAdapter().getItem(position) == null) {
+        Log.i(TAG, "[moveToMessagePosition] Position " + position + " not currently populated. Scheduling a jump.");
+        conversationViewModel.scheduleForNextMessageUpdate(() -> {
+          list.scrollToPosition(position);
           getListAdapter().pulseHighlightItem(position);
+        });
       } else {
-      Log.w(TAG, "Tried to navigate to message, but it wasn't found.");
+        getListAdapter().pulseHighlightItem(position);
+      }
+    } else {
+      Log.w(TAG, "[moveToMessagePosition] Tried to navigate to message, but it wasn't found.");
       if (onMessageNotFound != null) {
         onMessageNotFound.run();
       }

app/src/main/java/org/thoughtcrime/securesms/conversation/ConversationViewModel.java

@@ -3,8 +3,11 @@ package org.thoughtcrime.securesms.conversation;
 import android.app.Application;
 
 import androidx.annotation.NonNull;
+import androidx.arch.core.util.Function;
 import androidx.lifecycle.LiveData;
+import androidx.lifecycle.MediatorLiveData;
 import androidx.lifecycle.MutableLiveData;
+import androidx.lifecycle.Observer;
 import androidx.lifecycle.Transformations;
 import androidx.lifecycle.ViewModel;
 import androidx.lifecycle.ViewModelProvider;
@@ -17,9 +20,13 @@ import org.thoughtcrime.securesms.dependencies.ApplicationDependencies;
 import org.thoughtcrime.securesms.logging.Log;
 import org.thoughtcrime.securesms.mediasend.Media;
 import org.thoughtcrime.securesms.mediasend.MediaRepository;
+import org.thoughtcrime.securesms.util.Util;
 import org.thoughtcrime.securesms.util.concurrent.SignalExecutors;
+import org.whispersystems.libsignal.util.Pair;
 
+import java.util.LinkedList;
 import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
 
 class ConversationViewModel extends ViewModel {
 
@@ -32,6 +39,7 @@ class ConversationViewModel extends ViewModel {
   private final MutableLiveData<Long>              threadId;
   private final LiveData<PagedList<MessageRecord>> messages;
   private final LiveData<ConversationData>         conversationMetadata;
+  private final List<Runnable>                     onNextMessageLoad;
 
   private int jumpToPosition;
 
@@ -41,21 +49,27 @@ class ConversationViewModel extends ViewModel {
     this.conversationRepository = new ConversationRepository();
     this.recentMedia            = new MutableLiveData<>();
     this.threadId               = new MutableLiveData<>();
+    this.onNextMessageLoad      = new CopyOnWriteArrayList<>();
 
-    messages = Transformations.switchMap(threadId, thread -> {
+    LiveData<Pair<Long, PagedList<MessageRecord>>> messagesForThreadId = Transformations.switchMap(threadId, thread -> {
-      DataSource.Factory<Integer, MessageRecord> factory = new ConversationDataSource.Factory(context, thread);
+      DataSource.Factory<Integer, MessageRecord> factory = new ConversationDataSource.Factory(context, thread, this::onMessagesUpdated);
       PagedList.Config                           config  = new PagedList.Config.Builder()
                                                                                .setPageSize(25)
                                                                                .setInitialLoadSizeHint(25)
                                                                                .build();
 
-      return new LivePagedListBuilder<>(factory, config).setFetchExecutor(SignalExecutors.BOUNDED)
+      return Transformations.map(new LivePagedListBuilder<>(factory, config).setFetchExecutor(SignalExecutors.BOUNDED)
                                                                             .setInitialLoadKey(Math.max(jumpToPosition, 0))
-                                                        .build();
+                                                                            .build(),
+                                 input -> new Pair<>(thread, input));
     });
 
-    conversationMetadata = Transformations.switchMap(threadId, thread -> {
+    this.messages = Transformations.map(messagesForThreadId, Pair::second);
-      LiveData<ConversationData> data = conversationRepository.getConversationData(thread, jumpToPosition);
+
+    LiveData<Long> threadIdForLoadedMessages = Transformations.distinctUntilChanged(Transformations.map(messagesForThreadId, Pair::first));
+
+    conversationMetadata = Transformations.switchMap(threadIdForLoadedMessages, m -> {
+      LiveData<ConversationData> data = conversationRepository.getConversationData(m, jumpToPosition);
       jumpToPosition = -1;
       return data;
     });
@@ -92,6 +106,18 @@ class ConversationViewModel extends ViewModel {
     return conversationMetadata.getValue() != null ? conversationMetadata.getValue().getLastSeenPosition() : 0;
   }
 
+  void scheduleForNextMessageUpdate(@NonNull Runnable runnable) {
+    onNextMessageLoad.add(runnable);
+  }
+
+  private void onMessagesUpdated() {
+    for (Runnable runnable : onNextMessageLoad) {
+      runnable.run();
+    }
+
+    onNextMessageLoad.clear();
+  }
+
   static class Factory extends ViewModelProvider.NewInstanceFactory {
     @Override
     public @NonNull<T extends ViewModel> T create(@NonNull Class<T> modelClass) {
@@ -99,6 +125,4 @@ class ConversationViewModel extends ViewModel {
       return modelClass.cast(new ConversationViewModel());
     }
   }
-
-
 }