Shift authority for message availability to the pub/sub event system

signalapp · Nov 7, 2024 · ef716aa · ef716aa
1 parent c91242e
commit ef716aa
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 13 deletions.
diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/websocket/WebSocketConnection.java b/service/src/main/java/org/whispersystems/textsecuregcm/websocket/WebSocketConnection.java
@@ -472,10 +472,6 @@ public boolean handleNewMessagesAvailable() {
         PRESENCE_MANAGER_TAG, "legacy")
         .increment();
 
-    storedMessageState.compareAndSet(StoredMessageState.EMPTY, StoredMessageState.CACHED_NEW_MESSAGES_AVAILABLE);
-
-    processStoredMessages();
-
     return true;
   }
 
@@ -484,6 +480,10 @@ public void handleNewMessageAvailable() {
     Metrics.counter(MESSAGE_AVAILABLE_COUNTER_NAME,
             PRESENCE_MANAGER_TAG, "pubsub")
         .increment();
+
+    storedMessageState.compareAndSet(StoredMessageState.EMPTY, StoredMessageState.CACHED_NEW_MESSAGES_AVAILABLE);
+
+    processStoredMessages();
   }
 
   @Override
@@ -498,10 +498,6 @@ public boolean handleMessagesPersisted() {
         PRESENCE_MANAGER_TAG, "legacy")
         .increment();
 
-    storedMessageState.set(StoredMessageState.PERSISTED_NEW_MESSAGES_AVAILABLE);
-
-    processStoredMessages();
-
     return true;
   }
 
@@ -510,6 +506,10 @@ public void handleMessagesPersistedPubSub() {
     Metrics.counter(MESSAGES_PERSISTED_COUNTER_NAME,
             PRESENCE_MANAGER_TAG, "pubsub")
         .increment();
+
+    storedMessageState.set(StoredMessageState.PERSISTED_NEW_MESSAGES_AVAILABLE);
+
+    processStoredMessages();
   }
 
   @Override

diff --git a/...ice/src/test/java/org/whispersystems/textsecuregcm/websocket/WebSocketConnectionTest.java b/...ice/src/test/java/org/whispersystems/textsecuregcm/websocket/WebSocketConnectionTest.java
@@ -268,15 +268,15 @@ public void testOnlineSend() {
       // or wait for anything.
       connection.start();
 
-      connection.handleNewMessagesAvailable();
+      connection.handleNewMessageAvailable();
 
       synchronized (sendCounter) {
         while (sendCounter.get() < 1) {
           sendCounter.wait();
         }
       }
 
-      connection.handleNewMessagesAvailable();
+      connection.handleNewMessageAvailable();
 
       synchronized (sendCounter) {
         while (sendCounter.get() < 2) {
@@ -693,7 +693,7 @@ public void testRequeryOnStateMismatch() {
 
     when(client.sendRequest(eq("PUT"), eq("/api/v1/message"), any(List.class), any(Optional.class)))
         .thenAnswer(invocation -> {
-          connection.handleNewMessagesAvailable();
+          connection.handleNewMessageAvailable();
 
           return CompletableFuture.completedFuture(successResponse);
         });
@@ -741,7 +741,7 @@ void testProcessCachedMessagesOnly() {
 
     verify(messagesManager).getMessagesForDeviceReactive(account.getUuid(), device, false);
 
-    connection.handleNewMessagesAvailable();
+    connection.handleNewMessageAvailable();
 
     verify(messagesManager).getMessagesForDeviceReactive(account.getUuid(), device, true);
   }
@@ -769,7 +769,7 @@ void testProcessDatabaseMessagesAfterPersist() {
     // whenComplete method will get called immediately on THIS thread, so we don't need to synchronize or wait for
     // anything.
     connection.processStoredMessages();
-    connection.handleMessagesPersisted();
+    connection.handleMessagesPersistedPubSub();
 
     verify(messagesManager, times(2)).getMessagesForDeviceReactive(account.getUuid(), device, false);
   }