Distractic
04/05/2023, 9:27 AMpublic suspend inline fun <T> connect(crossinline body: suspend (RedisCoroutinesCommands<ByteArray, ByteArray>) -> T): T {
contract {
callsInPlace(body, InvocationKind.EXACTLY_ONCE)
}
return connectionManager.poolStateful.acquire {
val newConnection = it.coroutines()
body(newConnection)
}
}
And my code works.
I changed to store the current connection like :
public suspend inline fun <T> connect(crossinline body: suspend (RedisCoroutinesCommands<ByteArray, ByteArray>) -> T): T {
contract {
callsInPlace(body, InvocationKind.EXACTLY_ONCE)
}
val currentContext = currentCoroutineContext()
val connection = currentContext[RedisConnection]
if (connection != null) return body(connection.connection)
return connectionManager.poolStateful.acquire {
val newConnection = it.coroutines()
withContext(currentContext + RedisConnection(newConnection)) {
body(newConnection)
}
}
}
And I have a new error:
Flow invariant is violated:
Flow was collected in [RunningInRunTest, kotlinx.coroutines.test.TestCoroutineScheduler@458031da, kotlinx.coroutines.test.TestScopeKt$TestScope$$inlined$CoroutineExceptionHandler$1@7be94cd6, CoroutineId(9), "coroutine#9":StandaloneCoroutine{Active}@20cff21e, StandardTestDispatcher[scheduler=kotlinx.coroutines.test.TestCoroutineScheduler@458031da]],
but emission happened in [RunningInRunTest, kotlinx.coroutines.test.TestCoroutineScheduler@458031da, kotlinx.coroutines.test.TestScopeKt$TestScope$$inlined$CoroutineExceptionHandler$1@7be94cd6, CoroutineId(9), RedisConnectionImpl@463045fb, kotlinx.coroutines.UndispatchedMarker@27ab206, "coroutine#9":UndispatchedCoroutine{Active}@3344d163, StandardTestDispatcher[scheduler=kotlinx.coroutines.test.TestCoroutineScheduler@458031da]].
Please refer to 'flow' documentation or use 'flowOn' instead
How can I use the new current coroutine context to handle a flow ?