refactor[collab]: split token flow and stateflow

GitOrigin-RevId: ed257b3bfc36c80ebb2ea52777a4c799c76964c8

platform/collaboration-tools/src/com/intellij/collaboration/auth/AccountManager.kt

@@ -1,6 +1,7 @@
 // Copyright 2000-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license that can be found in the LICENSE file.
 package com.intellij.collaboration.auth
 
+import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.StateFlow
 
@@ -42,7 +43,12 @@ interface AccountManager<A : Account, Cred> {
 
   /**
    * Flow of account credentials
-   * Will be closed when the account is removed
    */
-  fun getCredentialsFlow(account: A, withCurrent: Boolean = true): Flow<Cred?>
+  fun getCredentialsFlow(account: A): Flow<Cred?>
+
+  /**
+   * Flow of account credentials with the latest state
+   * Credentials are acquired and updated under [scope]
+   */
+  suspend fun getCredentialsState(scope: CoroutineScope, account: A): StateFlow<Cred?>
 }

platform/collaboration-tools/src/com/intellij/collaboration/auth/AccountManagerBase.kt

@@ -4,10 +4,13 @@ package com.intellij.collaboration.auth
 import com.intellij.collaboration.async.disposingScope
 import com.intellij.openapi.Disposable
 import com.intellij.openapi.diagnostic.Logger
