如何组合输入依赖于其他输出的两个流？

Question

我有一个复杂的场景，其中一组相互依赖的coroutine flows 相互依赖并链接在一起：

viewModelScope.launch {
            repository.cacheAccount(person)
                .flatMapConcat { it->
                    Log.d(App.TAG, "[2] create account call (server)")
                    repository.createAccount(person)
                }
                .flatMapConcat { it ->
                    if (it is Response.Data) {
                        repository.cacheAccount(it.data)
                            .collect { it ->
                                // no op, just execute the command
                                Log.d(App.TAG, "account has been cached")
                            }
                    }
                    flow {
                        emit(it)
                    }
                }
                .catch { e ->
                    Log.d(App.TAG, "[3] get an exception in catch block")
                    Log.e(App.TAG, "Got an exception during network call", e)
                    state.update { state ->
                        val errors = state.errors + getErrorMessage(PersonRepository.Response.Error.Exception(e))
                        state.copy(errors = errors, isLoading = false)
                    }
                }
                .collect { it ->
                    Log.d(App.TAG, "[4] collect the result")
                    updateStateProfile(it)
                }
        }

1. 在本地磁盘上缓存一个帐户
2. 在后端创建一个账户
3. 在积极的情况下，将新创建的帐户缓存在本地磁盘中

   现在我必须向新的 API 端点添加更多调用，并且场景变得更加复杂。此端点是 ethereum chain。

   4a.正向场景，放入本地磁盘（缓存）发起的事务cacheRepository.createChainTx()

   4b.在负面情况下，只需进一步发出来自后端的响应

   4a.->5。在第二个端点注册用户repository.registerUser()

   6. 来自第二个端点的响应通过更新现有行放入缓存。除了异常之外，即使是负面情况也应该被缓存以更新 tx 的状态。

           viewModelScope.launch {
               lateinit var newTx: ITransaction
               cacheRepository.createChainTxAsFlow(RegisterUserTransaction(userWalletAddress = userWalletAddress))
                   .map { it ->
                       newTx= it
                       repository.registerUserOnSwapMarket(userWalletAddress)
                   }
                   .onEach { it -> preProcessResponse(it, newTx) }
                   .flowOn(backgroundDispatcher)
                   .collect { it -> processResponse(it) }
           }
   

   这是一个应该集成到第一个Flow chain 中的场景。

   问题是我在Flow chain 中看不到如何弄清楚。我可以在不链接的情况下重写代码，但它也带来了多种 if else 语句。

   你会如何做这个场景人类可读方式？

Answer 1

我将在过渡期得到这段代码：

        viewModelScope.launch(backgroundDispatcher) {
            try {
                var cachedPersonProfile = repository.cacheAccount(person)
                var createAccountResponse = repository.createAccount(person)
                when(createAccountResponse) {
                    is Response.Data -> {
                        repository.cacheAccount(createAccountResponse.data)
                        val cachedTx = cacheRepository.createChainTx(RegisterUserTransaction(userWalletAddress = person.userWalletAddress))
                        val chainTx = walletRepository.registerUserOnSwapMarket(userWalletAddress = person.userWalletAddress)
                        when(chainTx) {
                            is ru.home.swap.core.network.Response.Data -> {
                                if (chainTx.data.isStatusOK()) {
                                    cachedTx.status = TxStatus.TX_MINED
                                } else {
                                    cachedTx.status = TxStatus.TX_REVERTED
                                }
                            }
                            is ru.home.swap.core.network.Response.Error.Message -> {
                                cachedTx.status = TxStatus.TX_EXCEPTION
                            }
                            is ru.home.swap.core.network.Response.Error.Exception -> {
                                cachedTx.status = TxStatus.TX_EXCEPTION
                            }
                        }
                        cacheRepository.createChainTx(cachedTx)
                        withContext(Dispatchers.Main) {
                            state.update { state ->
                                if (cachedTx.status == TxStatus.TX_MINED) {
                                    state.copy(
                                        isLoading = false,
                                        profile = createAccountResponse.data,
                                        status = StateFlagV2.PROFILE
                                    )
                                } else {
                                    val txError = "Failed register the profile on chain with status ${TxStatus.TX_MINED}"
                                    state.copy(
                                        isLoading = false,
                                        errors = state.errors + txError
                                    )
                                }
                            }
                        }
                    }
                    else -> { updateStateProfile(createAccountResponse) }
                }
            } catch (ex: Exception) {
                withContext(Dispatchers.Main) {
                    state.update { state ->
                        val errors = state.errors + getErrorMessage(PersonRepository.Response.Error.Exception(ex))
                        state.copy(errors = errors, isLoading = false)
                    }
                }
            }
        }

如果您有更好的选择，请在帖子中分享作为答案。