1 /*
<lambda>null2  * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
3  */
4 
5 package kotlinx.coroutines.rx3
6 
7 import kotlinx.coroutines.*
8 import org.junit.*
9 import java.util.*
10 import kotlin.coroutines.*
11 
12 class ObservableCompletionStressTest : TestBase() {
13     private val N_REPEATS = 10_000 * stressTestMultiplier
14 
15     private fun CoroutineScope.range(context: CoroutineContext, start: Int, count: Int) = rxObservable(context) {
16         for (x in start until start + count) send(x)
17     }
18 
19     @Test
20     fun testCompletion() {
21         val rnd = Random()
22         repeat(N_REPEATS) {
23             val count = rnd.nextInt(5)
24             runBlocking {
25                 withTimeout(5000) {
26                     var received = 0
27                     range(Dispatchers.Default, 1, count).collect { x ->
28                         received++
29                         if (x != received) error("$x != $received")
30                     }
31                     if (received != count) error("$received != $count")
32                 }
33             }
34         }
35     }
36 }
37