Roman Elizarov | 331750b | 2017-02-15 17:59:17 +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 | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 3 | */ |
| 4 | |
| 5 | package kotlinx.coroutines.experimental.reactive |
| 6 | |
Roman Elizarov | 2adf8bc | 2018-01-24 20:09:57 +0300 | [diff] [blame] | 7 | import kotlinx.atomicfu.* |
Roman Elizarov | c0d559b | 2017-09-28 14:27:05 +0300 | [diff] [blame] | 8 | import kotlinx.coroutines.experimental.* |
Roman Elizarov | 2adf8bc | 2018-01-24 20:09:57 +0300 | [diff] [blame] | 9 | import kotlinx.coroutines.experimental.channels.* |
| 10 | import kotlinx.coroutines.experimental.selects.* |
| 11 | import kotlinx.coroutines.experimental.sync.* |
| 12 | import org.reactivestreams.* |
| 13 | import kotlin.coroutines.experimental.* |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 14 | |
| 15 | /** |
| 16 | * Creates cold reactive [Publisher] that runs a given [block] in a coroutine. |
Roman Elizarov | c0d559b | 2017-09-28 14:27:05 +0300 | [diff] [blame] | 17 | * Every time the returned publisher is subscribed, it starts a new coroutine. |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 18 | * Coroutine emits items with `send`. Unsubscribing cancels running coroutine. |
| 19 | * |
| 20 | * Invocations of `send` are suspended appropriately when subscribers apply back-pressure and to ensure that |
| 21 | * `onNext` is not invoked concurrently. |
| 22 | * |
| 23 | * | **Coroutine action** | **Signal to subscriber** |
| 24 | * | -------------------------------------------- | ------------------------ |
| 25 | * | `send` | `onNext` |
| 26 | * | Normal completion or `close` without cause | `onComplete` |
| 27 | * | Failure with exception or `close` with cause | `onError` |
Roman Elizarov | c0d559b | 2017-09-28 14:27:05 +0300 | [diff] [blame] | 28 | * |
| 29 | * The [context] for the new coroutine can be explicitly specified. |
| 30 | * See [CoroutineDispatcher] for the standard context implementations that are provided by `kotlinx.coroutines`. |
Roman Elizarov | c7d10a4 | 2018-03-13 18:28:42 +0300 | [diff] [blame] | 31 | * The [coroutineContext] of the parent coroutine may be used, |
Roman Elizarov | c0d559b | 2017-09-28 14:27:05 +0300 | [diff] [blame] | 32 | * in which case the [Job] of the resulting coroutine is a child of the job of the parent coroutine. |
Roman Elizarov | e8f694e | 2017-11-28 10:12:00 +0300 | [diff] [blame] | 33 | * The parent job may be also explicitly specified using [parent] parameter. |
| 34 | |
Roman Elizarov | c0d559b | 2017-09-28 14:27:05 +0300 | [diff] [blame] | 35 | * If the context does not have any dispatcher nor any other [ContinuationInterceptor], then [DefaultDispatcher] is used. |
| 36 | * |
| 37 | * @param context context of the coroutine. The default value is [DefaultDispatcher]. |
Roman Elizarov | e8f694e | 2017-11-28 10:12:00 +0300 | [diff] [blame] | 38 | * @param parent explicitly specifies the parent job, overrides job from the [context] (if any). |
Roman Elizarov | c0d559b | 2017-09-28 14:27:05 +0300 | [diff] [blame] | 39 | * @param block the coroutine code. |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 40 | */ |
Roman Elizarov | e8f694e | 2017-11-28 10:12:00 +0300 | [diff] [blame] | 41 | public fun <T> publish( |
| 42 | context: CoroutineContext = DefaultDispatcher, |
| 43 | parent: Job? = null, |
| 44 | block: suspend ProducerScope<T>.() -> Unit |
| 45 | ): Publisher<T> = Publisher { subscriber -> |
| 46 | val newContext = newCoroutineContext(context, parent) |
| 47 | val coroutine = PublisherCoroutine(newContext, subscriber) |
Roman Elizarov | e8f694e | 2017-11-28 10:12:00 +0300 | [diff] [blame] | 48 | subscriber.onSubscribe(coroutine) // do it first (before starting coroutine), to avoid unnecessary suspensions |
Roman Elizarov | 2adf8bc | 2018-01-24 20:09:57 +0300 | [diff] [blame] | 49 | coroutine.start(CoroutineStart.DEFAULT, coroutine, block) |
Roman Elizarov | e8f694e | 2017-11-28 10:12:00 +0300 | [diff] [blame] | 50 | } |
| 51 | |
| 52 | /** @suppress **Deprecated**: Binary compatibility */ |
| 53 | @Deprecated(message = "Binary compatibility", level = DeprecationLevel.HIDDEN) |
Roman Elizarov | c0d559b | 2017-09-28 14:27:05 +0300 | [diff] [blame] | 54 | @JvmOverloads // for binary compatibility with older code compiled before context had a default |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 55 | public fun <T> publish( |
Roman Elizarov | c0d559b | 2017-09-28 14:27:05 +0300 | [diff] [blame] | 56 | context: CoroutineContext = DefaultDispatcher, |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 57 | block: suspend ProducerScope<T>.() -> Unit |
Roman Elizarov | e8f694e | 2017-11-28 10:12:00 +0300 | [diff] [blame] | 58 | ): Publisher<T> = |
| 59 | publish(context, block = block) |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 60 | |
Roman Elizarov | 7753f8e | 2017-08-15 11:16:33 +0300 | [diff] [blame] | 61 | private const val CLOSED = -1L // closed, but have not signalled onCompleted/onError yet |
| 62 | private const val SIGNALLED = -2L // already signalled subscriber onCompleted/onError |
| 63 | |
Roman Elizarov | db0e22d | 2017-08-29 18:15:57 +0300 | [diff] [blame] | 64 | private class PublisherCoroutine<in T>( |
Roman Elizarov | 2b12d58 | 2017-06-22 20:12:19 +0300 | [diff] [blame] | 65 | parentContext: CoroutineContext, |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 66 | private val subscriber: Subscriber<T> |
Roman Elizarov | db0e22d | 2017-08-29 18:15:57 +0300 | [diff] [blame] | 67 | ) : AbstractCoroutine<Unit>(parentContext, true), ProducerScope<T>, Subscription, SelectClause2<T, SendChannel<T>> { |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 68 | override val channel: SendChannel<T> get() = this |
| 69 | |
| 70 | // Mutex is locked when either nRequested == 0 or while subscriber.onXXX is being invoked |
| 71 | private val mutex = Mutex(locked = true) |
| 72 | |
Roman Elizarov | 7753f8e | 2017-08-15 11:16:33 +0300 | [diff] [blame] | 73 | private val _nRequested = atomic(0L) // < 0 when closed (CLOSED or SIGNALLED) |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 74 | |
| 75 | override val isClosedForSend: Boolean get() = isCompleted |
| 76 | override val isFull: Boolean = mutex.isLocked |
| 77 | override fun close(cause: Throwable?): Boolean = cancel(cause) |
| 78 | |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 79 | override fun offer(element: T): Boolean { |
| 80 | if (!mutex.tryLock()) return false |
| 81 | doLockedNext(element) |
| 82 | return true |
| 83 | } |
| 84 | |
Roman Elizarov | 6640b2b | 2018-01-17 19:08:55 +0300 | [diff] [blame] | 85 | public override suspend fun send(element: T) { |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 86 | // fast-path -- try send without suspension |
| 87 | if (offer(element)) return |
| 88 | // slow-path does suspend |
| 89 | return sendSuspend(element) |
| 90 | } |
| 91 | |
| 92 | private suspend fun sendSuspend(element: T) { |
| 93 | mutex.lock() |
| 94 | doLockedNext(element) |
| 95 | } |
| 96 | |
Roman Elizarov | db0e22d | 2017-08-29 18:15:57 +0300 | [diff] [blame] | 97 | override val onSend: SelectClause2<T, SendChannel<T>> |
| 98 | get() = this |
| 99 | |
| 100 | // registerSelectSend |
| 101 | @Suppress("PARAMETER_NAME_CHANGED_ON_OVERRIDE") |
| 102 | override fun <R> registerSelectClause2(select: SelectInstance<R>, element: T, block: suspend (SendChannel<T>) -> R) { |
| 103 | mutex.onLock.registerSelectClause2(select, null) { |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 104 | doLockedNext(element) |
Roman Elizarov | db0e22d | 2017-08-29 18:15:57 +0300 | [diff] [blame] | 105 | block(this) |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 106 | } |
Roman Elizarov | db0e22d | 2017-08-29 18:15:57 +0300 | [diff] [blame] | 107 | } |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 108 | |
| 109 | // assert: mutex.isLocked() |
| 110 | private fun doLockedNext(elem: T) { |
| 111 | // check if already closed for send |
Roman Elizarov | 29affbb | 2017-07-21 13:47:41 +0300 | [diff] [blame] | 112 | if (!isActive) { |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 113 | doLockedSignalCompleted() |
Roman Elizarov | 8b38fa2 | 2017-09-27 17:44:31 +0300 | [diff] [blame] | 114 | throw getCancellationException() |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 115 | } |
| 116 | // notify subscriber |
| 117 | try { |
| 118 | subscriber.onNext(elem) |
| 119 | } catch (e: Throwable) { |
| 120 | try { |
| 121 | if (!cancel(e)) |
Roman Elizarov | 9fe5f46 | 2018-02-21 19:05:52 +0300 | [diff] [blame] | 122 | handleCoroutineException(context, e) |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 123 | } finally { |
| 124 | doLockedSignalCompleted() |
| 125 | } |
Roman Elizarov | 8b38fa2 | 2017-09-27 17:44:31 +0300 | [diff] [blame] | 126 | throw getCancellationException() |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 127 | } |
| 128 | // now update nRequested |
| 129 | while (true) { // lock-free loop on nRequested |
Roman Elizarov | 7753f8e | 2017-08-15 11:16:33 +0300 | [diff] [blame] | 130 | val cur = _nRequested.value |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 131 | if (cur < 0) break // closed from inside onNext => unlock |
| 132 | if (cur == Long.MAX_VALUE) break // no back-pressure => unlock |
| 133 | val upd = cur - 1 |
Roman Elizarov | 7753f8e | 2017-08-15 11:16:33 +0300 | [diff] [blame] | 134 | if (_nRequested.compareAndSet(cur, upd)) { |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 135 | if (upd == 0L) return // return to keep locked due to back-pressure |
| 136 | break // unlock if upd > 0 |
| 137 | } |
| 138 | } |
| 139 | /* |
Roman Elizarov | 29affbb | 2017-07-21 13:47:41 +0300 | [diff] [blame] | 140 | There is no sense to check for `isActive` before doing `unlock`, because cancellation/completion might |
| 141 | happen after this check and before `unlock` (see `onCancellation` that does not do anything |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 142 | if it fails to acquire the lock that we are still holding). |
Roman Elizarov | 29affbb | 2017-07-21 13:47:41 +0300 | [diff] [blame] | 143 | We have to recheck `isActive` after `unlock` anyway. |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 144 | */ |
| 145 | mutex.unlock() |
Roman Elizarov | 29affbb | 2017-07-21 13:47:41 +0300 | [diff] [blame] | 146 | // recheck isActive |
| 147 | if (!isActive && mutex.tryLock()) |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 148 | doLockedSignalCompleted() |
| 149 | } |
| 150 | |
| 151 | // assert: mutex.isLocked() |
| 152 | private fun doLockedSignalCompleted() { |
| 153 | try { |
Roman Elizarov | 7753f8e | 2017-08-15 11:16:33 +0300 | [diff] [blame] | 154 | if (_nRequested.value >= CLOSED) { |
| 155 | _nRequested.value = SIGNALLED // we'll signal onError/onCompleted (that the final state -- no CAS needed) |
Roman Elizarov | 3ef4fca | 2017-07-12 12:42:20 +0300 | [diff] [blame] | 156 | val cause = getCompletionCause() |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 157 | try { |
Roman Elizarov | e89cd68 | 2018-04-25 13:03:40 +0300 | [diff] [blame] | 158 | if (cause != null && cause !is CancellationException) |
Roman Elizarov | 3ef4fca | 2017-07-12 12:42:20 +0300 | [diff] [blame] | 159 | subscriber.onError(cause) |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 160 | else |
| 161 | subscriber.onComplete() |
| 162 | } catch (e: Throwable) { |
Roman Elizarov | 9fe5f46 | 2018-02-21 19:05:52 +0300 | [diff] [blame] | 163 | handleCoroutineException(context, e) |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 164 | } |
| 165 | } |
| 166 | } finally { |
| 167 | mutex.unlock() |
| 168 | } |
| 169 | } |
| 170 | |
| 171 | override fun request(n: Long) { |
| 172 | if (n < 0) { |
| 173 | cancel(IllegalArgumentException("Must request non-negative number, but $n requested")) |
| 174 | return |
| 175 | } |
| 176 | while (true) { // lock-free loop for nRequested |
Roman Elizarov | 7753f8e | 2017-08-15 11:16:33 +0300 | [diff] [blame] | 177 | val cur = _nRequested.value |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 178 | if (cur < 0) return // already closed for send, ignore requests |
| 179 | var upd = cur + n |
| 180 | if (upd < 0 || n == Long.MAX_VALUE) |
| 181 | upd = Long.MAX_VALUE |
| 182 | if (cur == upd) return // nothing to do |
Roman Elizarov | 7753f8e | 2017-08-15 11:16:33 +0300 | [diff] [blame] | 183 | if (_nRequested.compareAndSet(cur, upd)) { |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 184 | // unlock the mutex when we don't have back-pressure anymore |
Roman Elizarov | d14a390 | 2017-03-13 12:03:35 +0300 | [diff] [blame] | 185 | if (cur == 0L) { |
| 186 | mutex.unlock() |
Roman Elizarov | 29affbb | 2017-07-21 13:47:41 +0300 | [diff] [blame] | 187 | // recheck isActive |
| 188 | if (!isActive && mutex.tryLock()) |
Roman Elizarov | d14a390 | 2017-03-13 12:03:35 +0300 | [diff] [blame] | 189 | doLockedSignalCompleted() |
| 190 | } |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 191 | return |
| 192 | } |
| 193 | } |
| 194 | } |
| 195 | |
Roman Elizarov | 6640b2b | 2018-01-17 19:08:55 +0300 | [diff] [blame] | 196 | override fun onCancellation(cause: Throwable?) { |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 197 | while (true) { // lock-free loop for nRequested |
Roman Elizarov | 7753f8e | 2017-08-15 11:16:33 +0300 | [diff] [blame] | 198 | val cur = _nRequested.value |
Roman Elizarov | 29affbb | 2017-07-21 13:47:41 +0300 | [diff] [blame] | 199 | if (cur == SIGNALLED) return // some other thread holding lock already signalled cancellation/completion |
Roman Elizarov | 3ef4fca | 2017-07-12 12:42:20 +0300 | [diff] [blame] | 200 | check(cur >= 0) // no other thread could have marked it as CLOSED, because onCancellation is invoked once |
Roman Elizarov | 7753f8e | 2017-08-15 11:16:33 +0300 | [diff] [blame] | 201 | if (!_nRequested.compareAndSet(cur, CLOSED)) continue // retry on failed CAS |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 202 | // Ok -- marked as CLOSED, now can unlock the mutex if it was locked due to backpressure |
| 203 | if (cur == 0L) { |
| 204 | doLockedSignalCompleted() |
| 205 | } else { |
| 206 | // otherwise mutex was either not locked or locked in concurrent onNext... try lock it to signal completion |
| 207 | if (mutex.tryLock()) |
| 208 | doLockedSignalCompleted() |
| 209 | // Note: if failed `tryLock`, then `doLockedNext` will signal after performing `unlock` |
| 210 | } |
| 211 | return // done anyway |
| 212 | } |
| 213 | } |
| 214 | |
| 215 | // Subscription impl |
| 216 | override fun cancel() { |
| 217 | cancel(cause = null) |
| 218 | } |
| 219 | } |