1 /*
2  * 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
6 
7 import kotlinx.coroutines.channels.*
8 import kotlinx.coroutines.selects.*
9 import kotlin.test.*
10 
11 class AtomicCancellationTest : TestBase() {
12 
13     @Test
14     fun testSendAtomicCancel() = runBlocking {
15         expect(1)
16         val channel = Channel<Int>()
17         val job = launch(start = CoroutineStart.UNDISPATCHED) {
18             expect(2)
19             channel.send(42) // suspends
20             expect(4) // should execute despite cancellation
21         }
22         expect(3)
23         assertEquals(42, channel.receive()) // will schedule sender for further execution
24         job.cancel() // cancel the job next
25         yield() // now yield
26         finish(5)
27     }
28 
29     @Test
30     fun testSelectSendAtomicCancel() = runBlocking {
31         expect(1)
32         val channel = Channel<Int>()
33         val job = launch(start = CoroutineStart.UNDISPATCHED) {
34             expect(2)
35             val result = select<String> { // suspends
36                 channel.onSend(42) {
37                     expect(4)
38                     "OK"
39                 }
40             }
41             assertEquals("OK", result)
42             expect(5) // should execute despite cancellation
43         }
44         expect(3)
45         assertEquals(42, channel.receive()) // will schedule sender for further execution
46         job.cancel() // cancel the job next
47         yield() // now yield
48         finish(6)
49     }
50 
51     @Test
52     fun testReceiveAtomicCancel() = runBlocking {
53         expect(1)
54         val channel = Channel<Int>()
55         val job = launch(start = CoroutineStart.UNDISPATCHED) {
56             expect(2)
57             assertEquals(42, channel.receive()) // suspends
58             expect(4) // should execute despite cancellation
59         }
60         expect(3)
61         channel.send(42) // will schedule receiver for further execution
62         job.cancel() // cancel the job next
63         yield() // now yield
64         finish(5)
65     }
66 
67     @Test
68     fun testSelectReceiveAtomicCancel() = runBlocking {
69         expect(1)
70         val channel = Channel<Int>()
71         val job = launch(start = CoroutineStart.UNDISPATCHED) {
72             expect(2)
73             val result = select<String> { // suspends
74                 channel.onReceive {
75                     assertEquals(42, it)
76                     expect(4)
77                     "OK"
78                 }
79             }
80             assertEquals("OK", result)
81             expect(5) // should execute despite cancellation
82         }
83         expect(3)
84         channel.send(42) // will schedule receiver for further execution
85         job.cancel() // cancel the job next
86         yield() // now yield
87         finish(6)
88     }
89 
90     @Test
91     fun testSelectDeferredAwaitCancellable() = runBlocking {
92         expect(1)
93         val deferred = async { // deferred, not yet complete
94             expect(4)
95             "OK"
96         }
97         assertEquals(false, deferred.isCompleted)
98         var job: Job? = null
99         launch { // will cancel job as soon as deferred completes
100             expect(5)
101             assertEquals(true, deferred.isCompleted)
102             job!!.cancel()
103         }
104         job = launch(start = CoroutineStart.UNDISPATCHED) {
105             expect(2)
106             try {
107                 select<Unit> { // suspends
108                     deferred.onAwait { expectUnreached() }
109                 }
110                 expectUnreached() // will not execute -- cancelled while dispatched
111             } finally {
112                 finish(7) // but will execute finally blocks
113             }
114         }
115         expect(3) // continues to execute when the job suspends
116         yield() // to deferred & canceller
117         expect(6)
118     }
119 
120     @Test
121     fun testSelectJobJoinCancellable() = runBlocking {
122         expect(1)
123         val jobToJoin = launch { // not yet complete
124             expect(4)
125         }
126         assertEquals(false, jobToJoin.isCompleted)
127         var job: Job? = null
128         launch { // will cancel job as soon as jobToJoin completes
129             expect(5)
130             assertEquals(true, jobToJoin.isCompleted)
131             job!!.cancel()
132         }
133         job = launch(start = CoroutineStart.UNDISPATCHED) {
134             expect(2)
135             try {
136                 select<Unit> { // suspends
137                     jobToJoin.onJoin { expectUnreached() }
138                 }
139                 expectUnreached() // will not execute -- cancelled while dispatched
140             } finally {
141                 finish(7) // but will execute finally blocks
142             }
143         }
144         expect(3) // continues to execute when the job suspends
145         yield() // to jobToJoin & canceller
146         expect(6)
147     }
148 }