Integrate new closed group poller

app/src/main/java/org/thoughtcrime/securesms/ApplicationContext.java

@@ -36,7 +36,7 @@ import org.session.libsession.messaging.file_server.FileServerAPI;
 import org.session.libsession.messaging.mentions.MentionsManager;
 import org.session.libsession.messaging.open_groups.OpenGroupAPI;
 import org.session.libsession.messaging.sending_receiving.notifications.MessageNotifier;
-import org.session.libsession.messaging.sending_receiving.pollers.ClosedGroupPoller;
+import org.session.libsession.messaging.sending_receiving.pollers.ClosedGroupPollerV2;
 import org.session.libsession.messaging.sending_receiving.pollers.Poller;
 import org.session.libsession.snode.SnodeModule;
 import org.session.libsession.utilities.Address;
@@ -131,7 +131,6 @@ public class ApplicationContext extends MultiDexApplication implements Dependenc
     // Loki
     public MessageNotifier messageNotifier = null;
     public Poller poller = null;
-    public ClosedGroupPoller closedGroupPoller = null;
     public Broadcaster broadcaster = null;
     public SignalCommunicationModule communicationModule;
     private Job firebaseInstanceIdJob;
@@ -221,9 +220,7 @@ public class ApplicationContext extends MultiDexApplication implements Dependenc
         if (poller != null) {
             poller.stopIfNeeded();
         }
-        if (closedGroupPoller != null) {
-            closedGroupPoller.stopIfNeeded();
-        }
+        ClosedGroupPollerV2.getShared().stop();
     }
 
     @Override
@@ -449,7 +446,6 @@ public class ApplicationContext extends MultiDexApplication implements Dependenc
             return;
         }
         poller = new Poller();
