blob: 371d01472e44b14803c4ad076df53d9f9aadad06 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
|
/*
* Copyright 2016-2020 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
*/
package kotlinx.coroutines.flow
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
import kotlin.test.*
class ShareInFusionTest : TestBase() {
/**
* Test perfect fusion for operators **after** [shareIn].
*/
@Test
fun testOperatorFusion() = runTest {
val sh = emptyFlow<Int>().shareIn(this, SharingStarted.Eagerly)
assertTrue(sh !is MutableSharedFlow<*>) // cannot be cast to mutable shared flow!!!
assertSame(sh, (sh as Flow<*>).cancellable())
assertSame(sh, (sh as Flow<*>).flowOn(Dispatchers.Default))
assertSame(sh, sh.buffer(Channel.RENDEZVOUS))
coroutineContext.cancelChildren()
}
@Test
fun testFlowOnContextFusion() = runTest {
val flow = flow {
assertEquals("FlowCtx", currentCoroutineContext()[CoroutineName]?.name)
emit("OK")
}.flowOn(CoroutineName("FlowCtx"))
assertEquals("OK", flow.shareIn(this, SharingStarted.Eagerly, 1).first())
coroutineContext.cancelChildren()
}
/**
* Tests that `channelFlow { ... }.buffer(x)` works according to the [channelFlow] docs, and subsequent
* application of [shareIn] does not leak upstream.
*/
@Test
fun testChannelFlowBufferShareIn() = runTest {
expect(1)
val flow = channelFlow {
// send a batch of 10 elements using [offer]
for (i in 1..10) {
assertTrue(offer(i)) // offer must succeed, because buffer
}
send(0) // done
}.buffer(10) // request a buffer of 10
// ^^^^^^^^^ buffer stays here
val shared = flow.shareIn(this, SharingStarted.Eagerly)
shared
.takeWhile { it > 0 }
.collect { i -> expect(i + 1) }
finish(12)
}
}
|