-import kotlinx.coroutines.*
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.NonCancellable
 import kotlinx.coroutines.flow.*
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
+import kotlinx.coroutines.withContext
 
 /**
  * Base class for account management application service
@@ -115,34 +118,26 @@ abstract class AccountManagerBase<A : Account, Cred : Any>(
 
   override suspend fun findCredentials(account: A): Cred? = persistentCredentials.retrieveCredentials(account)
 
-  override fun getCredentialsFlow(account: A, withCurrent: Boolean): Flow<Cred?> = channelFlow {
-    mutex.withLock {
-      // subscribe to map updates first
-      launch(start = CoroutineStart.UNDISPATCHED) {
-        accountsEventsFlow.collect {
-          when (it) {
-            is Event.AccountsAddedOrUpdated -> {
-              send(it.map[account])
-            }
-            is Event.AccountsRemoved -> {
-              if (account in it.accounts) {
-                cancel()
-              }
-            }
+  override suspend fun getCredentialsState(scope: CoroutineScope, account: A): StateFlow<Cred?> =
+    withContext(scope.coroutineContext + Dispatchers.Default) {
+      mutex.withLock {
+        getCredentialsFlow(account).stateIn(scope, SharingStarted.Eagerly, findCredentials(account))
+      }
+    }
+
+  override fun getCredentialsFlow(account: A): Flow<Cred?> =
+    accountsEventsFlow.transform {
+      when (it) {
+        is Event.AccountsAddedOrUpdated -> {
+          emit(it.map[account])
+        }
+        is Event.AccountsRemoved -> {
+          if (account in it.accounts) {
+            emit(null)
           }
         }
       }
-      if (withCurrent) {
-        try {
-          send(persistentCredentials.retrieveCredentials(account))
-        }
-        catch (e: Exception) {
-          logger.warn("Failed to retrieve credentials", e)
-          send(null)
-        }
-      }
-    }
-  }.flowOn(Dispatchers.Default)
+    }.flowOn(Dispatchers.Default)
 
   override fun dispose() = Unit
 

platform/collaboration-tools/src/com/intellij/collaboration/auth/AccountUrlAuthenticationFailuresHolder.kt

@@ -17,7 +17,7 @@ class AccountUrlAuthenticationFailuresHolder<A : Account>(
   fun markFailed(account: A, url: String) {
     storeMap.computeIfAbsent(account) {
       cs.launch {
-        accountManager().getCredentialsFlow(account, false).first()
+        accountManager().getCredentialsFlow(account).first()
         storeMap.remove(account)
       }
       ContainerUtil.newConcurrentSet()

platform/collaboration-tools/src/com/intellij/collaboration/auth/ui/LazyLoadingAccountsDetailsProvider.kt

@@ -134,7 +134,7 @@ fun <A : Account> LazyLoadingAccountsDetailsProvider<A, *>.cancelOnRemoval(scope
       coroutineScope {
         for (account in it) {
           launch {
-            accountManager.getCredentialsFlow(account, false).collect {
+            accountManager.getCredentialsFlow(account).collect {
               clearDetails(account)
             }
           }

platform/collaboration-tools/test/com/intellij/collaboration/auth/AccountUrlAuthenticationFailuresHolderTest.kt

@@ -17,7 +17,7 @@ class AccountUrlAuthenticationFailuresHolderTest {
   fun `test not failed on token change`() = runTest(UnconfinedTestDispatcher()) {
     val credsFlow = MutableSharedFlow<String>()
     val accountManager = mock<AccountManager<Account, String>> {
-      on(it.getCredentialsFlow(any(), any())).thenReturn(credsFlow)
+      on(it.getCredentialsFlow(any())).thenReturn(credsFlow)
     }
     val holder = AccountUrlAuthenticationFailuresHolder(this) {
       accountManager

plugins/git4idea/src/git4idea/remote/hosting/ui/RepositoryAndAccountSelectorViewModelBase.kt

@@ -8,7 +8,9 @@ import com.intellij.collaboration.util.URIUtil
 import git4idea.remote.hosting.HostedGitRepositoriesManager
 import git4idea.remote.hosting.HostedGitRepositoryMapping
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.FlowPreview
+import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.flow.*
 
 abstract class RepositoryAndAccountSelectorViewModelBase<M : HostedGitRepositoryMapping, A : ServerAccount>(
@@ -32,14 +34,15 @@ abstract class RepositoryAndAccountSelectorViewModelBase<M : HostedGitRepository
 
   final override val accountSelectionState = MutableStateFlow<A?>(null)
 
+  @OptIn(ExperimentalCoroutinesApi::class)
   final override val missingCredentialsState: StateFlow<Boolean> =
-    channelFlow {
-      accountSelectionState.collectLatest {
-        if(it == null) {
-          send(false)
-        } else {
-          accountManager.getCredentialsFlow(it, true).collect { creds ->
-            send(creds == null)
+    accountSelectionState.transformLatest {
+      if(it == null) {
+        emit(false)
+      } else {
+        coroutineScope {
+          accountManager.getCredentialsState(this, it).collect { creds ->
+            emit(creds == null)
           }
         }
       }

plugins/github/src/org/jetbrains/plugins/github/authentication/GithubAuthenticationManager.kt

@@ -47,7 +47,7 @@ class GithubAuthenticationManager internal constructor() {
         listener.onAccountListChanged(prev, current)
         current.forEach { acc ->
           async {
-            accountManager.getCredentialsFlow(acc, false).collectLatest {
+            accountManager.getCredentialsFlow(acc).collectLatest {
               listener.onAccountCredentialsChanged(acc)
             }
           }

plugins/github/src/org/jetbrains/plugins/github/pullrequest/ui/toolwindow/GHRepositoryConnectionManager.kt

@@ -40,7 +40,7 @@ internal class GHRepositoryConnectionManager(scope: CoroutineScope,
             }
           }.collectLatest {
             coroutineScope {
-              accountManager.getCredentialsFlow(account).collectLatest { token ->
+              accountManager.getCredentialsState(this, account).collectLatest { token ->
                 if (token == null) {
                   throw CancellationException()
                 }

plugins/github/src/org/jetbrains/plugins/github/ui/cloneDialog/GHCloneDialogExtensionComponentBase.kt

@@ -375,7 +375,7 @@ internal abstract class GHCloneDialogExtensionComponentBase(
             if (!currentAccounts.contains(it)) {
               model.add(it)
               async {
-                accountManager.getCredentialsFlow(it, false).collect { _ ->
+                accountManager.getCredentialsFlow(it).collect { _ ->
                   model.contentsChanged(it)
                 }
               }

plugins/gitlab/src/org/jetbrains/plugins/gitlab/api/GitLabProjectConnectionManager.kt

@@ -41,7 +41,7 @@ internal class GitLabProjectConnectionManagerImpl(scope: CoroutineScope,
             }
           }.collectLatest {
             coroutineScope {
-              accountManager.getCredentialsFlow(account).collectLatest { token ->
+              accountManager.getCredentialsState(this, account).collectLatest { token ->
                 if (token == null) {
                   throw CancellationException()
                 }

plugins/gitlab/test/org/jetbrains/plugins/gitlab/mergerequest/ui/GitLabRepositoryAndAccountSelectorViewModelTest.kt

@@ -22,10 +22,7 @@ import org.junit.Test
 import org.mockito.Mock
 import org.mockito.junit.MockitoJUnit
 import org.mockito.junit.MockitoRule
-import org.mockito.kotlin.doAnswer
-import org.mockito.kotlin.doReturn
-import org.mockito.kotlin.mock
-import org.mockito.kotlin.whenever
+import org.mockito.kotlin.*
 
 @OptIn(ExperimentalCoroutinesApi::class)
 internal class GitLabRepositoryAndAccountSelectorViewModelTest {
@@ -62,6 +59,7 @@ internal class GitLabRepositoryAndAccountSelectorViewModelTest {
 
     val account = GitLabAccount(name = "test", server = GitLabServerPath.DEFAULT_SERVER)
     whenever(accountManager.accountsState) doReturn MutableStateFlow(setOf(account))
+    whenever(accountManager.getCredentialsState(any(), any())) doReturn MutableStateFlow("")
 
     val scope = childScope(Dispatchers.Main)
     val vm = GitLabRepositoryAndAccountSelectorViewModel(scope, connectionManager, projectManager, accountManager)