Turn all consumeEach and withLock into inline suspend functions

Author: elizarov

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/channels/Channels.kt

@@ -21,17 +21,29 @@ internal const val DEFAULT_CLOSE_MESSAGE = "Channel was closed"
 /**
  * Performs the given [action] for each received element.
  */
-// :todo: make it inline when this bug is fixed: https://youtrack.jetbrains.com/issue/KT-16448
-public suspend fun <E> ReceiveChannel<E>.consumeEach(action: suspend (E) -> Unit) {
+public inline suspend fun <E> ReceiveChannel<E>.consumeEach(action: (E) -> Unit) {
     for (element in this) action(element)
 }
 
+/**
+ * @suppress: **Deprecated**: binary compatibility with old code
+ */
+@Deprecated("binary compatibility with old code", level = DeprecationLevel.HIDDEN)
+public suspend fun <E> ReceiveChannel<E>.consumeEach(action: suspend (E) -> Unit) =
+    consumeEach { action(it) }
+
 /**
  * Subscribes to this [BroadcastChannel] and performs the specified action for each received element.
  */
-// :todo: make it inline when this bug is fixed: https://youtrack.jetbrains.com/issue/KT-16448
-public suspend fun <E> BroadcastChannel<E>.consumeEach(action: suspend (E) -> Unit) {
+public inline suspend fun <E> BroadcastChannel<E>.consumeEach(action: (E) -> Unit) {
     openSubscription().use { channel ->
         for (x in channel) action(x)
     }
 }
+
+/**
+ * @suppress: **Deprecated**: binary compatibility with old code
+ */
+@Deprecated("binary compatibility with old code", level = DeprecationLevel.HIDDEN)
+public suspend fun <E> BroadcastChannel<E>.consumeEach(action: suspend (E) -> Unit) =
+    consumeEach { action(it) }

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/sync/Mutex.kt

@@ -121,8 +121,7 @@ public fun Mutex(locked: Boolean = false): Mutex = MutexImpl(locked)
  *
  * @return the return value of the action.
  */
-// :todo: this function needs to be make inline as soon as this bug is fixed: https://youtrack.jetbrains.com/issue/KT-16448
-public suspend fun <T> Mutex.withLock(owner: Any? = null, action: suspend () -> T): T {
+public inline suspend fun <T> Mutex.withLock(owner: Any? = null, action: () -> T): T {
     lock(owner)
     try {
         return action()
@@ -131,24 +130,26 @@ public suspend fun <T> Mutex.withLock(owner: Any? = null, action: suspend () ->
     }
 }
 
+/**
+ * @suppress: **Deprecated**: binary compatibility with old code
+ */
+@Deprecated("binary compatibility with old code", level = DeprecationLevel.HIDDEN)
+public suspend fun <T> Mutex.withLock(owner: Any? = null, action: suspend () -> T): T =
+    withLock(owner) { action() }
+
 /**
  * @suppress: **Deprecated**: Use [withLock]
  */
 @Deprecated("Use `withLock(owner, action)", level = DeprecationLevel.HIDDEN)
-public suspend fun <T> Mutex.withLock(action: suspend () -> T): T = withLock(null, action)
+public suspend fun <T> Mutex.withLock(action: suspend () -> T): T =
+    withLock { action() }
 
 /**
  * @suppress: **Deprecated**: Use [withLock]
  */
 @Deprecated("Use `withLock`", replaceWith = ReplaceWith("withLock(action)"))
-public suspend fun <T> Mutex.withMutex(action: suspend () -> T): T {
-    lock()
-    try {
-        return action()
-    } finally {
-        unlock()
-    }
-}
+public suspend fun <T> Mutex.withMutex(action: suspend () -> T): T =
+    withLock { action() }
 
 private val LOCK_FAIL = Symbol("LOCK_FAIL")
 private val ENQUEUE_FAIL = Symbol("ENQUEUE_FAIL")

reactive/kotlinx-coroutines-reactive/src/main/kotlin/kotlinx/coroutines/experimental/reactive/Channel.kt

@@ -58,13 +58,19 @@ public operator fun <T> Publisher<T>.iterator() = openSubscription().iterator()
 /**
  * Subscribes to this [Publisher] and performs the specified action for each received element.
  */
-// :todo: make it inline when this bug is fixed: https://youtrack.jetbrains.com/issue/KT-16448
-public suspend fun <T> Publisher<T>.consumeEach(action: suspend (T) -> Unit) {
+public inline suspend fun <T> Publisher<T>.consumeEach(action: (T) -> Unit) {
     openSubscription().use { channel ->
         for (x in channel) action(x)
     }
 }
 
+/**
+ * @suppress: **Deprecated**: binary compatibility with old code
+ */
+@Deprecated("binary compatibility with old code", level = DeprecationLevel.HIDDEN)
+public suspend fun <T> Publisher<T>.consumeEach(action: suspend (T) -> Unit) =
+    consumeEach { action(it) }
+
 private class SubscriptionChannel<T> : LinkedListChannel<T>(), SubscriptionReceiveChannel<T>, Subscriber<T> {
     @Volatile
     @JvmField

reactive/kotlinx-coroutines-rx1/src/main/kotlin/kotlinx/coroutines/experimental/rx1/RxChannel.kt

@@ -59,13 +59,19 @@ public operator fun <T> Observable<T>.iterator() = openSubscription().iterator()
 /**
  * Subscribes to this [Observable] and performs the specified action for each received element.
  */
-// :todo: make it inline when this bug is fixed: https://youtrack.jetbrains.com/issue/KT-16448
-public suspend fun <T> Observable<T>.consumeEach(action: suspend (T) -> Unit) {
+public inline suspend fun <T> Observable<T>.consumeEach(action: (T) -> Unit) {
     openSubscription().use { channel ->
         for (x in channel) action(x)
     }
 }
 
+/**
+ * @suppress: **Deprecated**: binary compatibility with old code
+ */
+@Deprecated("binary compatibility with old code", level = DeprecationLevel.HIDDEN)
+public suspend fun <T> Observable<T>.consumeEach(action: suspend (T) -> Unit) =
+    consumeEach { action(it) }
+
 private class SubscriptionChannel<T> : LinkedListChannel<T>(), SubscriptionReceiveChannel<T> {
     @JvmField
     val subscriber: ChannelSubscriber = ChannelSubscriber()

reactive/kotlinx-coroutines-rx2/src/main/kotlin/kotlinx/coroutines/experimental/rx2/RxChannel.kt

@@ -71,8 +71,7 @@ public operator fun <T> ObservableSource<T>.iterator() = openSubscription().iter
 /**
  * Subscribes to this [MaybeSource] and performs the specified action for each received element.
  */
-// :todo: make it inline when this bug is fixed: https://youtrack.jetbrains.com/issue/KT-16448
-public suspend fun <T> MaybeSource<T>.consumeEach(action: suspend (T) -> Unit) {
+public inline suspend fun <T> MaybeSource<T>.consumeEach(action: (T) -> Unit) {
     openSubscription().use { channel ->
         for (x in channel) action(x)
     }
@@ -81,13 +80,19 @@ public suspend fun <T> MaybeSource<T>.consumeEach(action: suspend (T) -> Unit) {
 /**
  * Subscribes to this [ObservableSource] and performs the specified action for each received element.
  */
-// :todo: make it inline when this bug is fixed: https://youtrack.jetbrains.com/issue/KT-16448
-public suspend fun <T> ObservableSource<T>.consumeEach(action: suspend (T) -> Unit) {
+public inline suspend fun <T> ObservableSource<T>.consumeEach(action: (T) -> Unit) {
     openSubscription().use { channel ->
         for (x in channel) action(x)
     }
 }
 
+/**
+ * @suppress: **Deprecated**: binary compatibility with old code
+ */
+@Deprecated("binary compatibility with old code", level = DeprecationLevel.HIDDEN)
+public suspend fun <T> ObservableSource<T>.consumeEach(action: suspend (T) -> Unit) =
+    consumeEach { action(it) }
+
 private class SubscriptionChannel<T> : LinkedListChannel<T>(), SubscriptionReceiveChannel<T>, Observer<T>, MaybeObserver<T> {
     @Volatile
     var subscription: Disposable? = null