-        closedGroupPoller = new ClosedGroupPoller();
     }
 
     public void startPollingIfNeeded() {
@@ -457,9 +453,7 @@ public class ApplicationContext extends MultiDexApplication implements Dependenc
         if (poller != null) {
             poller.startIfNeeded();
         }
-        if (closedGroupPoller != null) {
-            closedGroupPoller.startIfNeeded();
-        }
+        ClosedGroupPollerV2.getShared().start();
     }
 
     private void resubmitProfilePictureIfNeeded() {

app/src/main/java/org/thoughtcrime/securesms/loki/api/BackgroundPollWorker.kt

@@ -7,9 +7,9 @@ import androidx.work.*
 import nl.komponents.kovenant.Promise
 import nl.komponents.kovenant.all
 import nl.komponents.kovenant.functional.map
+import org.session.libsession.messaging.MessagingModuleConfiguration
 import org.session.libsession.messaging.jobs.MessageReceiveJob
-import org.session.libsession.messaging.open_groups.OpenGroupV2
-import org.session.libsession.messaging.sending_receiving.pollers.ClosedGroupPoller
+import org.session.libsession.messaging.sending_receiving.pollers.ClosedGroupPollerV2
 import org.session.libsession.messaging.sending_receiving.pollers.OpenGroupPollerV2
 import org.session.libsession.snode.SnodeAPI
 import org.session.libsession.utilities.TextSecurePreferences
@@ -57,7 +57,10 @@ class BackgroundPollWorker(val context: Context, params: WorkerParameters) : Wor
             promises.addAll(dmsPromise.get())
 
             // Closed groups
-            promises.addAll(ClosedGroupPoller().pollOnce())
+            val closedGroupPoller = ClosedGroupPollerV2() // Intentionally don't use shared
+            val storage = MessagingModuleConfiguration.shared.storage
+            val allGroupPublicKeys = storage.getAllClosedGroupPublicKeys()
+            allGroupPublicKeys.forEach { closedGroupPoller.poll(it) }
 
             // Open Groups
             val threadDB = DatabaseFactory.getLokiThreadDatabase(context)

libsession/src/main/java/org/session/libsession/messaging/sending_receiving/pollers/ClosedGroupPoller.kt

@@ -1,82 +0,0 @@
-package org.session.libsession.messaging.sending_receiving.pollers
-
-import android.os.Handler
-import nl.komponents.kovenant.Promise
-import nl.komponents.kovenant.functional.bind
-import nl.komponents.kovenant.functional.map
-import org.session.libsession.messaging.MessagingModuleConfiguration
-import org.session.libsession.messaging.jobs.JobQueue
-import org.session.libsession.messaging.jobs.MessageReceiveJob
-import org.session.libsession.snode.SnodeAPI
-import org.session.libsignal.crypto.getRandomElementOrNull
-import org.session.libsignal.utilities.Log
-import org.session.libsignal.utilities.successBackground
-
-class ClosedGroupPoller {
-    private var isPolling = false
-    private val handler: Handler by lazy { Handler() }
-
-    private val task = object : Runnable {
-
-        override fun run() {
-            poll()
-            handler.postDelayed(this, pollInterval)
-        }
-    }
-
-    // region Settings
-    companion object {
-        private val pollInterval: Long = 6 * 1000
-    }
-    // endregion
-
-    // region Error
-    class InsufficientSnodesException() : Exception("No snodes left to poll.")
-    class PollingCanceledException() : Exception("Polling canceled.")
-    // endregion
-
-    // region Public API
-    public fun startIfNeeded() {
-        if (isPolling) { return }
-        isPolling = true
-        task.run()
-    }
-
-    public fun pollOnce(): List<Promise<Unit, Exception>> {
-        if (isPolling) { return listOf() }
-        isPolling = true
-        return poll()
-    }
-
-    public fun stopIfNeeded() {
-        isPolling = false
-        handler.removeCallbacks(task)
-    }
-    // endregion
-
-    // region Private API
-    private fun poll(): List<Promise<Unit, Exception>> {
-        if (!isPolling) { return listOf() }
-        val storage = MessagingModuleConfiguration.shared.storage
-        val publicKeys = storage.getAllActiveClosedGroupPublicKeys()
-        return publicKeys.map { publicKey ->
-            val promise = SnodeAPI.getSwarm(publicKey).bind { swarm ->
-                val snode = swarm.getRandomElementOrNull() ?: throw InsufficientSnodesException() // Should be cryptographically secure
-                if (!isPolling) { throw PollingCanceledException() }
-                SnodeAPI.getRawMessages(snode, publicKey).map {SnodeAPI.parseRawMessagesResponse(it, snode, publicKey) }
-            }
-            promise.successBackground { messages ->
-                if (!storage.isGroupActive(publicKey)) { return@successBackground }
-                messages.forEach { envelope ->
-                    val job = MessageReceiveJob(envelope.toByteArray())
-                    JobQueue.shared.add(job)
-                }
-            }
-            promise.fail {
-                Log.d("Loki", "Polling failed for closed group with public key: $publicKey due to error: $it.")
-            }
-            promise.map { }
-        }
-    }
-    // endregion
-}

libsession/src/main/java/org/session/libsession/messaging/sending_receiving/pollers/ClosedGroupPollerV2.kt

@@ -28,6 +28,7 @@ class ClosedGroupPollerV2 {
         private val minPollInterval = 4 * 1000
         private val maxPollInterval = 2 * 60 * 1000
 
+        @JvmStatic
         val shared = ClosedGroupPollerV2()
     }
 
@@ -86,7 +87,7 @@ class ClosedGroupPollerV2 {
         }, nextPollInterval.toLong(), TimeUnit.MILLISECONDS)
     }
 
-    private fun poll(groupPublicKey: String): Promise<Unit, Exception> {
+    fun poll(groupPublicKey: String): Promise<Unit, Exception> {
         if (!isPolling(groupPublicKey)) { return Promise.of(Unit) }
         val promise = SnodeAPI.getSwarm(groupPublicKey).bind { swarm ->
             val snode = swarm.getRandomElementOrNull() ?: throw InsufficientSnodesException() // Should be cryptographically secure