Roman Elizarov | e3aa8ff | 2017-04-27 19:16:40 +0300 | [diff] [blame] | 1 | /* |
Roman Elizarov | 1f74a2d | 2018-06-29 19:19:45 +0300 | [diff] [blame^] | 2 | * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license. |
Roman Elizarov | e3aa8ff | 2017-04-27 19:16:40 +0300 | [diff] [blame] | 3 | */ |
| 4 | |
| 5 | package kotlinx.coroutines.experimental.channels |
| 6 | |
| 7 | import kotlinx.coroutines.experimental.* |
Roman Elizarov | 9fe5f46 | 2018-02-21 19:05:52 +0300 | [diff] [blame] | 8 | import kotlin.coroutines.experimental.* |
Vsevolod Tolstopyatov | 9619134 | 2018-04-20 18:13:33 +0300 | [diff] [blame] | 9 | import kotlin.test.* |
Roman Elizarov | e3aa8ff | 2017-04-27 19:16:40 +0300 | [diff] [blame] | 10 | |
| 11 | class ArrayBroadcastChannelTest : TestBase() { |
Vsevolod Tolstopyatov | 167ca63 | 2018-06-29 11:49:00 +0300 | [diff] [blame] | 12 | |
| 13 | @Test |
| 14 | fun testConcurrentModification() = runTest { |
| 15 | val channel = ArrayBroadcastChannel<Int>(1) |
| 16 | val s1 = channel.openSubscription() |
| 17 | val s2 = channel.openSubscription() |
| 18 | |
| 19 | val job1 = launch(Unconfined, CoroutineStart.UNDISPATCHED) { |
| 20 | expect(1) |
| 21 | s1.receive() |
| 22 | s1.cancel() |
| 23 | } |
| 24 | |
| 25 | val job2 = launch(Unconfined, CoroutineStart.UNDISPATCHED) { |
| 26 | expect(2) |
| 27 | s2.receive() |
| 28 | } |
| 29 | |
| 30 | expect(3) |
| 31 | channel.send(1) |
| 32 | joinAll(job1, job2) |
| 33 | finish(4) |
| 34 | } |
| 35 | |
Roman Elizarov | e3aa8ff | 2017-04-27 19:16:40 +0300 | [diff] [blame] | 36 | @Test |
Vsevolod Tolstopyatov | 9619134 | 2018-04-20 18:13:33 +0300 | [diff] [blame] | 37 | fun testBasic() = runTest { |
Roman Elizarov | e3aa8ff | 2017-04-27 19:16:40 +0300 | [diff] [blame] | 38 | expect(1) |
| 39 | val broadcast = ArrayBroadcastChannel<Int>(1) |
Vsevolod Tolstopyatov | 9619134 | 2018-04-20 18:13:33 +0300 | [diff] [blame] | 40 | assertFalse(broadcast.isClosedForSend) |
Roman Elizarov | 0f66a6d | 2017-06-22 14:57:53 +0300 | [diff] [blame] | 41 | val first = broadcast.openSubscription() |
Roman Elizarov | 43e3af7 | 2017-07-21 16:01:31 +0300 | [diff] [blame] | 42 | launch(coroutineContext, CoroutineStart.UNDISPATCHED) { |
Roman Elizarov | e3aa8ff | 2017-04-27 19:16:40 +0300 | [diff] [blame] | 43 | expect(2) |
Vsevolod Tolstopyatov | 9619134 | 2018-04-20 18:13:33 +0300 | [diff] [blame] | 44 | assertEquals(1, first.receive()) // suspends |
| 45 | assertFalse(first.isClosedForReceive) |
Roman Elizarov | e3aa8ff | 2017-04-27 19:16:40 +0300 | [diff] [blame] | 46 | expect(5) |
Vsevolod Tolstopyatov | 9619134 | 2018-04-20 18:13:33 +0300 | [diff] [blame] | 47 | assertEquals(2, first.receive()) // suspends |
| 48 | assertFalse(first.isClosedForReceive) |
Roman Elizarov | e3aa8ff | 2017-04-27 19:16:40 +0300 | [diff] [blame] | 49 | expect(10) |
Vsevolod Tolstopyatov | 9619134 | 2018-04-20 18:13:33 +0300 | [diff] [blame] | 50 | assertNull(first.receiveOrNull()) // suspends |
| 51 | assertTrue(first.isClosedForReceive) |
Roman Elizarov | e3aa8ff | 2017-04-27 19:16:40 +0300 | [diff] [blame] | 52 | expect(14) |
| 53 | } |
| 54 | expect(3) |
| 55 | broadcast.send(1) |
| 56 | expect(4) |
| 57 | yield() // to the first receiver |
| 58 | expect(6) |
Vsevolod Tolstopyatov | 9619134 | 2018-04-20 18:13:33 +0300 | [diff] [blame] | 59 | |
Roman Elizarov | 0f66a6d | 2017-06-22 14:57:53 +0300 | [diff] [blame] | 60 | val second = broadcast.openSubscription() |
Roman Elizarov | 43e3af7 | 2017-07-21 16:01:31 +0300 | [diff] [blame] | 61 | launch(coroutineContext, CoroutineStart.UNDISPATCHED) { |
Roman Elizarov | e3aa8ff | 2017-04-27 19:16:40 +0300 | [diff] [blame] | 62 | expect(7) |
Vsevolod Tolstopyatov | 9619134 | 2018-04-20 18:13:33 +0300 | [diff] [blame] | 63 | assertEquals(2, second.receive()) // suspends |
| 64 | assertFalse(second.isClosedForReceive) |
Roman Elizarov | e3aa8ff | 2017-04-27 19:16:40 +0300 | [diff] [blame] | 65 | expect(11) |
Vsevolod Tolstopyatov | 9619134 | 2018-04-20 18:13:33 +0300 | [diff] [blame] | 66 | assertNull(second.receiveOrNull()) // suspends |
| 67 | assertTrue(second.isClosedForReceive) |
Roman Elizarov | e3aa8ff | 2017-04-27 19:16:40 +0300 | [diff] [blame] | 68 | expect(15) |
| 69 | } |
| 70 | expect(8) |
| 71 | broadcast.send(2) |
| 72 | expect(9) |
| 73 | yield() // to first & second receivers |
| 74 | expect(12) |
| 75 | broadcast.close() |
| 76 | expect(13) |
Vsevolod Tolstopyatov | 9619134 | 2018-04-20 18:13:33 +0300 | [diff] [blame] | 77 | assertTrue(broadcast.isClosedForSend) |
Roman Elizarov | e3aa8ff | 2017-04-27 19:16:40 +0300 | [diff] [blame] | 78 | yield() // to first & second receivers |
| 79 | finish(16) |
| 80 | } |
| 81 | |
| 82 | @Test |
Vsevolod Tolstopyatov | 9619134 | 2018-04-20 18:13:33 +0300 | [diff] [blame] | 83 | fun testSendSuspend() = runTest { |
Roman Elizarov | e3aa8ff | 2017-04-27 19:16:40 +0300 | [diff] [blame] | 84 | expect(1) |
| 85 | val broadcast = ArrayBroadcastChannel<Int>(1) |
Roman Elizarov | 0f66a6d | 2017-06-22 14:57:53 +0300 | [diff] [blame] | 86 | val first = broadcast.openSubscription() |
Roman Elizarov | 43e3af7 | 2017-07-21 16:01:31 +0300 | [diff] [blame] | 87 | launch(coroutineContext) { |
Roman Elizarov | e3aa8ff | 2017-04-27 19:16:40 +0300 | [diff] [blame] | 88 | expect(4) |
Vsevolod Tolstopyatov | 9619134 | 2018-04-20 18:13:33 +0300 | [diff] [blame] | 89 | assertEquals(1, first.receive()) |
Roman Elizarov | e3aa8ff | 2017-04-27 19:16:40 +0300 | [diff] [blame] | 90 | expect(5) |
Vsevolod Tolstopyatov | 9619134 | 2018-04-20 18:13:33 +0300 | [diff] [blame] | 91 | assertEquals(2, first.receive()) |
Roman Elizarov | e3aa8ff | 2017-04-27 19:16:40 +0300 | [diff] [blame] | 92 | expect(6) |
| 93 | } |
| 94 | expect(2) |
| 95 | broadcast.send(1) // puts to buffer, receiver not running yet |
| 96 | expect(3) |
| 97 | broadcast.send(2) // suspends |
| 98 | finish(7) |
| 99 | } |
| 100 | |
| 101 | @Test |
Vsevolod Tolstopyatov | 9619134 | 2018-04-20 18:13:33 +0300 | [diff] [blame] | 102 | fun testConcurrentSendCompletion() = runTest { |
Roman Elizarov | e3aa8ff | 2017-04-27 19:16:40 +0300 | [diff] [blame] | 103 | expect(1) |
| 104 | val broadcast = ArrayBroadcastChannel<Int>(1) |
Roman Elizarov | 0f66a6d | 2017-06-22 14:57:53 +0300 | [diff] [blame] | 105 | val sub = broadcast.openSubscription() |
Roman Elizarov | e3aa8ff | 2017-04-27 19:16:40 +0300 | [diff] [blame] | 106 | // launch 3 concurrent senders (one goes buffer, two other suspend) |
| 107 | for (x in 1..3) { |
Roman Elizarov | 43e3af7 | 2017-07-21 16:01:31 +0300 | [diff] [blame] | 108 | launch(coroutineContext, CoroutineStart.UNDISPATCHED) { |
Roman Elizarov | e3aa8ff | 2017-04-27 19:16:40 +0300 | [diff] [blame] | 109 | expect(x + 1) |
| 110 | broadcast.send(x) |
| 111 | } |
| 112 | } |
| 113 | // and close it for send |
| 114 | expect(5) |
| 115 | broadcast.close() |
| 116 | // now must receive all 3 items |
| 117 | expect(6) |
Vsevolod Tolstopyatov | 9619134 | 2018-04-20 18:13:33 +0300 | [diff] [blame] | 118 | assertFalse(sub.isClosedForReceive) |
Roman Elizarov | e3aa8ff | 2017-04-27 19:16:40 +0300 | [diff] [blame] | 119 | for (x in 1..3) |
Vsevolod Tolstopyatov | 9619134 | 2018-04-20 18:13:33 +0300 | [diff] [blame] | 120 | assertEquals(x, sub.receiveOrNull()) |
Roman Elizarov | e3aa8ff | 2017-04-27 19:16:40 +0300 | [diff] [blame] | 121 | // and receive close signal |
Vsevolod Tolstopyatov | 9619134 | 2018-04-20 18:13:33 +0300 | [diff] [blame] | 122 | assertNull(sub.receiveOrNull()) |
| 123 | assertTrue(sub.isClosedForReceive) |
Roman Elizarov | e3aa8ff | 2017-04-27 19:16:40 +0300 | [diff] [blame] | 124 | finish(7) |
| 125 | } |
Roman Elizarov | 921b0cf | 2017-06-22 14:36:55 +0300 | [diff] [blame] | 126 | |
| 127 | @Test |
Vsevolod Tolstopyatov | 9619134 | 2018-04-20 18:13:33 +0300 | [diff] [blame] | 128 | fun testForgetUnsubscribed() = runTest { |
Roman Elizarov | 921b0cf | 2017-06-22 14:36:55 +0300 | [diff] [blame] | 129 | expect(1) |
| 130 | val broadcast = ArrayBroadcastChannel<Int>(1) |
| 131 | broadcast.send(1) |
| 132 | broadcast.send(2) |
| 133 | broadcast.send(3) |
| 134 | expect(2) // should not suspend anywhere above |
Roman Elizarov | 0f66a6d | 2017-06-22 14:57:53 +0300 | [diff] [blame] | 135 | val sub = broadcast.openSubscription() |
Roman Elizarov | 43e3af7 | 2017-07-21 16:01:31 +0300 | [diff] [blame] | 136 | launch(coroutineContext, CoroutineStart.UNDISPATCHED) { |
Roman Elizarov | 921b0cf | 2017-06-22 14:36:55 +0300 | [diff] [blame] | 137 | expect(3) |
Vsevolod Tolstopyatov | 9619134 | 2018-04-20 18:13:33 +0300 | [diff] [blame] | 138 | assertEquals(4, sub.receive()) // suspends |
Roman Elizarov | 921b0cf | 2017-06-22 14:36:55 +0300 | [diff] [blame] | 139 | expect(5) |
| 140 | } |
| 141 | expect(4) |
| 142 | broadcast.send(4) // sends |
| 143 | yield() |
| 144 | finish(6) |
| 145 | } |
Roman Elizarov | c2adef5 | 2017-08-17 16:13:41 +0300 | [diff] [blame] | 146 | |
| 147 | @Test |
Vsevolod Tolstopyatov | 9619134 | 2018-04-20 18:13:33 +0300 | [diff] [blame] | 148 | fun testReceiveFullAfterClose() = runTest { |
Roman Elizarov | c2adef5 | 2017-08-17 16:13:41 +0300 | [diff] [blame] | 149 | val channel = BroadcastChannel<Int>(10) |
| 150 | val sub = channel.openSubscription() |
| 151 | // generate into buffer & close |
| 152 | for (x in 1..5) channel.send(x) |
| 153 | channel.close() |
| 154 | // make sure all of them are consumed |
| 155 | check(!sub.isClosedForReceive) |
| 156 | for (x in 1..5) check(sub.receive() == x) |
| 157 | check(sub.receiveOrNull() == null) |
| 158 | check(sub.isClosedForReceive) |
| 159 | } |
| 160 | |
| 161 | @Test |
Vsevolod Tolstopyatov | 9619134 | 2018-04-20 18:13:33 +0300 | [diff] [blame] | 162 | fun testCloseSubDuringIteration() = runTest { |
Roman Elizarov | c2adef5 | 2017-08-17 16:13:41 +0300 | [diff] [blame] | 163 | val channel = BroadcastChannel<Int>(1) |
| 164 | // launch generator (for later) in this context |
| 165 | launch(coroutineContext) { |
| 166 | for (x in 1..5) channel.send(x) |
| 167 | channel.close() |
| 168 | } |
| 169 | // start consuming |
| 170 | val sub = channel.openSubscription() |
| 171 | var expected = 0 |
| 172 | sub.consumeEach { |
| 173 | check(it == ++expected) |
| 174 | if (it == 2) { |
Roman Elizarov | f2bdf60 | 2018-04-26 11:29:47 +0300 | [diff] [blame] | 175 | sub.cancel() |
Roman Elizarov | c2adef5 | 2017-08-17 16:13:41 +0300 | [diff] [blame] | 176 | } |
| 177 | } |
| 178 | check(expected == 2) |
| 179 | } |
Roman Elizarov | b2b5c06 | 2018-02-07 12:55:44 +0300 | [diff] [blame] | 180 | |
| 181 | @Test |
Vsevolod Tolstopyatov | 9619134 | 2018-04-20 18:13:33 +0300 | [diff] [blame] | 182 | fun testReceiveFromClosedSub() = runTest({ it is ClosedReceiveChannelException }) { |
Roman Elizarov | b2b5c06 | 2018-02-07 12:55:44 +0300 | [diff] [blame] | 183 | val channel = BroadcastChannel<Int>(1) |
| 184 | val sub = channel.openSubscription() |
| 185 | assertFalse(sub.isClosedForReceive) |
Roman Elizarov | f2bdf60 | 2018-04-26 11:29:47 +0300 | [diff] [blame] | 186 | sub.cancel() |
Roman Elizarov | b2b5c06 | 2018-02-07 12:55:44 +0300 | [diff] [blame] | 187 | assertTrue(sub.isClosedForReceive) |
| 188 | sub.receive() |
| 189 | } |
Vsevolod Tolstopyatov | 4b9a559 | 2018-04-11 13:17:14 +0300 | [diff] [blame] | 190 | |
Roman Elizarov | f2bdf60 | 2018-04-26 11:29:47 +0300 | [diff] [blame] | 191 | @Test |
| 192 | fun testCancelWithCause() = runTest({ it is TestException }) { |
Vsevolod Tolstopyatov | 4b9a559 | 2018-04-11 13:17:14 +0300 | [diff] [blame] | 193 | val channel = BroadcastChannel<Int>(1) |
| 194 | val subscription = channel.openSubscription() |
Roman Elizarov | f2bdf60 | 2018-04-26 11:29:47 +0300 | [diff] [blame] | 195 | subscription.cancel(TestException()) |
Vsevolod Tolstopyatov | 4b9a559 | 2018-04-11 13:17:14 +0300 | [diff] [blame] | 196 | subscription.receiveOrNull() |
| 197 | } |
Roman Elizarov | f2bdf60 | 2018-04-26 11:29:47 +0300 | [diff] [blame] | 198 | |
Roman Elizarov | 89f8ff7 | 2018-03-14 13:39:03 +0300 | [diff] [blame] | 199 | @Test |
| 200 | fun testReceiveNoneAfterCancel() = runTest { |
| 201 | val channel = BroadcastChannel<Int>(10) |
| 202 | val sub = channel.openSubscription() |
| 203 | // generate into buffer & cancel |
| 204 | for (x in 1..5) channel.send(x) |
| 205 | channel.cancel() |
| 206 | assertTrue(channel.isClosedForSend) |
| 207 | assertTrue(sub.isClosedForReceive) |
| 208 | check(sub.receiveOrNull() == null) |
| 209 | } |
| 210 | |
Roman Elizarov | f2bdf60 | 2018-04-26 11:29:47 +0300 | [diff] [blame] | 211 | private class TestException : Exception() |
Vsevolod Tolstopyatov | 4b9a559 | 2018-04-11 13:17:14 +0300 | [diff] [blame] | 212 | } |