Fixed bugs in Kotlin flow operators.

Kotlin/kotlinx.coroutines#2051

Author: denyshorman

src/main/kotlin/com/gitlab/dhorman/cryptotrader/api/PoloniexTraderApi.kt

@@ -14,6 +14,7 @@ import com.gitlab.dhorman.cryptotrader.trader.model.TranIntentMarket
 import com.gitlab.dhorman.cryptotrader.trader.model.TranIntentMarketCompleted
 import com.gitlab.dhorman.cryptotrader.trader.model.TranIntentMarketExtensions
 import com.gitlab.dhorman.cryptotrader.util.CsvGenerator
+import com.gitlab.dhorman.cryptotrader.util.first
 import io.swagger.annotations.ApiOperation
 import io.vavr.Tuple2
 import io.vavr.Tuple3

src/main/kotlin/com/gitlab/dhorman/cryptotrader/core/SimulatedPath.kt

@@ -8,6 +8,7 @@ import com.gitlab.dhorman.cryptotrader.service.poloniex.volume
 import com.gitlab.dhorman.cryptotrader.trader.model.TranIntentMarket
 import com.gitlab.dhorman.cryptotrader.trader.model.TranIntentMarketPartiallyCompleted
 import com.gitlab.dhorman.cryptotrader.trader.model.TranIntentMarketPredicted
+import com.gitlab.dhorman.cryptotrader.util.first
 import io.vavr.Tuple2
 import io.vavr.collection.Array
 import io.vavr.collection.Map
@@ -17,7 +18,6 @@ import io.vavr.kotlin.component2
 import io.vavr.kotlin.getOrNull
 import io.vavr.kotlin.tuple
 import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.first
 import java.math.BigDecimal
 import java.math.RoundingMode
 import java.util.*

src/main/kotlin/com/gitlab/dhorman/cryptotrader/service/poloniex/ExtendedPoloniexApi.kt

@@ -6,6 +6,7 @@ import com.gitlab.dhorman.cryptotrader.core.*
 import com.gitlab.dhorman.cryptotrader.service.poloniex.exception.DisconnectedException
 import com.gitlab.dhorman.cryptotrader.service.poloniex.model.*
 import com.gitlab.dhorman.cryptotrader.trader.core.AdjustedPoloniexBuySellAmountCalculator
+import com.gitlab.dhorman.cryptotrader.util.first
 import com.gitlab.dhorman.cryptotrader.util.share
 import io.netty.handler.timeout.ReadTimeoutException
 import io.netty.handler.timeout.WriteTimeoutException

src/main/kotlin/com/gitlab/dhorman/cryptotrader/trader/DelayedTradeProcessor.kt

@@ -11,6 +11,7 @@ import com.gitlab.dhorman.cryptotrader.trader.exception.CantMoveOrderSafelyExcep
 import com.gitlab.dhorman.cryptotrader.trader.exception.CompletePlaceMoveOrderLoop
 import com.gitlab.dhorman.cryptotrader.trader.exception.MoveNotRequiredException
 import com.gitlab.dhorman.cryptotrader.trader.exception.RepeatPlaceMoveOrderLoopAgain
+import com.gitlab.dhorman.cryptotrader.util.first
 import com.gitlab.dhorman.cryptotrader.util.returnLastIfNoValueWithinSpecifiedTime
 import io.netty.handler.timeout.ReadTimeoutException
 import io.netty.handler.timeout.WriteTimeoutException

src/main/kotlin/com/gitlab/dhorman/cryptotrader/trader/PathGenerator.kt

@@ -9,6 +9,7 @@ import com.gitlab.dhorman.cryptotrader.trader.core.AdjustedPoloniexBuySellAmount
 import com.gitlab.dhorman.cryptotrader.trader.dao.BlacklistedMarketsDao
 import com.gitlab.dhorman.cryptotrader.trader.dao.TransactionsDao
 import com.gitlab.dhorman.cryptotrader.trader.model.TranIntentMarketCompleted
+import com.gitlab.dhorman.cryptotrader.util.first
 import io.vavr.Tuple2
 import io.vavr.Tuple3
 import io.vavr.collection.List

