Skip to content

Commit a7a0dde

Browse files
qwwdfsadqwwdfsad
committed
Scan and emitAll operators
Fixes #1094
1 parent 1c16c72 commit a7a0dde

File tree

7 files changed

+136
-16
lines changed

7 files changed

+136
-16
lines changed

‎binary-compatibility-validator/reference-public-api/kotlinx-coroutines-core.txt

+5
Original file line numberDiff line numberDiff line change
@@ -816,6 +816,7 @@ public final class kotlinx/coroutines/flow/FlowKt {
816816
public static final fun distinctUntilChangedBy (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function1;)Lkotlinx/coroutines/flow/Flow;
817817
public static final fun drop (Lkotlinx/coroutines/flow/Flow;I)Lkotlinx/coroutines/flow/Flow;
818818
public static final fun dropWhile (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
819+
public static final fun emitAll (Lkotlinx/coroutines/flow/FlowCollector;Lkotlinx/coroutines/flow/Flow;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
819820
public static final fun emptyFlow ()Lkotlinx/coroutines/flow/Flow;
820821
public static final fun filter (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
821822
public static final fun filterNot (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
@@ -848,6 +849,8 @@ public final class kotlinx/coroutines/flow/FlowKt {
848849
public static final fun retry (Lkotlinx/coroutines/flow/Flow;ILkotlin/jvm/functions/Function1;)Lkotlinx/coroutines/flow/Flow;
849850
public static synthetic fun retry$default (Lkotlinx/coroutines/flow/Flow;ILkotlin/jvm/functions/Function1;ILjava/lang/Object;)Lkotlinx/coroutines/flow/Flow;
850851
public static final fun sample (Lkotlinx/coroutines/flow/Flow;J)Lkotlinx/coroutines/flow/Flow;
852+
public static final fun scan (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function0;Lkotlin/jvm/functions/Function3;)Lkotlinx/coroutines/flow/Flow;
853+
public static final fun scan (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function3;)Lkotlinx/coroutines/flow/Flow;
851854
public static final fun single (Lkotlinx/coroutines/flow/Flow;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
852855
public static final fun singleOrNull (Lkotlinx/coroutines/flow/Flow;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
853856
public static final fun switchMap (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
@@ -867,13 +870,15 @@ public final class kotlinx/coroutines/flow/MigrationKt {
867870
public static final fun BehaviourSubject ()Ljava/lang/Object;
868871
public static final fun PublishSubject ()Ljava/lang/Object;
869872
public static final fun ReplaySubject ()Ljava/lang/Object;
873+
public static final fun compose (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function1;)Lkotlinx/coroutines/flow/Flow;
870874
public static final fun concatMap (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function1;)Lkotlinx/coroutines/flow/Flow;
871875
public static final fun flatMap (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
872876
public static final fun flatten (Lkotlinx/coroutines/flow/Flow;)Lkotlinx/coroutines/flow/Flow;
873877
public static final fun merge (Lkotlinx/coroutines/flow/Flow;)Lkotlinx/coroutines/flow/Flow;
874878
public static final fun observeOn (Lkotlinx/coroutines/flow/Flow;Lkotlin/coroutines/CoroutineContext;)Lkotlinx/coroutines/flow/Flow;
875879
public static final fun onErrorResume (Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;)Lkotlinx/coroutines/flow/Flow;
876880
public static final fun publishOn (Lkotlinx/coroutines/flow/Flow;Lkotlin/coroutines/CoroutineContext;)Lkotlinx/coroutines/flow/Flow;
881+
public static final fun skip (Lkotlinx/coroutines/flow/Flow;I)Lkotlinx/coroutines/flow/Flow;
877882
public static final fun subscribe (Lkotlinx/coroutines/flow/Flow;)V
878883
public static final fun subscribe (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function1;)V
879884
public static final fun subscribe (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function1;)V

‎kotlinx-coroutines-core/common/src/flow/Migration.kt

+10
Original file line numberDiff line numberDiff line change
@@ -230,3 +230,13 @@ public fun <T> Flow<Flow<T>>.flatten(): Flow<T> = error("Should not be called")
230230
replaceWith = ReplaceWith("let(transformer)")
231231
)
232232
public fun <T, R> Flow<T>.compose(transformer: Flow<T>.() -> Flow<R>): Flow<R> = error("Should not be called")
233+
234+
/**
235+
* @suppress
236+
*/
237+
@Deprecated(
238+
level = DeprecationLevel.ERROR,
239+
message = "Kotlin analogue of 'skip' is 'drop'",
240+
replaceWith = ReplaceWith("drop(count)")
241+
)
242+
public fun <T> Flow<T>.skip(count: Int): Flow<T> = error("Should not be called")

‎kotlinx-coroutines-core/common/src/flow/operators/Errors.kt

+1-3
Original file line numberDiff line numberDiff line change
@@ -27,9 +27,7 @@ public fun <T> Flow<T>.onErrorCollect(
2727
predicate: ExceptionPredicate = ALWAYS_TRUE
2828
): Flow<T> = collectSafely { e ->
2929
if (!predicate(e)) throw e
30-
fallback.collect { value ->
31-
emit(value)
32-
}
30+
emitAll(fallback)
3331
}
3432

3533
/**

‎kotlinx-coroutines-core/common/src/flow/operators/Merge.kt

+3-13
Original file line numberDiff line numberDiff line change
@@ -24,11 +24,7 @@ import kotlinx.coroutines.flow.unsafeFlow as flow
2424
*/
2525
@FlowPreview
2626
public fun <T, R> Flow<T>.flatMapConcat(transform: suspend (value: T) -> Flow<R>): Flow<R> = flow {
27-
collect { value ->
28-
transform(value).collect { innerValue ->
29-
emit(innerValue)
30-
}
31-
}
27+
collect { value -> emitAll(transform(value)) }
3228
}
3329

3430
/**
@@ -72,11 +68,7 @@ public fun <T, R> Flow<T>.flatMapMerge(concurrency: Int = 16, bufferSize: Int =
7268
*/
7369
@FlowPreview
7470
public fun <T> Flow<Flow<T>>.flattenConcat(): Flow<T> = flow {
75-
collect { value ->
76-
value.collect { innerValue ->
77-
emit(innerValue)
78-
}
79-
}
71+
collect { value -> emitAll(value) }
8072
}
8173

8274
/**
@@ -118,9 +110,7 @@ public fun <T, R> Flow<T>.switchMap(transform: suspend (value: T) -> Flow<R>): F
118110
previousFlow?.cancelAndJoin()
119111
// Undispatched to have better user experience in case of synchronous flows
120112
previousFlow = launch(start = CoroutineStart.UNDISPATCHED) {
121-
transform(value).collect { innerValue ->
122-
emit(innerValue)
123-
}
113+
emitAll(transform(value))
124114
}
125115
}
126116
}

‎kotlinx-coroutines-core/common/src/flow/operators/Transform.kt

+43
Original file line numberDiff line numberDiff line change
@@ -4,10 +4,12 @@
44

55
@file:JvmMultifileClass
66
@file:JvmName("FlowKt")
7+
@file:Suppress("UNCHECKED_CAST")
78

89
package kotlinx.coroutines.flow
910

1011
import kotlinx.coroutines.*
12+
import kotlinx.coroutines.flow.internal.NullSurrogate
1113
import kotlin.jvm.*
1214
import kotlinx.coroutines.flow.unsafeFlow as flow
1315

@@ -97,3 +99,44 @@ public fun <T> Flow<T>.onEach(action: suspend (T) -> Unit): Flow<T> = flow {
9799
emit(value)
98100
}
99101
}
102+
103+
/**
104+
* Reduces the given flow with [operation], emitting every intermediate result, including initial value.
105+
* The first element is takes as initial value for operation accumulator.
106+
* For example:
107+
* ```
108+
* flowOf(1, 2, 3, 4).scan { (v1, v2) -> v1 + v2 }.toList()
109+
* ```
110+
* will produce `[1, 3, 6, 10]`
111+
*/
112+
@FlowPreview
113+
public fun <T> Flow<T>.scan(operation: suspend (accumulator: T, value: T) -> T): Flow<T> = flow {
114+
var accumulator: Any? = NullSurrogate
115+
collect { value ->
116+
accumulator = if (accumulator === NullSurrogate) {
117+
value
118+
} else {
119+
operation(accumulator as T, value)
120+
}
121+
emit(accumulator as T)
122+
}
123+
}
124+
125+
/**
126+
* Reduces the given flow with [operation], emitting every intermediate result, including initial value.
127+
* An initial value is provided lazily by [initialSupplier] and is always immediately emitted.
128+
* For example:
129+
* ```
130+
* flowOf(1, 2, 3).scan(::emptyList) { acc: List<Int>, value -> acc + value }.toList()
131+
* ```
132+
* will produce `[], [1], [1, 2], [1, 2, 3]]`.
133+
*/
134+
@FlowPreview
135+
public fun <T, R> Flow<T>.scan(initialSupplier: () -> R, @BuilderInference operation: suspend (accumulator: R, value: T) -> R): Flow<R> = flow {
136+
var accumulator: R = initialSupplier()
137+
emit(accumulator)
138+
collect { value ->
139+
accumulator = operation(accumulator, value)
140+
emit(accumulator)
141+
}
142+
}

‎kotlinx-coroutines-core/common/src/flow/terminal/Collect.kt

+6
Original file line numberDiff line numberDiff line change
@@ -32,3 +32,9 @@ public suspend inline fun <T> Flow<T>.collect(crossinline action: suspend (value
3232
collect(object : FlowCollector<T> {
3333
override suspend fun emit(value: T) = action(value)
3434
})
35+
36+
/**
37+
* Collects all the values from the given [flow] and emits them to the collector.
38+
* Shortcut for `flow.collect { value -> emit(value) }`.
39+
*/
40+
public suspend inline fun <T> FlowCollector<T>.emitAll(flow: Flow<T>) = flow.collect { value -> emit(value) }

‎kotlinx-coroutines-core/common/test/flow/operators/ScanTest.kt

+68
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,68 @@
1+
/*
2+
* Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
3+
*/
4+
5+
package kotlinx.coroutines.flow
6+
7+
import kotlinx.coroutines.*
8+
import kotlinx.coroutines.channels.*
9+
import kotlin.test.*
10+
11+
class ScanTest : TestBase() {
12+
@Test
13+
fun testScan() = runTest {
14+
val flow = flowOf(1, 2, 3, 4, 5)
15+
val result = flow.scan { acc, v -> acc + v }.toList()
16+
assertEquals(listOf(1, 3, 6, 10, 15), result)
17+
}
18+
19+
@Test
20+
fun testScanWithInitial() = runTest {
21+
val flow = flowOf(1, 2, 3)
22+
val result = flow.scan(::emptyList) { acc: List<Int>, value -> acc + value }.toList()
23+
assertEquals(listOf(emptyList(), listOf(1), listOf(1, 2), listOf(1, 2, 3)), result)
24+
}
25+
26+
@Test
27+
fun testNulls() = runTest {
28+
val flow = flowOf(null, 2, null, null, null, 5)
29+
val result = flow.scan { acc, v -> if (v == null) acc else (if (acc == null) v else acc + v) }.toList()
30+
assertEquals(listOf(null, 2, 2, 2, 2, 7), result)
31+
}
32+
33+
@Test
34+
fun testEmptyFlow() = runTest {
35+
val result = emptyFlow<Int>().scan { _, _ -> 1 }.toList()
36+
assertTrue(result.isEmpty())
37+
}
38+
39+
@Test
40+
fun testErrorCancelsUpstream() = runTest {
41+
expect(1)
42+
val latch = Channel<Unit>()
43+
val flow = flow {
44+
coroutineScope {
45+
launch {
46+
latch.send(Unit)
47+
hang { expect(3) }
48+
}
49+
emit(1)
50+
emit(2)
51+
}
52+
}.scan { _, value ->
53+
expect(value) // 2
54+
latch.receive()
55+
throw TestException()
56+
}.onErrorCollect(emptyFlow())
57+
58+
assertEquals(1, flow.single())
59+
finish(4)
60+
}
61+
62+
public operator fun <T> Collection<T>.plus(element: T): List<T> {
63+
val result = ArrayList<T>(size + 1)
64+
result.addAll(this)
65+
result.add(element)
66+
return result
67+
}
68+
}

0 commit comments

Comments
 (0)