Fixed balance sync (#474)

* Fixed balance sync * Fixed suspending in flow * Moved builder to next lines
novasamatech · Oct 3, 2022 · 45b4382 · 45b4382
1 parent 004b4e5
commit 45b4382
Show file tree

Hide file tree

Showing 2 changed files with 22 additions and 29 deletions.
diff --git a/common/src/test/java/io/novafoundation/nova/common/utils/FlowExtKtTest.kt b/common/src/test/java/io/novafoundation/nova/common/utils/FlowExtKtTest.kt
@@ -43,10 +43,10 @@ class FlowExtKtTest {
 
             performTest(
                 first = listOf(1, 2),
-                second =  listOf(2, 3),
+                second = listOf(2, 3),
                 expectedDiff = ListDiff(
                     removed = listOf(1),
-                    addedOrModified =  listOf(3),
+                    addedOrModified = listOf(3),
                     all = listOf(2, 3)
                 )
             )

diff --git a/...nova/feature_wallet_impl/data/network/blockchain/updaters/balance/BalancesUpdateSystem.kt b/...nova/feature_wallet_impl/data/network/blockchain/updaters/balance/BalancesUpdateSystem.kt
@@ -12,10 +12,9 @@ import io.novafoundation.nova.runtime.multiNetwork.chain.model.Chain
 import io.novafoundation.nova.runtime.multiNetwork.getSocket
 import jp.co.soramitsu.fearless_utils.wsrpc.request.runtime.storage.subscribeUsing
 import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.catch
-import kotlinx.coroutines.flow.emitAll
+import kotlinx.coroutines.flow.emptyFlow
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.flatMapLatest
 import kotlinx.coroutines.flow.flow
@@ -30,39 +29,33 @@ class BalancesUpdateSystem(
     private val accountUpdateScope: AccountUpdateScope
 ) : UpdateSystem {
 
-    @OptIn(ExperimentalCoroutinesApi::class)
     override fun start(): Flow<Updater.SideEffect> {
         return accountUpdateScope.invalidationFlow().flatMapLatest {
             val chains = chainRegistry.currentChains.first()
 
-            val mergedFlow = chains.map { chain ->
-                flow {
-                    val socket = chainRegistry.getSocket(chain.id)
-                    val subscriptionBuilder = StorageSubscriptionBuilder.create(socket)
-
-                    val updaters: List<Updater> = listOf(
-                        paymentUpdaterFactory.create(chain),
-                        balanceLocksUpdater.create(chain)
-                    )
-
-                    kotlin.runCatching {
-                        updaters.map { it.listenForUpdates(subscriptionBuilder) }
-                            .merge()
-                            .catch { logError(chain, it) }
-                    }.onSuccess { updatersFlow ->
-                        val cancellable = socket.subscribeUsing(subscriptionBuilder.build())
+            chains.map { balanceChainUpdaters(it) }
+                .merge()
+        }.flowOn(Dispatchers.Default)
+    }
 
-                        updatersFlow.onCompletion { cancellable.cancel() }
+    private suspend fun balanceChainUpdaters(chain: Chain): Flow<Updater.SideEffect> {
+        return flow {
+            val socket = chainRegistry.getSocket(chain.id)
+            val subscriptionBuilder = StorageSubscriptionBuilder.create(socket)
 
-                        emitAll(updatersFlow)
-                    }.onFailure {
-                        logError(chain, it)
-                    }
+            val updaters: List<Updater> = listOf(paymentUpdaterFactory.create(chain), balanceLocksUpdater.create(chain))
+            val sideEffectFlows = updaters.map { updater ->
+                try {
+                    updater.listenForUpdates(subscriptionBuilder).catch { logError(chain, it) }
+                } catch (e: Exception) {
+                    emptyFlow()
                 }
-            }.merge()
+            }
 
-            mergedFlow
-        }.flowOn(Dispatchers.Default)
+            val cancellable = socket.subscribeUsing(subscriptionBuilder.build())
+            sideEffectFlows.merge()
+                .onCompletion { cancellable.cancel() }
+        }
     }
 
     private fun logError(chain: Chain, error: Throwable) {