src/main/kotlin/com/gitlab/dhorman/cryptotrader/trader/PoloniexTrader.kt

@@ -19,13 +19,13 @@ import com.gitlab.dhorman.cryptotrader.trader.model.TranIntentMarketCompleted
 import com.gitlab.dhorman.cryptotrader.trader.model.TranIntentMarketExtensions
 import com.gitlab.dhorman.cryptotrader.trader.model.TranIntentMarketPartiallyCompleted
 import com.gitlab.dhorman.cryptotrader.util.defaultTran
+import com.gitlab.dhorman.cryptotrader.util.first
 import com.gitlab.dhorman.cryptotrader.util.repeatableReadTran
 import io.vavr.Tuple2
 import io.vavr.collection.Array
 import io.vavr.kotlin.*
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.Channel
-import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.reactive.collect
 import mu.KotlinLogging
 import org.springframework.beans.factory.annotation.Qualifier

src/main/kotlin/com/gitlab/dhorman/cryptotrader/trader/TransactionIntent.kt

@@ -31,7 +31,10 @@ import io.vavr.collection.List
 import io.vavr.kotlin.*
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.Channel
-import kotlinx.coroutines.flow.*
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
 import mu.KotlinLogging

src/main/kotlin/com/gitlab/dhorman/cryptotrader/trader/model/TraderModels.kt

@@ -11,10 +11,10 @@ import com.gitlab.dhorman.cryptotrader.service.poloniex.model.Currency
 import com.gitlab.dhorman.cryptotrader.service.poloniex.model.CurrencyType
 import com.gitlab.dhorman.cryptotrader.service.poloniex.model.OrderType
 import com.gitlab.dhorman.cryptotrader.trader.core.AdjustedBuySellAmountCalculator
+import com.gitlab.dhorman.cryptotrader.util.first
 import io.vavr.collection.Array
 import io.vavr.kotlin.component1
 import io.vavr.kotlin.component2
