Merge pull request #2075 from OneSignal/add-operationRepo-onLoaded-Listener

Follow Up: add operationRepoLoadedListener and implement it to RecoverFromDroppe…

Author: jinliu9508

OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/IOperationRepo.kt

@@ -38,6 +38,8 @@ interface IOperationRepo {
      * Check if the queue contains a specific operation type
      */
     fun <T : Operation> containsInstanceOf(type: KClass<T>): Boolean
+
+    fun addOperationLoadedListener(handler: IOperationRepoLoadedListener)
 }
 
 // Extension function so the syntax containsInstanceOf<Operation>() can be used over

OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/IOperationRepoLoadedListener.kt

@@ -0,0 +1,5 @@
+package com.onesignal.core.internal.operations
+
+interface IOperationRepoLoadedListener {
+    fun onOperationRepoLoaded()
+}

OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt

@@ -1,11 +1,14 @@
 package com.onesignal.core.internal.operations.impl
 
+import com.onesignal.common.events.EventProducer
+import com.onesignal.common.events.IEventNotifier
 import com.onesignal.common.threading.WaiterWithValue
 import com.onesignal.core.internal.config.ConfigModelStore
 import com.onesignal.core.internal.operations.ExecutionResult
 import com.onesignal.core.internal.operations.GroupComparisonType
 import com.onesignal.core.internal.operations.IOperationExecutor
 import com.onesignal.core.internal.operations.IOperationRepo
+import com.onesignal.core.internal.operations.IOperationRepoLoadedListener
 import com.onesignal.core.internal.operations.Operation
 import com.onesignal.core.internal.startup.IStartableService
 import com.onesignal.core.internal.time.ITime
@@ -27,7 +30,7 @@ internal class OperationRepo(
     private val _configModelStore: ConfigModelStore,
     private val _time: ITime,
     private val _newRecordState: NewRecordsState,
-) : IOperationRepo, IStartableService {
+) : IOperationRepo, IStartableService, IEventNotifier<IOperationRepoLoadedListener> {
     internal class OperationQueueItem(
         val operation: Operation,
         val waiter: WaiterWithValue<Boolean>? = null,
@@ -49,6 +52,18 @@ internal class OperationRepo(
     private val waiter = WaiterWithValue<LoopWaiterMessage>()
     private var paused = false
     private var coroutineScope = CoroutineScope(newSingleThreadContext(name = "OpRepo"))
+    private val loadedSubscription: EventProducer<IOperationRepoLoadedListener> = EventProducer()
+
+    override val hasSubscribers: Boolean
+        get() = loadedSubscription.hasSubscribers
+
+    override fun unsubscribe(handler: IOperationRepoLoadedListener) {
+        loadedSubscription.unsubscribe(handler)
+    }
+
+    override fun subscribe(handler: IOperationRepoLoadedListener) {
+        loadedSubscription.subscribe(handler)
+    }
 
     /** *** Buckets ***
      * Purpose: Bucketing is a pattern we are using to help save network
@@ -86,6 +101,10 @@ internal class OperationRepo(
         }
     }
 
+    override fun addOperationLoadedListener(handler: IOperationRepoLoadedListener) {
+        subscribe(handler)
+    }
+
     override fun start() {
         paused = false
         coroutineScope.launch {
@@ -393,5 +412,6 @@ internal class OperationRepo(
                 operation.index,
             )
         }
+        loadedSubscription.fire { it.onOperationRepoLoaded() }
     }
 }

OneSignalSDK/onesignal/core/src/main/java/com/onesignal/user/internal/migrations/RecoverFromDroppedLoginBug.kt

@@ -3,6 +3,7 @@ package com.onesignal.user.internal.migrations
 import com.onesignal.common.IDManager
 import com.onesignal.core.internal.config.ConfigModelStore
 import com.onesignal.core.internal.operations.IOperationRepo
+import com.onesignal.core.internal.operations.IOperationRepoLoadedListener
 import com.onesignal.core.internal.operations.containsInstanceOf
 import com.onesignal.core.internal.startup.IStartableService
 import com.onesignal.debug.internal.logging.Logging
@@ -30,8 +31,12 @@ class RecoverFromDroppedLoginBug(
     private val _operationRepo: IOperationRepo,
     private val _identityModelStore: IdentityModelStore,
     private val _configModelStore: ConfigModelStore,
-) : IStartableService {
+) : IStartableService, IOperationRepoLoadedListener {
     override fun start() {
+        _operationRepo.addOperationLoadedListener(this)
+    }
+
+    override fun onOperationRepoLoaded() {
         if (isInBadState()) {
             Logging.warn(
                 "User with externalId:" +

OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRepoTests.kt

@@ -590,6 +590,24 @@ class OperationRepoTests : FunSpec({
         // Then
         result shouldBe null
     }
+
+    test("ensure onOperationRepoLoaded is called once loading is completed") {
+        // Given
+        val mocks = Mocks()
+        val spyListener = spyk<IOperationRepoLoadedListener>()
+
+        // When
+        mocks.operationRepo.addOperationLoadedListener(spyListener)
+        mocks.operationRepo.start()
+
+        // Then
+        mocks.operationRepo.hasSubscribers shouldBe true
+        coVerifyOrder {
+            mocks.operationRepo.subscribe(any())
+            mocks.operationModelStore.loadOperations()
+            spyListener.onOperationRepoLoaded()
+        }
+    }
 }) {
     companion object {
         private fun mockOperation(

OneSignalSDK/onesignal/core/src/test/java/com/onesignal/user/internal/migrations/RecoverFromDroppedLoginBugTests.kt

@@ -0,0 +1,48 @@
+package com.onesignal.user.internal.migrations
+
+import com.onesignal.core.internal.config.ConfigModelStore
+import com.onesignal.core.internal.operations.impl.OperationModelStore
+import com.onesignal.core.internal.operations.impl.OperationRepo
+import com.onesignal.core.internal.time.impl.Time
+import com.onesignal.mocks.MockHelper
+import com.onesignal.user.internal.operations.ExecutorMocks
+import io.kotest.core.spec.style.FunSpec
+import io.mockk.every
+import io.mockk.just
+import io.mockk.mockk
+import io.mockk.runs
+import io.mockk.spyk
+import io.mockk.verify
+
+class RecoverFromDroppedLoginBugTests : FunSpec({
+    test("ensure RecoverFromDroppedLoginBug receive onOperationRepoLoaded callback from operationRepo") {
+        // Given
+        val mockOperationModelStore = mockk<OperationModelStore>()
+        val mockConfigModelStore = mockk<ConfigModelStore>()
+        val operationRepo =
+            spyk(
+                OperationRepo(
+                    listOf(),
+                    mockOperationModelStore,
+                    mockConfigModelStore,
+                    Time(),
+                    ExecutorMocks.getNewRecordState(mockConfigModelStore),
+                ),
+            )
+        every { mockOperationModelStore.loadOperations() } just runs
+        every { mockOperationModelStore.list() } returns listOf()
+
+        val recovery = RecoverFromDroppedLoginBug(operationRepo, MockHelper.identityModelStore(), mockConfigModelStore)
+        every { recovery.onOperationRepoLoaded() } just runs
+
+        // When
+        operationRepo.start()
+        recovery.start()
+
+        // Then
+        verify {
+            operationRepo.subscribe(recovery)
+            recovery.onOperationRepoLoaded()
+        }
+    }
+})