Introduce trySend and tryReceive channel operations as a future replacement for error-prone offer, poll and receiveOrNull

Fixes #974

Author: qwwdfsad

Diff for: kotlinx-coroutines-core/api/kotlinx-coroutines-core.api

@@ -555,6 +555,7 @@ public abstract interface class kotlinx/coroutines/channels/ActorScope : kotlinx
 
 public final class kotlinx/coroutines/channels/ActorScope$DefaultImpls {
 	public static synthetic fun cancel (Lkotlinx/coroutines/channels/ActorScope;)V
+	public static fun poll (Lkotlinx/coroutines/channels/ActorScope;)Ljava/lang/Object;
 }
 
 public abstract interface class kotlinx/coroutines/channels/BroadcastChannel : kotlinx/coroutines/channels/SendChannel {
@@ -566,6 +567,7 @@ public abstract interface class kotlinx/coroutines/channels/BroadcastChannel : k
 public final class kotlinx/coroutines/channels/BroadcastChannel$DefaultImpls {
 	public static synthetic fun cancel$default (Lkotlinx/coroutines/channels/BroadcastChannel;Ljava/lang/Throwable;ILjava/lang/Object;)Z
 	public static synthetic fun cancel$default (Lkotlinx/coroutines/channels/BroadcastChannel;Ljava/util/concurrent/CancellationException;ILjava/lang/Object;)V
+	public static fun offer (Lkotlinx/coroutines/channels/BroadcastChannel;Ljava/lang/Object;)Z
 }
 
 public final class kotlinx/coroutines/channels/BroadcastChannelKt {
@@ -598,6 +600,8 @@ public abstract interface class kotlinx/coroutines/channels/Channel : kotlinx/co
 
 public final class kotlinx/coroutines/channels/Channel$DefaultImpls {
 	public static synthetic fun cancel (Lkotlinx/coroutines/channels/Channel;)V
+	public static fun offer (Lkotlinx/coroutines/channels/Channel;Ljava/lang/Object;)Z
+	public static fun poll (Lkotlinx/coroutines/channels/Channel;)Ljava/lang/Object;
 }
 
 public final class kotlinx/coroutines/channels/Channel$Factory {
@@ -628,7 +632,7 @@ public final class kotlinx/coroutines/channels/ChannelKt {
 public final class kotlinx/coroutines/channels/ChannelResult {
 	public static final field Companion Lkotlinx/coroutines/channels/ChannelResult$Companion;
 	public static final synthetic fun box-impl (Ljava/lang/Object;)Lkotlinx/coroutines/channels/ChannelResult;
-	public static synthetic fun constructor-impl (Ljava/lang/Object;Lkotlin/jvm/internal/DefaultConstructorMarker;)Ljava/lang/Object;
+	public static fun constructor-impl (Ljava/lang/Object;)Ljava/lang/Object;
 	public fun equals (Ljava/lang/Object;)Z
 	public static fun equals-impl (Ljava/lang/Object;Ljava/lang/Object;)Z
 	public static final fun equals-impl0 (Ljava/lang/Object;Ljava/lang/Object;)Z
@@ -645,6 +649,12 @@ public final class kotlinx/coroutines/channels/ChannelResult {
 	public final synthetic fun unbox-impl ()Ljava/lang/Object;
 }
 
+public final class kotlinx/coroutines/channels/ChannelResult$Companion {
+	public final fun closed-JP2dKIU (Ljava/lang/Throwable;)Ljava/lang/Object;
+	public final fun failure-PtdJZtk ()Ljava/lang/Object;
+	public final fun success-JP2dKIU (Ljava/lang/Object;)Ljava/lang/Object;
+}
+
 public final class kotlinx/coroutines/channels/ChannelsKt {
 	public static final fun all (Lkotlinx/coroutines/channels/ReceiveChannel;Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public static final fun any (Lkotlinx/coroutines/channels/ReceiveChannel;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
@@ -789,6 +799,7 @@ public final class kotlinx/coroutines/channels/ConflatedBroadcastChannel : kotli
 	public fun offer (Ljava/lang/Object;)Z
 	public fun openSubscription ()Lkotlinx/coroutines/channels/ReceiveChannel;
 	public fun send (Ljava/lang/Object;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public fun trySend-JP2dKIU (Ljava/lang/Object;)Ljava/lang/Object;
 }
 
 public final class kotlinx/coroutines/channels/ProduceKt {
@@ -804,6 +815,10 @@ public abstract interface class kotlinx/coroutines/channels/ProducerScope : kotl
 	public abstract fun getChannel ()Lkotlinx/coroutines/channels/SendChannel;
 }
 
+public final class kotlinx/coroutines/channels/ProducerScope$DefaultImpls {
+	public static fun offer (Lkotlinx/coroutines/channels/ProducerScope;Ljava/lang/Object;)Z
+}
+
 public abstract interface class kotlinx/coroutines/channels/ReceiveChannel {
 	public abstract synthetic fun cancel ()V
 	public abstract synthetic fun cancel (Ljava/lang/Throwable;)Z
@@ -818,12 +833,14 @@ public abstract interface class kotlinx/coroutines/channels/ReceiveChannel {
 	public abstract fun receive (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public abstract fun receiveCatching-JP2dKIU (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public abstract fun receiveOrNull (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public abstract fun tryReceive-PtdJZtk ()Ljava/lang/Object;
 }
 
 public final class kotlinx/coroutines/channels/ReceiveChannel$DefaultImpls {
 	public static synthetic fun cancel (Lkotlinx/coroutines/channels/ReceiveChannel;)V
 	public static synthetic fun cancel$default (Lkotlinx/coroutines/channels/ReceiveChannel;Ljava/lang/Throwable;ILjava/lang/Object;)Z
 	public static synthetic fun cancel$default (Lkotlinx/coroutines/channels/ReceiveChannel;Ljava/util/concurrent/CancellationException;ILjava/lang/Object;)V
+	public static fun poll (Lkotlinx/coroutines/channels/ReceiveChannel;)Ljava/lang/Object;
 }
 
 public abstract interface class kotlinx/coroutines/channels/SendChannel {
@@ -834,10 +851,12 @@ public abstract interface class kotlinx/coroutines/channels/SendChannel {
 	public abstract fun isFull ()Z
 	public abstract fun offer (Ljava/lang/Object;)Z
 	public abstract fun send (Ljava/lang/Object;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public abstract fun trySend-JP2dKIU (Ljava/lang/Object;)Ljava/lang/Object;
 }
 
 public final class kotlinx/coroutines/channels/SendChannel$DefaultImpls {
 	public static synthetic fun close$default (Lkotlinx/coroutines/channels/SendChannel;Ljava/lang/Throwable;ILjava/lang/Object;)Z
+	public static fun offer (Lkotlinx/coroutines/channels/SendChannel;Ljava/lang/Object;)Z
 }
 
 public final class kotlinx/coroutines/channels/TickerChannelsKt {

Diff for: kotlinx-coroutines-core/common/src/channels/AbstractChannel.kt

@@ -137,20 +137,22 @@ internal abstract class AbstractSendChannel<E>(
         return sendSuspend(element)
     }
 
-    public final override fun offer(element: E): Boolean {
+    public final override fun trySend(element: E): ChannelResult<Unit> {
         val result = offerInternal(element)
         return when {
-            result === OFFER_SUCCESS -> true
+            result === OFFER_SUCCESS -> ChannelResult.success(Unit)
             result === OFFER_FAILED -> {
-                // We should check for closed token on offer as well, otherwise offer won't be linearizable
+                // We should check for closed token on trySend as well, otherwise trySend won't be linearizable
                 // in the face of concurrent close()
                 // See https://github.com/Kotlin/kotlinx.coroutines/issues/359
-                throw recoverStackTrace(helpCloseAndGetSendException(element, closedForSend ?: return false))
+                val closedForSend = closedForSend ?: return ChannelResult.failure()
+                helpClose(closedForSend)
+                ChannelResult.closed(closedForSend.sendException)
             }
             result is Closed<*> -> {
-                throw recoverStackTrace(helpCloseAndGetSendException(element, result))
+                ChannelResult.closed(helpCloseAndGetSendException(element, result))
             }
-            else -> error("offerInternal returned $result")
+            else -> error("trySend returned $result")
         }
     }
 
@@ -632,9 +634,11 @@ internal abstract class AbstractChannel<E>(
     }
 
     @Suppress("UNCHECKED_CAST")
-    public final override fun poll(): E? {
+    public final override fun tryReceive(): ChannelResult<E> {
         val result = pollInternal()
-        return if (result === POLL_FAILED) null else receiveOrNullResult(result)
+        if (result === POLL_FAILED) return ChannelResult.failure()
+        if (result is Closed<*>) return ChannelResult.closed(result.closeCause)
+        return ChannelResult.success(result as E)
     }
 
     @Deprecated(level = DeprecationLevel.HIDDEN, message = "Since 1.2.0, binary compatibility with versions <= 1.1.x")
@@ -905,7 +909,7 @@ internal abstract class AbstractChannel<E>(
         @JvmField val receiveMode: Int
     ) : Receive<E>() {
         fun resumeValue(value: E): Any? = when (receiveMode) {
-            RECEIVE_RESULT -> ChannelResult.value(value)
+            RECEIVE_RESULT -> ChannelResult.success(value)
             else -> value
         }
 
@@ -990,7 +994,7 @@ internal abstract class AbstractChannel<E>(
         @Suppress("UNCHECKED_CAST")
         override fun completeResumeReceive(value: E) {
             block.startCoroutineCancellable(
-                if (receiveMode == RECEIVE_RESULT) ChannelResult.value(value) else value,
+                if (receiveMode == RECEIVE_RESULT) ChannelResult.success(value) else value,
                 select.completion,
                 resumeOnCancellationFun(value)
             )
@@ -1144,7 +1148,7 @@ internal abstract class Receive<in E> : LockFreeLinkedListNode(), ReceiveOrClose
 
 @Suppress("NOTHING_TO_INLINE", "UNCHECKED_CAST")
 private inline fun <E> Any?.toResult(): ChannelResult<E> =
-    if (this is Closed<*>) ChannelResult.closed(closeCause) else ChannelResult.value(this as E)
+    if (this is Closed<*>) ChannelResult.closed(closeCause) else ChannelResult.success(this as E)
 
 @Suppress("NOTHING_TO_INLINE")
 private inline fun <E> Closed<*>.toResult(): ChannelResult<E> = ChannelResult.closed(closeCause)

Diff for: kotlinx-coroutines-core/common/src/channels/Channel.kt

@@ -85,7 +85,23 @@ public interface SendChannel<in E> {
      * then it calls `onUndeliveredElement` before throwing an exception.
      * See "Undelivered elements" section in [Channel] documentation for details on handling undelivered elements.
      */
-    public fun offer(element: E): Boolean
+    public fun offer(element: E): Boolean {
+        val result = trySend(element)
+        if (result.isSuccess) return true
+        throw recoverStackTrace(result.exceptionOrNull() ?: return false)
+    }
+
+    /**
+     * Immediately adds the specified [element] to this channel, if this doesn't violate its capacity restrictions,
+     * and returns the successful result. Otherwise, returns failed or closed result.
+     * This is synchronous variant of [send], which backs off in situations when `send` suspends or throws.
+     *
+     * When `trySend` call returns a non-successful result, it guarantees that the element was not delivered to the consumer, and
+     * it does not call `onUndeliveredElement` that was installed for this channel. If the channel was closed,
+     * then it calls `onUndeliveredElement` before throwing an exception.
+     * See "Undelivered elements" section in [Channel] documentation for details on handling undelivered elements.
+     */
+    public fun trySend(element: E): ChannelResult<Unit>
 
     /**
      * Closes this channel.
@@ -271,11 +287,22 @@ public interface ReceiveChannel<out E> {
     public val onReceiveCatching: SelectClause1<ChannelResult<E>>
 
     /**
-     * Retrieves and removes an element from this channel if its not empty, or returns `null` if the channel is empty
+     * Retrieves and removes an element from this channel if it's not empty or returns `null` if the channel is empty
      * or is [is closed for `receive`][isClosedForReceive] without a cause.
      * It throws the original [close][SendChannel.close] cause exception if the channel has _failed_.
      */
-    public fun poll(): E?
+    public fun poll(): E? {
+        val result = tryReceive()
+        if (result.isSuccess) return result.getOrThrow()
+        throw recoverStackTrace(result.exceptionOrNull() ?: return null)
+    }
+
+    /**
+     * Retrieves and removes an element from this channel if it's not empty, returning a [successful][ChannelResult.success]
+     * result, returns [failed][ChannelResult.failed] result if the channel is empty, and [closed][ChannelResult.closed]
+     * result if the channel is closed.
+     */
+    public fun tryReceive(): ChannelResult<E>
 
     /**
      * Returns a new iterator to receive elements from this channel using a `for` loop.
@@ -315,35 +342,35 @@ public interface ReceiveChannel<out E> {
 
 /**
  * A discriminated union of channel operation result.
- * It encapsulates successful or failed result of a channel operation, or a failed operation to a closed channel with
+ * It encapsulates the successful or failed result of a channel operation or a failed operation to a closed channel with
  * an optional cause.
  *
- * Successful result represents a successful operation with value of type [T], for example, result of [Channel.receiveCatching]
- * operation or a successfully sent element as a result of [Channel.trySend].
+ * The successful result represents a successful operation with a value of type [T], for example,
+ * the result of [Channel.receiveCatching] operation or a successfully sent element as a result of [Channel.trySend].
  *
- * Failed result represents a failed operation attempt to a channel, but it doesn't necessary indicate that the channel is failed.
+ * The failed result represents a failed operation attempt to a channel, but it doesn't necessary indicate that the channel is failed.
  * E.g. when the channel is full, [Channel.trySend] returns failed result, but the channel itself is not in the failed state.
  *
- * Closed result represents an operation attempt to a closed channel and also implies that the operation was failed.
+ * The closed result represents an operation attempt to a closed channel and also implies that the operation was failed.
  */
 @Suppress("UNCHECKED_CAST")
 public inline class ChannelResult<out T>
-internal constructor(private val holder: Any?) {
+@PublishedApi internal constructor(private val holder: Any?) {
     /**
      * Returns `true` if this instance represents a successful
      * operation outcome.
      *
-     * In this case [isFailure] and [isClosed] return false.
+     * In this case [isFailure] and [isClosed] return `false`.
      */
-    public val isSuccess: Boolean get() = holder !is Closed
+    public val isSuccess: Boolean get() = holder !is Failed
 
     /**
-     * Returns true if this instance represents unsuccessful operation.
+     * Returns `true` if this instance represents unsuccessful operation.
      *
      * In this case [isSuccess] returns false, but it does not imply
      * that the channel is failed or closed.
      *
-     * Example of failed operation without an exception and channel being closed
+     * Example of a failed operation without an exception and channel being closed
      * is [Channel.trySend] attempt to a channel that is full.
      */
     public val isFailure: Boolean get() = holder is Failed
@@ -352,7 +379,7 @@ internal constructor(private val holder: Any?) {
      * Returns `true` if this instance represents unsuccessful operation
      * to a closed or cancelled channel.
      *
-     * In this case [isSuccess] returns false, [isFailure] returns `true`, but it does not imply
+     * In this case [isSuccess] returns `false`, [isFailure] returns `true`, but it does not imply
      * that [exceptionOrNull] returns non-null value.
      *
      * It can happen if the channel was [closed][Channel.close] normally without an exception.
@@ -374,7 +401,7 @@ internal constructor(private val holder: Any?) {
     }
 
     /**
-     * Returns the encapsulated exception if this instance represents failure or null if it is success
+     * Returns the encapsulated exception if this instance represents failure or `null` if it is success
      * or unsuccessful operation to closed channel.
      */
     public fun exceptionOrNull(): Throwable? = (holder as? Closed)?.cause
@@ -389,13 +416,21 @@ internal constructor(private val holder: Any?) {
         override fun toString(): String = "Closed($cause)"
     }
 
-    internal companion object {
-        @Suppress("NOTHING_TO_INLINE")
-        internal inline fun <E> value(value: E): ChannelResult<E> =
+    @Suppress("NOTHING_TO_INLINE")
+    @InternalCoroutinesApi
+    public companion object {
+        private val failed = Failed()
+
+        @InternalCoroutinesApi
+        public fun <E> success(value: E): ChannelResult<E> =
             ChannelResult(value)
 
-        @Suppress("NOTHING_TO_INLINE")
-        internal inline fun <E> closed(cause: Throwable?): ChannelResult<E> =
+        @InternalCoroutinesApi
+        public fun <E> failure(): ChannelResult<E> =
+            ChannelResult(failed)
+
+        @InternalCoroutinesApi
+        public fun <E> closed(cause: Throwable?): ChannelResult<E> =
             ChannelResult(Closed(cause))
     }
 

Diff for: kotlinx-coroutines-core/common/src/channels/ConflatedBroadcastChannel.kt

@@ -229,12 +229,12 @@ public class ConflatedBroadcastChannel<E>() : BroadcastChannel<E> {
 
     /**
      * Sends the value to all subscribed receives and stores this value as the most recent state for
-     * future subscribers. This implementation always returns `true`.
-     * It throws exception if the channel [isClosedForSend] (see [close] for details).
+     * future subscribers. This implementation always returns either successful result
+     * or closed with an exception.
      */
-    public override fun offer(element: E): Boolean {
-        offerInternal(element)?.let { throw it.sendException }
-        return true
+    public override fun trySend(element: E): ChannelResult<Unit> {
+        offerInternal(element)?.let { return ChannelResult.closed(it.sendException)  }
+        return ChannelResult.success(Unit)
     }
 
     @Suppress("UNCHECKED_CAST")

Diff for: kotlinx-coroutines-core/common/test/channels/BasicOperationsTest.kt

@@ -142,7 +142,7 @@ class BasicOperationsTest : TestBase() {
         val result = channel.receiveCatching()
         assertEquals(1, result.getOrThrow())
         assertEquals(1, result.getOrNull())
-        assertTrue(ChannelResult.value(1) == result)
+        assertTrue(ChannelResult.success(1) == result)
 
         expect(3)
         launch {

Diff for: kotlinx-coroutines-core/common/test/channels/ChannelReceiveCatchingTest.kt

@@ -45,7 +45,7 @@ class ChannelReceiveCatchingTest : TestBase() {
         assertEquals(1, element.getOrThrow())
         assertEquals(1, element.getOrNull())
         assertEquals("Value(1)", element.toString())
-        assertTrue(ChannelResult.value(1) == element) // Don't box
+        assertTrue(ChannelResult.success(1) == element) // Don't box
         assertFalse(element.isFailure)
         assertFalse(element.isClosed)
 
@@ -54,7 +54,7 @@ class ChannelReceiveCatchingTest : TestBase() {
         assertNull(nullElement.getOrThrow())
         assertNull(nullElement.getOrNull())
         assertEquals("Value(null)", nullElement.toString())
-        assertTrue(ChannelResult.value(null) == nullElement) // Don't box
+        assertTrue(ChannelResult.success(null) == nullElement) // Don't box
         assertFalse(element.isFailure)
         assertFalse(element.isClosed)
 
@@ -113,7 +113,7 @@ class ChannelReceiveCatchingTest : TestBase() {
     fun testReceiveResultChannel() = runTest {
         val channel = Channel<ChannelResult<UInt>>()
         launch {
-            channel.send(ChannelResult.value(1u))
+            channel.send(ChannelResult.success(1u))
             channel.send(ChannelResult.closed(TestException1()))
             channel.close(TestException2())
         }

Diff for: kotlinx-coroutines-core/common/test/channels/TestChannelKind.kt

@@ -44,9 +44,9 @@ private class ChannelViaBroadcast<E>(
     override suspend fun receive(): E = sub.receive()
     override suspend fun receiveOrNull(): E? = sub.receiveOrNull()
     override suspend fun receiveCatching(): ChannelResult<E> = sub.receiveCatching()
-    override fun poll(): E? = sub.poll()
     override fun iterator(): ChannelIterator<E> = sub.iterator()
-    
+    override fun tryReceive(): ChannelResult<E> = sub.tryReceive()
+
     override fun cancel(cause: CancellationException?) = sub.cancel(cause)
 
     // implementing hidden method anyway, so can cast to an internal class

Diff for: kotlinx-coroutines-core/jvm/src/channels/Actor.kt

@@ -164,6 +164,11 @@ private class LazyActorCoroutine<E>(
         return super.offer(element)
     }
 
+    override fun trySend(element: E): ChannelResult<Unit> {
+        start()
+        return super.trySend(element)
+    }
+
     override fun close(cause: Throwable?): Boolean {
         // close the channel _first_
         val closed = super.close(cause)