-import kotlinx.coroutines.flow.first
 import java.math.BigDecimal
 
 @JsonTypeInfo(

src/main/kotlin/com/gitlab/dhorman/cryptotrader/util/FlowOperators.kt

@@ -1,8 +1,5 @@
 package com.gitlab.dhorman.cryptotrader.util
 
-import io.vavr.Tuple2
-import io.vavr.collection.HashMap
-import io.vavr.collection.Map
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.ProducerScope
@@ -28,7 +25,6 @@ private open class ShareOperator<T>(
     private val gracePeriod: Duration? = null,
     private val scope: CoroutineScope? = null
 ) {
-    private var subscribers = 0L
     private val subscriberChannels = mutableSetOf<SendChannel<T>>()
     private val subscriberChannelsMutex = Mutex()
     private val upstreamSubscriptionLock = Mutex()
@@ -57,8 +53,8 @@ private open class ShareOperator<T>(
                     }
                 }
             } catch (e: CancellationException) {
-                if (subscriberChannels.size > 0) {
-                    logger.warn("upstream cancelled when subscriberChannels (${subscriberChannels.size}) variable is not empty.")
+                if (subscriberChannels.isNotEmpty()) {
+                    logger.warn("$upstream cancelled when subscriberChannels (${subscriberChannels.size}) variable is not empty.")
                 }
                 throw e
             }
@@ -123,7 +119,7 @@ private open class ShareOperator<T>(
     val shareOperator = channelFlow<T> {
         try {
             upstreamSubscriptionLock.withLock {
-                if (++subscribers == 1L) {
+                if (subscriberChannels.isEmpty()) {
                     cancelGracePeriodTimerJob()
                     processQueueAndSubscribeSelf()
                     subscribeToUpstream()
@@ -138,7 +134,7 @@ private open class ShareOperator<T>(
                 upstreamSubscriptionLock.withLock {
                     unsubscribeSelf()
 
-                    if (--subscribers == 0L) {
+                    if (subscriberChannels.isEmpty()) {
                         val launched = launchGracePeriodTimerJob()
 
                         if (!launched) {
@@ -197,18 +193,21 @@ fun <T> Channel<T>.buffer(scope: CoroutineScope, timespan: Duration): Channel<Li
     return upstream
 }
 
-fun <K, V> flowFromMap(map: Map<K, V>): Flow<Tuple2<K, V>> = flow {
-    map.forEach { emit(it) }
-}
-
-suspend fun <K, V> Flow<Tuple2<K, V>>.collectMap(): Map<K, V> {
-    var map = HashMap.empty<K, V>()
+private object AbortException : Throwable("", null, true, false)
 
-    collect {
-        map = map.put(it)
+suspend fun <T> Flow<T>.first(): T {
+    var result: T? = null
+    try {
+        collect { value ->
+            result = value
+            throw AbortException
+        }
+    } catch (e: AbortException) {
+        // Do nothing
     }
 
-    return map
+    if (result == null) throw NoSuchElementException("Expected at least one element")
+    return result!!
 }
 
 suspend fun <T> Flow<T>.firstOrNull(): T? {
@@ -217,9 +216,9 @@ suspend fun <T> Flow<T>.firstOrNull(): T? {
     try {
         collect {
             result = it
-            throw CancellationException()
+            throw AbortException
         }
-    } catch (e: CancellationException) {
+    } catch (e: AbortException) {
         // Ignore
     }
 

src/test/kotlin/com/gitlab/dhorman/cryptotrader/service/poloniex/ExtendedPoloniexApiTest.kt

@@ -2,12 +2,12 @@ package com.gitlab.dhorman.cryptotrader.service.poloniex
 
 import com.gitlab.dhorman.cryptotrader.core.toMarket
 import com.gitlab.dhorman.cryptotrader.util.TestClock
+import com.gitlab.dhorman.cryptotrader.util.first
 import com.gitlab.dhorman.cryptotrader.util.firstOrNull
 import io.vavr.kotlin.getOrNull
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.collect
-import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.onStart
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking

src/test/kotlin/com/gitlab/dhorman/cryptotrader/trader/PathGeneratorTest.kt

@@ -5,19 +5,21 @@ import com.gitlab.dhorman.cryptotrader.service.poloniex.ExtendedPoloniexApi
 import com.gitlab.dhorman.cryptotrader.service.poloniex.core.PoloniexBuySellAmountCalculator
 import com.gitlab.dhorman.cryptotrader.service.poloniex.model.Amount
 import com.gitlab.dhorman.cryptotrader.service.poloniex.model.CurrencyType
-import com.gitlab.dhorman.cryptotrader.trader.PathGenerator
 import com.gitlab.dhorman.cryptotrader.trader.dao.TransactionsDao
-import com.gitlab.dhorman.cryptotrader.trader.findOne
 import com.gitlab.dhorman.cryptotrader.trader.model.TranIntentMarketCompleted
 import com.gitlab.dhorman.cryptotrader.trader.model.TranIntentMarketPartiallyCompleted
 import com.gitlab.dhorman.cryptotrader.trader.model.TranIntentMarketPredicted
 import com.gitlab.dhorman.cryptotrader.util.CsvGenerator
+import com.gitlab.dhorman.cryptotrader.util.first
 import com.nhaarman.mockitokotlin2.mock
 import com.nhaarman.mockitokotlin2.whenever
 import io.vavr.collection.Array
 import io.vavr.kotlin.*
 import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.flow.*
+import kotlinx.coroutines.flow.asFlow
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.flatMapMerge
+import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.runBlocking
 import mu.KotlinLogging
 import org.junit.jupiter.api.Test

src/test/kotlin/com/gitlab/dhorman/cryptotrader/trader/dao/BlacklistedMarketsDaoTest.kt

@@ -2,8 +2,8 @@ package com.gitlab.dhorman.cryptotrader.trader.dao
 
 import com.gitlab.dhorman.cryptotrader.core.Market
 import com.gitlab.dhorman.cryptotrader.util.TestClock
+import com.gitlab.dhorman.cryptotrader.util.first
 import kotlinx.coroutines.delay
-import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.runBlocking
 import org.junit.jupiter.api.*
 import org.springframework.beans.factory.annotation.Autowired