Use a new DatabaseObserver system.

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

@@ -1,7 +1,6 @@
 package org.thoughtcrime.securesms.conversation;
 
 import android.app.Application;
-import android.database.ContentObserver;
 
 import androidx.annotation.MainThread;
 import androidx.annotation.NonNull;
@@ -16,7 +15,7 @@ import org.signal.paging.PagedData;
 import org.signal.paging.PagingConfig;
 import org.signal.paging.PagingController;
 import org.signal.paging.ProxyPagingController;
-import org.thoughtcrime.securesms.database.DatabaseContentProviders;
+import org.thoughtcrime.securesms.database.DatabaseObserver;
 import org.thoughtcrime.securesms.dependencies.ApplicationDependencies;
 import org.thoughtcrime.securesms.mediasend.Media;
 import org.thoughtcrime.securesms.mediasend.MediaRepository;
@@ -41,11 +40,10 @@ class ConversationViewModel extends ViewModel {
   private final MutableLiveData<Boolean>            hasUnreadMentions;
   private final LiveData<Boolean>                   canShowAsBubble;
   private final ProxyPagingController               pagingController;
-  private final ContentObserver                     messageObserver;
+  private final DatabaseObserver.Observer           messageObserver;
 
   private ConversationIntents.Args args;
   private int                      jumpToPosition;
-  private boolean                  hasRegisteredObserver;
 
   private ConversationViewModel() {
     this.context                = ApplicationDependencies.getApplication();
@@ -56,12 +54,7 @@ class ConversationViewModel extends ViewModel {
     this.showScrollButtons      = new MutableLiveData<>(false);
     this.hasUnreadMentions      = new MutableLiveData<>(false);
     this.pagingController       = new ProxyPagingController();
-    this.messageObserver        = new ContentObserver(null) {
-      @Override
-      public void onChange(boolean selfChange) {
-        pagingController.onDataInvalidated();
-      }
-    };
+    this.messageObserver        = pagingController::onDataInvalidated;
 
     LiveData<ConversationData> metadata = Transformations.switchMap(threadId, thread -> {
       LiveData<ConversationData> conversationData = conversationRepository.getConversationData(thread, jumpToPosition);
@@ -83,12 +76,8 @@ class ConversationViewModel extends ViewModel {
         startPosition = data.getThreadSize();
       }
 
-      if (hasRegisteredObserver) {
-        context.getContentResolver().unregisterContentObserver(messageObserver);
-      }
-
-      context.getContentResolver().registerContentObserver(DatabaseContentProviders.Conversation.getUriForThread(data.getThreadId()), true, messageObserver);
-      hasRegisteredObserver = true;
+      ApplicationDependencies.getDatabaseObserver().unregisterObserver(messageObserver);
+      ApplicationDependencies.getDatabaseObserver().registerConversationObserver(data.getThreadId(), messageObserver);
 
       ConversationDataSource dataSource = new ConversationDataSource(context, data.getThreadId());
       PagingConfig           config     = new PagingConfig.Builder()
@@ -182,7 +171,7 @@ class ConversationViewModel extends ViewModel {
   @Override
   protected void onCleared() {
     super.onCleared();
-    context.getContentResolver().unregisterContentObserver(messageObserver);
+    ApplicationDependencies.getDatabaseObserver().unregisterObserver(messageObserver);
   }
 
   static class Factory extends ViewModelProvider.NewInstanceFactory {

app/src/main/java/org/thoughtcrime/securesms/conversationlist/ConversationListDataSource.java

@@ -1,7 +1,6 @@
 package org.thoughtcrime.securesms.conversationlist;
 
 import android.content.Context;
-import android.database.ContentObserver;
 import android.database.Cursor;
 import android.database.MatrixCursor;
 import android.database.MergeCursor;
@@ -15,8 +14,8 @@ import org.signal.core.util.concurrent.SignalExecutors;
 import org.signal.core.util.logging.Log;
 import org.thoughtcrime.securesms.conversationlist.model.Conversation;
 import org.thoughtcrime.securesms.conversationlist.model.ConversationReader;
-import org.thoughtcrime.securesms.database.DatabaseContentProviders;
 import org.thoughtcrime.securesms.database.DatabaseFactory;
+import org.thoughtcrime.securesms.database.DatabaseObserver;
 import org.thoughtcrime.securesms.database.ThreadDatabase;
 import org.thoughtcrime.securesms.database.model.ThreadRecord;
 import org.thoughtcrime.securesms.dependencies.ApplicationDependencies;
@@ -42,20 +41,20 @@ abstract class ConversationListDataSource extends PositionalDataSource<Conversat
   protected ConversationListDataSource(@NonNull Context context, @NonNull Invalidator invalidator) {
     this.threadDatabase = DatabaseFactory.getThreadDatabase(context);
 
-    ContentObserver contentObserver = new ContentObserver(null) {
+    DatabaseObserver.Observer observer = new DatabaseObserver.Observer() {
       @Override
-      public void onChange(boolean selfChange) {
+      public void onChanged() {
         invalidate();
-        context.getContentResolver().unregisterContentObserver(this);
+        ApplicationDependencies.getDatabaseObserver().unregisterObserver(this);
       }
     };
 
     invalidator.observe(() -> {
       invalidate();
-      context.getContentResolver().unregisterContentObserver(contentObserver);
+      ApplicationDependencies.getDatabaseObserver().unregisterObserver(observer);
     });
 
-    context.getContentResolver().registerContentObserver(DatabaseContentProviders.ConversationList.CONTENT_URI,  true, contentObserver);
+    ApplicationDependencies.getDatabaseObserver().registerConversationListObserver(observer);
   }
 
   private static ConversationListDataSource create(@NonNull Context context, @NonNull Invalidator invalidator, boolean isArchived) {

app/src/main/java/org/thoughtcrime/securesms/conversationlist/ConversationListViewModel.java

@@ -1,8 +1,6 @@
 package org.thoughtcrime.securesms.conversationlist;
 
 import android.app.Application;
-import android.database.ContentObserver;
-import android.os.Handler;
 import android.text.TextUtils;
 
 import androidx.annotation.NonNull;
@@ -19,8 +17,8 @@ import org.signal.core.util.concurrent.SignalExecutors;
 import org.signal.core.util.logging.Log;
 import org.thoughtcrime.securesms.conversationlist.model.Conversation;
 import org.thoughtcrime.securesms.conversationlist.model.SearchResult;
-import org.thoughtcrime.securesms.database.DatabaseContentProviders;
 import org.thoughtcrime.securesms.database.DatabaseFactory;
+import org.thoughtcrime.securesms.database.DatabaseObserver;
 import org.thoughtcrime.securesms.dependencies.ApplicationDependencies;
 import org.thoughtcrime.securesms.megaphone.Megaphone;
 import org.thoughtcrime.securesms.megaphone.MegaphoneRepository;
@@ -36,32 +34,27 @@ class ConversationListViewModel extends ViewModel {
 
   private static final String TAG = Log.tag(ConversationListViewModel.class);
 
-  private final Application                       application;
   private final MutableLiveData<Megaphone>        megaphone;
   private final MutableLiveData<SearchResult>     searchResult;
   private final LiveData<ConversationList>        conversationList;
   private final SearchRepository                  searchRepository;
   private final MegaphoneRepository               megaphoneRepository;
   private final Debouncer                         debouncer;
-  private final ContentObserver                   observer;
+  private final DatabaseObserver.Observer observer;
   private final Invalidator                       invalidator;
 
   private String lastQuery;
 
   private ConversationListViewModel(@NonNull Application application, @NonNull SearchRepository searchRepository, boolean isArchived) {
-    this.application         = application;
     this.megaphone           = new MutableLiveData<>();
     this.searchResult        = new MutableLiveData<>();
     this.searchRepository    = searchRepository;
     this.megaphoneRepository = ApplicationDependencies.getMegaphoneRepository();
     this.debouncer           = new Debouncer(300);
     this.invalidator         = new Invalidator();
-    this.observer            = new ContentObserver(new Handler()) {
-      @Override
-      public void onChange(boolean selfChange) {
-        if (!TextUtils.isEmpty(getLastQuery())) {
-          searchRepository.query(getLastQuery(), searchResult::postValue);
-        }
+    this.observer            = () -> {
+      if (!TextUtils.isEmpty(getLastQuery())) {
+        searchRepository.query(getLastQuery(), searchResult::postValue);
       }
     };
 
@@ -76,7 +69,7 @@ class ConversationListViewModel extends ViewModel {
                                                                                                     .setInitialLoadKey(0)
                                                                                                     .build();
 
-    application.getContentResolver().registerContentObserver(DatabaseContentProviders.ConversationList.CONTENT_URI, true, observer);
+    ApplicationDependencies.getDatabaseObserver().registerConversationListObserver(observer);
 
     this.conversationList = Transformations.switchMap(conversationList, conversation -> {
       if (conversation.getDataSource().isInvalid()) {
@@ -122,6 +115,7 @@ class ConversationListViewModel extends ViewModel {
 
   void onVisible() {
     megaphoneRepository.getNextMegaphone(megaphone::postValue);
+    ApplicationDependencies.getDatabaseObserver().notifyConversationListListeners();
   }
 
   void onMegaphoneCompleted(@NonNull Megaphones.Event event) {
@@ -157,7 +151,7 @@ class ConversationListViewModel extends ViewModel {
   protected void onCleared() {
     invalidator.invalidate();
     debouncer.clear();
-    application.getContentResolver().unregisterContentObserver(observer);
+    ApplicationDependencies.getDatabaseObserver().unregisterObserver(observer);
   }
 
   public static class Factory extends ViewModelProvider.NewInstanceFactory {

app/src/main/java/org/thoughtcrime/securesms/database/Database.java

@@ -23,6 +23,7 @@ import android.database.Cursor;
 import androidx.annotation.NonNull;
 
 import org.thoughtcrime.securesms.database.helpers.SQLCipherOpenHelper;
+import org.thoughtcrime.securesms.dependencies.ApplicationDependencies;
 
 import java.util.Set;
 
@@ -39,20 +40,27 @@ public abstract class Database {
   }
 
   protected void notifyConversationListeners(Set<Long> threadIds) {
-    for (long threadId : threadIds)
+    ApplicationDependencies.getDatabaseObserver().notifyConversationListeners(threadIds);
+
+    for (long threadId : threadIds) {
       notifyConversationListeners(threadId);
+    }
   }
 
   protected void notifyConversationListeners(long threadId) {
+    ApplicationDependencies.getDatabaseObserver().notifyConversationListeners(threadId);
+
     context.getContentResolver().notifyChange(DatabaseContentProviders.Conversation.getUriForThread(threadId), null);
     notifyVerboseConversationListeners(threadId);
   }
 
   protected void notifyVerboseConversationListeners(long threadId) {
+    ApplicationDependencies.getDatabaseObserver().notifyVerboseConversationListeners(threadId);
     context.getContentResolver().notifyChange(DatabaseContentProviders.Conversation.getVerboseUriForThread(threadId), null);
   }
 
   protected void notifyConversationListListeners() {
+    ApplicationDependencies.getDatabaseObserver().notifyConversationListListeners();
     context.getContentResolver().notifyChange(DatabaseContentProviders.ConversationList.CONTENT_URI, null);
   }
 
@@ -64,26 +72,32 @@ public abstract class Database {
     context.getContentResolver().notifyChange(DatabaseContentProviders.StickerPack.CONTENT_URI, null);
   }
 
+  @Deprecated
   protected void setNotifyConversationListeners(Cursor cursor, long threadId) {
     cursor.setNotificationUri(context.getContentResolver(), DatabaseContentProviders.Conversation.getUriForThread(threadId));
   }
 
+  @Deprecated
   protected void setNotifyConversationListeners(Cursor cursor) {
     cursor.setNotificationUri(context.getContentResolver(), DatabaseContentProviders.Conversation.getUriForAllThreads());
   }
 
+  @Deprecated
   protected void setNotifyVerboseConversationListeners(Cursor cursor, long threadId) {
     cursor.setNotificationUri(context.getContentResolver(), DatabaseContentProviders.Conversation.getVerboseUriForThread(threadId));
   }
 
+  @Deprecated
   protected void setNotifyConversationListListeners(Cursor cursor) {
     cursor.setNotificationUri(context.getContentResolver(), DatabaseContentProviders.ConversationList.CONTENT_URI);
   }
 
+  @Deprecated
   protected void setNotifyStickerListeners(Cursor cursor) {
     cursor.setNotificationUri(context.getContentResolver(), DatabaseContentProviders.Sticker.CONTENT_URI);
   }
 
+  @Deprecated
   protected void setNotifyStickerPackListeners(Cursor cursor) {
     cursor.setNotificationUri(context.getContentResolver(), DatabaseContentProviders.StickerPack.CONTENT_URI);
   }
@@ -101,5 +115,4 @@ public abstract class Database {
   public void reset(SQLCipherOpenHelper databaseHelper) {
     this.databaseHelper = databaseHelper;
   }
-
 }

app/src/main/java/org/thoughtcrime/securesms/database/DatabaseObserver.java

@@ -0,0 +1,142 @@
+package org.thoughtcrime.securesms.database;
+
+import android.app.Application;
+import android.database.ContentObserver;
+import android.database.Cursor;
+
+import androidx.annotation.NonNull;
+
+import org.signal.core.util.concurrent.SignalExecutors;
+import org.thoughtcrime.securesms.util.concurrent.SerialExecutor;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Executor;
+
+/**
+ * Allows listening to database changes to varying degrees of specificity.
+ *
+ * A replacement for the observer system in {@link Database}. We should move to this over time.
+ */
+public final class DatabaseObserver {
+
+  private final Application application;
+  private final Executor    executor;
+
+  private final Set<Observer>            conversationListObservers;
+  private final Map<Long, Set<Observer>> conversationObservers;
+  private final Map<Long, Set<Observer>> verboseConversationObservers;
+
+  public DatabaseObserver(Application application) {
+    this.application                  = application;
+    this.executor                     = new SerialExecutor(SignalExecutors.BOUNDED);
+    this.conversationListObservers    = new HashSet<>();
+    this.conversationObservers        = new HashMap<>();
+    this.verboseConversationObservers = new HashMap<>();
+  }
+
+  public void registerConversationListObserver(@NonNull Observer listener) {
+    executor.execute(() -> {
+      conversationListObservers.add(listener);
+    });
+  }
+
+  public void registerConversationObserver(long threadId, @NonNull Observer listener) {
+    executor.execute(() -> {
+      registerMapped(conversationObservers, threadId, listener);
+    });
+  }
+
+  public void registerVerboseConversationObserver(long threadId, @NonNull Observer listener) {
+    executor.execute(() -> {
+      registerMapped(verboseConversationObservers, threadId, listener);
+    });
+  }
+
+  public void unregisterObserver(@NonNull Observer listener) {
+    executor.execute(() -> {
+      conversationListObservers.remove(listener);
+      unregisterMapped(conversationObservers, listener);
+      unregisterMapped(verboseConversationObservers, listener);
+    });
+  }
+
+  public void notifyConversationListeners(Set<Long> threadIds) {
+    executor.execute(() -> {
+      for (long threadId : threadIds) {
+        notifyMapped(conversationObservers, threadId);
+        notifyMapped(verboseConversationObservers, threadId);
+      }
+    });
+
+    for (long threadId : threadIds) {
+      application.getContentResolver().notifyChange(DatabaseContentProviders.Conversation.getUriForThread(threadId), null);
+      application.getContentResolver().notifyChange(DatabaseContentProviders.Conversation.getVerboseUriForThread(threadId), null);
+    }
+  }
+
+  public void notifyConversationListeners(long threadId) {
+    executor.execute(() -> {
+      notifyMapped(conversationObservers, threadId);
+      notifyMapped(verboseConversationObservers, threadId);
+    });
+
+    application.getContentResolver().notifyChange(DatabaseContentProviders.Conversation.getUriForThread(threadId), null);
+    application.getContentResolver().notifyChange(DatabaseContentProviders.Conversation.getVerboseUriForThread(threadId), null);
+  }
+
+  public void notifyVerboseConversationListeners(long threadId) {
+    executor.execute(() -> {
+      notifyMapped(verboseConversationObservers, threadId);
+    });
+
+    application.getContentResolver().notifyChange(DatabaseContentProviders.Conversation.getVerboseUriForThread(threadId), null);
+  }
+
+  public void notifyConversationListListeners() {
+    executor.execute(() -> {
+      for (Observer listener : conversationListObservers) {
+        listener.onChanged();
+      }
+    });
+
+    application.getContentResolver().notifyChange(DatabaseContentProviders.ConversationList.CONTENT_URI, null);
+  }
+
+  private <K> void registerMapped(@NonNull Map<K, Set<Observer>> map, @NonNull K key, @NonNull Observer listener) {
+    Set<Observer> listeners = map.get(key);
+
+    if (listeners == null) {
+      listeners = new HashSet<>();
+    }
+
+    listeners.add(listener);
+    map.put(key, listeners);
+  }
+
+  private <K> void unregisterMapped(@NonNull Map<K, Set<Observer>> map, @NonNull Observer listener) {
+    for (Map.Entry<K, Set<Observer>> entry : map.entrySet()) {
+      entry.getValue().remove(listener);
+    }
+  }
+
+  private static <K> void notifyMapped(@NonNull Map<K, Set<Observer>> map, @NonNull K key) {
+    Set<Observer> listeners = map.get(key);
+
+    if (listeners != null) {
+      for (Observer listener : listeners) {
+        listener.onChanged();
+      }
+    }
+  }
+
+  public interface Observer {
+    /**
+     * Called when the relevant data changes. Executed on a serial executor, so don't do any
+     * long-running tasks!
+     */
+    void onChanged();
+  }
+}

app/src/main/java/org/thoughtcrime/securesms/dependencies/ApplicationDependencies.java

@@ -8,6 +8,7 @@ import androidx.annotation.NonNull;
 import org.thoughtcrime.securesms.KbsEnclave;
 import org.thoughtcrime.securesms.components.TypingStatusRepository;
 import org.thoughtcrime.securesms.components.TypingStatusSender;
+import org.thoughtcrime.securesms.database.DatabaseObserver;
 import org.thoughtcrime.securesms.groups.GroupsV2Authorization;
 import org.thoughtcrime.securesms.groups.GroupsV2AuthorizationMemoryValueCache;
 import org.thoughtcrime.securesms.groups.v2.processing.GroupsV2StateProcessor;
@@ -66,6 +67,7 @@ public class ApplicationDependencies {
   private static volatile EarlyMessageCache            earlyMessageCache;
   private static volatile TypingStatusRepository       typingStatusRepository;
   private static volatile TypingStatusSender           typingStatusSender;
+  private static volatile DatabaseObserver             databaseObserver;
 
   @MainThread
   public static void init(@NonNull Application application, @NonNull Provider provider) {
@@ -299,6 +301,18 @@ public class ApplicationDependencies {
     return typingStatusSender;
   }
 
+  public static @NonNull DatabaseObserver getDatabaseObserver() {
+    if (databaseObserver == null) {
+      synchronized (LOCK) {
+        if (databaseObserver == null) {
+          databaseObserver = provider.provideDatabaseObserver();
+        }
+      }
+    }
+
+    return databaseObserver;
+  }
+
   public interface Provider {
     @NonNull GroupsV2Operations provideGroupsV2Operations();
     @NonNull SignalServiceAccountManager provideSignalServiceAccountManager();
@@ -317,5 +331,6 @@ public class ApplicationDependencies {
     @NonNull TrimThreadsByDateManager provideTrimThreadsByDateManager();
     @NonNull TypingStatusRepository provideTypingStatusRepository();
     @NonNull TypingStatusSender provideTypingStatusSender();
+    @NonNull DatabaseObserver provideDatabaseObserver();
   }
 }

app/src/main/java/org/thoughtcrime/securesms/dependencies/ApplicationDependencyProvider.java

@@ -13,6 +13,7 @@ import org.thoughtcrime.securesms.components.TypingStatusRepository;
 import org.thoughtcrime.securesms.components.TypingStatusSender;
 import org.thoughtcrime.securesms.crypto.storage.SignalProtocolStoreImpl;
 import org.thoughtcrime.securesms.database.DatabaseFactory;
+import org.thoughtcrime.securesms.database.DatabaseObserver;
 import org.thoughtcrime.securesms.events.ReminderUpdateEvent;
 import org.thoughtcrime.securesms.jobmanager.JobManager;
 import org.thoughtcrime.securesms.jobmanager.JobMigrator;
@@ -190,6 +191,11 @@ public class ApplicationDependencyProvider implements ApplicationDependencies.Pr
     return new TypingStatusSender(context);
   }
 
+  @Override
+  public @NonNull DatabaseObserver provideDatabaseObserver() {
+    return new DatabaseObserver(context);
+  }
+
   private static class DynamicCredentialsProvider implements CredentialsProvider {
 
     private final Context context;

app/src/test/java/org/thoughtcrime/securesms/conversationlist/UnarchivedConversationListDataSourceTest.java

@@ -16,6 +16,7 @@ import org.robolectric.RobolectricTestRunner;
 import org.robolectric.annotation.Config;
 import org.thoughtcrime.securesms.conversationlist.model.ConversationReader;
 import org.thoughtcrime.securesms.database.DatabaseFactory;
+import org.thoughtcrime.securesms.database.DatabaseObserver;
 import org.thoughtcrime.securesms.database.ThreadDatabase;
 import org.thoughtcrime.securesms.dependencies.ApplicationDependencies;
 import org.thoughtcrime.securesms.util.paging.Invalidator;
@@ -34,7 +35,7 @@ import static org.powermock.api.mockito.PowerMockito.when;
 @RunWith(RobolectricTestRunner.class)
 @Config(manifest = Config.NONE, application = Application.class)
 @PowerMockIgnore({ "org.mockito.*", "org.robolectric.*", "android.*", "androidx.*" })
-@PrepareForTest({ ApplicationDependencies.class, DatabaseFactory.class })
+@PrepareForTest({ ApplicationDependencies.class, DatabaseFactory.class, DatabaseObserver.class })
 public class UnarchivedConversationListDataSourceTest {
 
   @Rule
@@ -52,6 +53,7 @@ public class UnarchivedConversationListDataSourceTest {
     threadDatabase = mock(ThreadDatabase.class);
 
     when(DatabaseFactory.getThreadDatabase(any())).thenReturn(threadDatabase);
+    when(ApplicationDependencies.getDatabaseObserver()).thenReturn(mock(DatabaseObserver.class));
 
     testSubject = new ConversationListDataSource.UnarchivedConversationListDataSource(ApplicationProvider.getApplicationContext(), mock(Invalidator.class));
   }