Kotlinx.coroutines: Provide `associate` terminal operators

[jcornaz]

Hi,

Today we realized that despite the terminal operators toList and toSet being provided for Flow, there isn't anything to create a map from a Flow.

Could we have associate, associateBy and associateWith for Flow, as we have for Sequence?

[qwwdfsad]

Definitely, we can implement these operators.

Initially, we've tried to avoid having the same variety of Flow operators as Sequence has because Flow has slightly different use-cases though it looks like a sequence. But if people keep lacking these operators for Flow, we will add them.

[fvasco]

Please consider Flow<Pair<K, V>>.toMap(), too.

[Globegitter]

Yep, right now we have to workaround by doing .toList().toMap() - would be nice to be able to convert to a map out of the box.

[Xaseron]

I'have came here because i find the toList().toMap() workaround kinda ugly.
Should we create a separate issue?

[elizarov]

Please, describe your specific use-cases in this issue.

[Xaseron]

i have functions that returns a Flow<Pair<K, V>>.
To convert this into a Map i have to use toList().toMap().
Having a terminal operator toMap on flow would elimante an intermediate collection created by toList and thus increase the performance.

[fvasco]

@Xaseron,
you should to explain a bit deeper your use case.

However you can consider to include an extension function like:

suspend fun <K, V> Flow<Pair<K, V>>.toMap(): Map<K, V> {
    val map = mutableMapOf<K, V>()
    collect { (k, v) ->
        map[k] = v
    }
    return map
}

[Xaseron]

Having this in the stdlib would make it even with collection api.
Isn't this a goal?

[elizarov]

@Xaseron So, what's your use-case? Can you elaborate, please.

[Xaseron]

I have this method where i use a Mongodb Aggregation Pipeline to gather all values for specific year/month in a list. Afterwards i'm calling collect on this publisher to map each year/month to the average value of this list and emit it.

    fun overallTrends(metricName: String) = flow {
        val mongoList = col.aggregate<OverallTrend>(
                """
[
    { $ group:
        {
             _id: {
                year: "$ metaData.reportingYear",
                month: "$ metaData.reportingMonth"
            },
            values: { $ push: "$ $metricName" }
        }
    }
]""".formatJson()
        )
        mongoList.collect { trend ->
            val yearMonth = trend.id.toYearMonth()
            emit(yearMonth to trend.averageValue)
        }
    }

I want to call a terminal operation on this flow to get a Map.
Therefore currently i have to it this way:
val overallTrendsMap = mongoMetricsService.overallTrends(mongoMetricName).toList().toMap()
I would be more performant and nicer to read if i could directly call it like this:
val overallTrendsMap = mongoMetricsService.overallTrends(mongoMetricName).toMap()

[elizarov]

Why do you define your overallTrends function to return a flow if you plan to get get a map out of it? Wouldn't it be easier to define it as a suspending function that returns a map in the first place?