Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 1 | /* |
| 2 | * Copyright 2016-2017 JetBrains s.r.o. |
| 3 | * |
| 4 | * Licensed under the Apache License, Version 2.0 (the "License"); |
| 5 | * you may not use this file except in compliance with the License. |
| 6 | * You may obtain a copy of the License at |
| 7 | * |
| 8 | * http://www.apache.org/licenses/LICENSE-2.0 |
| 9 | * |
| 10 | * Unless required by applicable law or agreed to in writing, software |
| 11 | * distributed under the License is distributed on an "AS IS" BASIS, |
| 12 | * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 13 | * See the License for the specific language governing permissions and |
| 14 | * limitations under the License. |
| 15 | */ |
| 16 | |
| 17 | package kotlinx.coroutines.experimental.rx2 |
| 18 | |
Roman Elizarov | 2adf8bc | 2018-01-24 20:09:57 +0300 | [diff] [blame] | 19 | import io.reactivex.* |
| 20 | import io.reactivex.functions.* |
| 21 | import kotlinx.atomicfu.* |
Roman Elizarov | c0d559b | 2017-09-28 14:27:05 +0300 | [diff] [blame] | 22 | import kotlinx.coroutines.experimental.* |
Roman Elizarov | 2adf8bc | 2018-01-24 20:09:57 +0300 | [diff] [blame] | 23 | import kotlinx.coroutines.experimental.channels.* |
| 24 | import kotlinx.coroutines.experimental.selects.* |
| 25 | import kotlinx.coroutines.experimental.sync.* |
| 26 | import kotlin.coroutines.experimental.* |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 27 | |
| 28 | /** |
| 29 | * Creates cold [observable][Observable] that will run a given [block] in a coroutine. |
Roman Elizarov | c0d559b | 2017-09-28 14:27:05 +0300 | [diff] [blame] | 30 | * Every time the returned observable is subscribed, it starts a new coroutine. |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 31 | * Coroutine emits items with `send`. Unsubscribing cancels running coroutine. |
| 32 | * |
Roman Elizarov | 50e3221 | 2017-03-10 17:40:50 +0300 | [diff] [blame] | 33 | * Invocations of `send` are suspended appropriately to ensure that `onNext` is not invoked concurrently. |
| 34 | * Note, that Rx 2.x [Observable] **does not support backpressure**. Use [rxFlowable]. |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 35 | * |
| 36 | * | **Coroutine action** | **Signal to subscriber** |
| 37 | * | -------------------------------------------- | ------------------------ |
| 38 | * | `send` | `onNext` |
| 39 | * | Normal completion or `close` without cause | `onComplete` |
| 40 | * | Failure with exception or `close` with cause | `onError` |
Roman Elizarov | c0d559b | 2017-09-28 14:27:05 +0300 | [diff] [blame] | 41 | * |
| 42 | * The [context] for the new coroutine can be explicitly specified. |
| 43 | * 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] | 44 | * The [coroutineContext] of the parent coroutine may be used, |
Roman Elizarov | c0d559b | 2017-09-28 14:27:05 +0300 | [diff] [blame] | 45 | * 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] | 46 | * The parent job may be also explicitly specified using [parent] parameter. |
| 47 | * |
Roman Elizarov | c0d559b | 2017-09-28 14:27:05 +0300 | [diff] [blame] | 48 | * If the context does not have any dispatcher nor any other [ContinuationInterceptor], then [DefaultDispatcher] is used. |
| 49 | * |
| 50 | * @param context context of the coroutine. The default value is [DefaultDispatcher]. |
Roman Elizarov | e8f694e | 2017-11-28 10:12:00 +0300 | [diff] [blame] | 51 | * @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] | 52 | * @param block the coroutine code. |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 53 | */ |
Roman Elizarov | e8f694e | 2017-11-28 10:12:00 +0300 | [diff] [blame] | 54 | public fun <T> rxObservable( |
| 55 | context: CoroutineContext = DefaultDispatcher, |
| 56 | parent: Job? = null, |
| 57 | block: suspend ProducerScope<T>.() -> Unit |
| 58 | ): Observable<T> = Observable.create { subscriber -> |
| 59 | val newContext = newCoroutineContext(context, parent) |
| 60 | val coroutine = RxObservableCoroutine(newContext, subscriber) |
Roman Elizarov | e8f694e | 2017-11-28 10:12:00 +0300 | [diff] [blame] | 61 | subscriber.setCancellable(coroutine) // do it first (before starting coroutine), to await unnecessary suspensions |
Roman Elizarov | 2adf8bc | 2018-01-24 20:09:57 +0300 | [diff] [blame] | 62 | coroutine.start(CoroutineStart.DEFAULT, coroutine, block) |
Roman Elizarov | e8f694e | 2017-11-28 10:12:00 +0300 | [diff] [blame] | 63 | } |
| 64 | |
| 65 | /** @suppress **Deprecated**: Binary compatibility */ |
| 66 | @Deprecated(message = "Binary compatibility", level = DeprecationLevel.HIDDEN) |
Roman Elizarov | c0d559b | 2017-09-28 14:27:05 +0300 | [diff] [blame] | 67 | @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] | 68 | public fun <T> rxObservable( |
Roman Elizarov | c0d559b | 2017-09-28 14:27:05 +0300 | [diff] [blame] | 69 | context: CoroutineContext = DefaultDispatcher, |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 70 | block: suspend ProducerScope<T>.() -> Unit |
Roman Elizarov | e8f694e | 2017-11-28 10:12:00 +0300 | [diff] [blame] | 71 | ): Observable<T> = |
| 72 | rxObservable(context, block = block) |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 73 | |
Roman Elizarov | 7753f8e | 2017-08-15 11:16:33 +0300 | [diff] [blame] | 74 | private const val OPEN = 0 // open channel, still working |
| 75 | private const val CLOSED = -1 // closed, but have not signalled onCompleted/onError yet |
| 76 | private const val SIGNALLED = -2 // already signalled subscriber onCompleted/onError |
| 77 | |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 78 | private class RxObservableCoroutine<T>( |
Roman Elizarov | 2b12d58 | 2017-06-22 20:12:19 +0300 | [diff] [blame] | 79 | parentContext: CoroutineContext, |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 80 | private val subscriber: ObservableEmitter<T> |
Roman Elizarov | db0e22d | 2017-08-29 18:15:57 +0300 | [diff] [blame] | 81 | ) : AbstractCoroutine<Unit>(parentContext, true), ProducerScope<T>, Cancellable, SelectClause2<T, SendChannel<T>> { |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 82 | override val channel: SendChannel<T> get() = this |
| 83 | |
| 84 | // Mutex is locked when while subscriber.onXXX is being invoked |
| 85 | private val mutex = Mutex() |
| 86 | |
Roman Elizarov | 7753f8e | 2017-08-15 11:16:33 +0300 | [diff] [blame] | 87 | private val _signal = atomic(OPEN) |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 88 | |
| 89 | override val isClosedForSend: Boolean get() = isCompleted |
| 90 | override val isFull: Boolean = mutex.isLocked |
| 91 | override fun close(cause: Throwable?): Boolean = cancel(cause) |
| 92 | |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 93 | override fun offer(element: T): Boolean { |
| 94 | if (!mutex.tryLock()) return false |
| 95 | doLockedNext(element) |
| 96 | return true |
| 97 | } |
| 98 | |
Roman Elizarov | 6640b2b | 2018-01-17 19:08:55 +0300 | [diff] [blame] | 99 | public override suspend fun send(element: T) { |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 100 | // fast-path -- try send without suspension |
| 101 | if (offer(element)) return |
| 102 | // slow-path does suspend |
| 103 | return sendSuspend(element) |
| 104 | } |
| 105 | |
| 106 | private suspend fun sendSuspend(element: T) { |
| 107 | mutex.lock() |
| 108 | doLockedNext(element) |
| 109 | } |
| 110 | |
Roman Elizarov | db0e22d | 2017-08-29 18:15:57 +0300 | [diff] [blame] | 111 | override val onSend: SelectClause2<T, SendChannel<T>> |
| 112 | get() = this |
| 113 | |
| 114 | // registerSelectSend |
| 115 | @Suppress("PARAMETER_NAME_CHANGED_ON_OVERRIDE") |
| 116 | override fun <R> registerSelectClause2(select: SelectInstance<R>, element: T, block: suspend (SendChannel<T>) -> R) { |
| 117 | mutex.onLock.registerSelectClause2(select, null) { |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 118 | doLockedNext(element) |
Roman Elizarov | db0e22d | 2017-08-29 18:15:57 +0300 | [diff] [blame] | 119 | block(this) |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 120 | } |
Roman Elizarov | db0e22d | 2017-08-29 18:15:57 +0300 | [diff] [blame] | 121 | } |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 122 | |
| 123 | // assert: mutex.isLocked() |
| 124 | private fun doLockedNext(elem: T) { |
| 125 | // check if already closed for send |
Roman Elizarov | 29affbb | 2017-07-21 13:47:41 +0300 | [diff] [blame] | 126 | if (!isActive) { |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 127 | doLockedSignalCompleted() |
Roman Elizarov | 8b38fa2 | 2017-09-27 17:44:31 +0300 | [diff] [blame] | 128 | throw getCancellationException() |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 129 | } |
| 130 | // notify subscriber |
| 131 | try { |
| 132 | subscriber.onNext(elem) |
| 133 | } catch (e: Throwable) { |
| 134 | try { |
| 135 | if (!cancel(e)) |
Roman Elizarov | 9fe5f46 | 2018-02-21 19:05:52 +0300 | [diff] [blame] | 136 | handleCoroutineException(context, e) |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 137 | } finally { |
| 138 | doLockedSignalCompleted() |
| 139 | } |
Roman Elizarov | 8b38fa2 | 2017-09-27 17:44:31 +0300 | [diff] [blame] | 140 | throw getCancellationException() |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 141 | } |
| 142 | /* |
Roman Elizarov | 29affbb | 2017-07-21 13:47:41 +0300 | [diff] [blame] | 143 | There is no sense to check for `isActive` before doing `unlock`, because cancellation/completion might |
| 144 | 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] | 145 | if it fails to acquire the lock that we are still holding). |
| 146 | We have to recheck `isCompleted` after `unlock` anyway. |
| 147 | */ |
| 148 | mutex.unlock() |
Roman Elizarov | 29affbb | 2017-07-21 13:47:41 +0300 | [diff] [blame] | 149 | // recheck isActive |
| 150 | if (!isActive && mutex.tryLock()) |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 151 | doLockedSignalCompleted() |
| 152 | } |
| 153 | |
| 154 | // assert: mutex.isLocked() |
| 155 | private fun doLockedSignalCompleted() { |
| 156 | try { |
Roman Elizarov | 7753f8e | 2017-08-15 11:16:33 +0300 | [diff] [blame] | 157 | if (_signal.value >= CLOSED) { |
| 158 | _signal.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] | 159 | val cause = getCompletionCause() |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 160 | try { |
Roman Elizarov | e89cd68 | 2018-04-25 13:03:40 +0300 | [diff] [blame] | 161 | if (cause != null && cause !is CancellationException) |
Roman Elizarov | 3ef4fca | 2017-07-12 12:42:20 +0300 | [diff] [blame] | 162 | subscriber.onError(cause) |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 163 | else |
| 164 | subscriber.onComplete() |
| 165 | } catch (e: Throwable) { |
Roman Elizarov | 9fe5f46 | 2018-02-21 19:05:52 +0300 | [diff] [blame] | 166 | handleCoroutineException(context, e) |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 167 | } |
| 168 | } |
| 169 | } finally { |
| 170 | mutex.unlock() |
| 171 | } |
| 172 | } |
| 173 | |
Roman Elizarov | 6640b2b | 2018-01-17 19:08:55 +0300 | [diff] [blame] | 174 | override fun onCancellation(cause: Throwable?) { |
Roman Elizarov | 7753f8e | 2017-08-15 11:16:33 +0300 | [diff] [blame] | 175 | if (!_signal.compareAndSet(OPEN, CLOSED)) return // abort, other thread invoked doLockedSignalCompleted |
Roman Elizarov | 331750b | 2017-02-15 17:59:17 +0300 | [diff] [blame] | 176 | if (mutex.tryLock()) // if we can acquire the lock |
| 177 | doLockedSignalCompleted() |
| 178 | } |
| 179 | |
| 180 | // Cancellable impl |
| 181 | override fun cancel() { cancel(cause = null) } |
| 182 | } |