Consolidate NullSurrogate with Symbol and rename it to NULL

* JvmStatic on unbox did not make sense anyway
* unbox is useful for other symbols

Author: elizarov

kotlinx-coroutines-core/common/src/channels/AbstractChannel.kt

@@ -978,8 +978,7 @@ internal abstract class AbstractChannel<E> : AbstractSendChannel<E>(), Channel<E
 
         @Suppress("UNCHECKED_CAST")
         override fun completeResumeReceive(token: Any) {
-            val value: E = (if (token === NULL_VALUE) null else token) as E
-            block.startCoroutine(value, select.completion)
+            block.startCoroutine(NULL_VALUE.unbox<E>(token), select.completion)
         }
 
         override fun resumeReceiveClosed(closed: Closed<*>) {
@@ -1035,7 +1034,7 @@ internal val SELECT_STARTED: Any = Symbol("SELECT_STARTED")
 
 @JvmField
 @SharedImmutable
-internal val NULL_VALUE: Any = Symbol("NULL_VALUE")
+internal val NULL_VALUE: Symbol = Symbol("NULL_VALUE")
 
 @JvmField
 @SharedImmutable

kotlinx-coroutines-core/common/src/channels/ConflatedBroadcastChannel.kt

@@ -88,17 +88,10 @@ public class ConflatedBroadcastChannel<E>() : BroadcastChannel<E> {
      * The most recently sent element to this channel or `null` when this class is constructed without
      * initial value and no value was sent yet or if it was [closed][close].
      */
-    @Suppress("UNCHECKED_CAST")
-    public val valueOrNull: E? get() {
-        val state = _state.value
-        when (state) {
-            is Closed -> return null
-            is State<*> -> {
-                if (state.value === UNDEFINED) return null
-                return state.value as E
-            }
-            else -> error("Invalid state $state")
-        }
+    public val valueOrNull: E? get() = when (val state = _state.value) {
+        is Closed -> null
+        is State<*> -> UNDEFINED.unbox(state.value)
+        else -> error("Invalid state $state")
     }
 
     public override val isClosedForSend: Boolean get() = _state.value is Closed

kotlinx-coroutines-core/common/src/flow/internal/NullSurrogate.kt

@@ -4,11 +4,13 @@
 
 package kotlinx.coroutines.flow.internal
 
+import kotlinx.coroutines.internal.*
 import kotlin.jvm.*
 
-internal object NullSurrogate {
-
-    @JvmStatic
-    @Suppress("UNCHECKED_CAST")
-    internal fun <T> unbox(value: Any?): T = if (value === NullSurrogate) null as T else value as T
-}
+/**
+ * This value is used a a surrogate `null` value when needed.
+ * It should never leak to the outside world.
+ */
+@JvmField
+@SharedImmutable
+internal val NULL = Symbol("NULL")

kotlinx-coroutines-core/common/src/flow/operators/Delay.kt

@@ -65,7 +65,7 @@ public fun <T> Flow<T>.debounce(timeoutMillis: Long): Flow<T> {
             val values = Channel<Any?>(Channel.CONFLATED) // Actually Any, KT-30796
             // Channel is not closed deliberately as there is no close with value
             val collector = async {
-                collect { value -> values.send(value ?: NullSurrogate) }
+                collect { value -> values.send(value ?: NULL) }
             }
 
             var isDone = false
@@ -79,13 +79,13 @@ public fun <T> Flow<T>.debounce(timeoutMillis: Long): Flow<T> {
                     lastValue?.let { value -> // set timeout when lastValue != null
                         onTimeout(timeoutMillis) {
                             lastValue = null // Consume the value
-                            emit(NullSurrogate.unbox(value))
+                            emit(NULL.unbox(value))
                         }
                     }
 
                     // Close with value 'idiom'
                     collector.onAwait {
-                        if (lastValue != null) emit(NullSurrogate.unbox(lastValue))
+                        if (lastValue != null) emit(NULL.unbox(lastValue))
                         isDone = true
                     }
                 }
@@ -115,7 +115,7 @@ public fun <T> Flow<T>.sample(periodMillis: Long): Flow<T> {
     return flow {
         coroutineScope {
             val values = produce<Any?>(capacity = Channel.CONFLATED) {  // Actually Any, KT-30796
-                collect { value -> send(value ?: NullSurrogate) }
+                collect { value -> send(value ?: NULL) }
             }
 
             var isDone = false
@@ -136,7 +136,7 @@ public fun <T> Flow<T>.sample(periodMillis: Long): Flow<T> {
                     ticker.onReceive {
                         val value = lastValue ?: return@onReceive
                         lastValue = null // Consume the value
-                        emit(NullSurrogate.unbox(value))
+                        emit(NULL.unbox(value))
                     }
                 }
             }

kotlinx-coroutines-core/common/src/flow/operators/Distinct.kt

@@ -25,7 +25,7 @@ public fun <T> Flow<T>.distinctUntilChanged(): Flow<T> = distinctUntilChangedBy
 @FlowPreview
 public fun <T, K> Flow<T>.distinctUntilChangedBy(keySelector: (T) -> K): Flow<T> =
     flow {
-        var previousKey: Any? = NullSurrogate
+        var previousKey: Any? = NULL
         collect { value ->
             val key = keySelector(value)
             if (previousKey != key) {

kotlinx-coroutines-core/common/src/flow/operators/Merge.kt

@@ -138,7 +138,7 @@ private class SerializingFlatMapCollector<T>(
 
     public suspend fun emit(value: T) {
         if (!inProgressLock.tryAcquire()) {
-            channel.send(value ?: NullSurrogate)
+            channel.send(value ?: NULL)
             if (inProgressLock.tryAcquire()) {
                 helpEmit()
             }
@@ -154,7 +154,7 @@ private class SerializingFlatMapCollector<T>(
         while (true) {
             var element = channel.poll()
             while (element != null) { // TODO receive or closed (#330)
-                downstream.emit(NullSurrogate.unbox(element))
+                downstream.emit(NULL.unbox(element))
                 element = channel.poll()
             }
 

kotlinx-coroutines-core/common/src/flow/operators/Zip.kt

@@ -61,14 +61,14 @@ public fun <T1, T2, R> Flow<T1>.combineLatest(other: Flow<T2>, transform: suspen
                 onReceive(firstIsClosed, firstChannel, { firstIsClosed = true }) { value ->
                     firstValue = value
                     if (secondValue !== null) {
-                        emit(transform(NullSurrogate.unbox(firstValue), NullSurrogate.unbox(secondValue)))
+                        emit(transform(NULL.unbox(firstValue), NULL.unbox(secondValue)))
                     }
                 }
 
                 onReceive(secondIsClosed, secondChannel, { secondIsClosed = true }) { value ->
                     secondValue = value
                     if (firstValue !== null) {
-                        emit(transform(NullSurrogate.unbox(firstValue), NullSurrogate.unbox(secondValue)))
+                        emit(transform(NULL.unbox(firstValue), NULL.unbox(secondValue)))
                     }
                 }
             }
@@ -163,7 +163,7 @@ internal fun <T, R> Flow<T>.combineLatest(vararg others: Flow<T>, arrayFactory:
                         if (latestValues.all { it !== null }) {
                             val arguments = arrayFactory()
                             for (index in 0 until size) {
-                                arguments[index] = NullSurrogate.unbox(latestValues[index])
+                                arguments[index] = NULL.unbox(latestValues[index])
                             }
                             emit(transform(arguments as Array<T>))
                         }
@@ -191,7 +191,7 @@ private inline fun SelectBuilder<Unit>.onReceive(
 private fun CoroutineScope.asFairChannel(flow: Flow<*>): ReceiveChannel<Any> = produce {
     val channel = channel as ChannelCoroutine<Any>
     flow.collect { value ->
-        channel.sendFair(value ?: NullSurrogate)
+        channel.sendFair(value ?: NULL)
     }
 }
 
@@ -235,8 +235,8 @@ public fun <T1, T2, R> Flow<T1>.zip(other: Flow<T2>, transform: suspend (T1, T2)
                 if (!otherIterator.hasNext()) {
                     return@consumeEach
                 }
-                val secondValue = NullSurrogate.unbox<T2>(otherIterator.next())
-                emit(transform(NullSurrogate.unbox(value), NullSurrogate.unbox(secondValue)))
+                val secondValue = NULL.unbox<T2>(otherIterator.next())
+                emit(transform(NULL.unbox(value), NULL.unbox(secondValue)))
             }
         } catch (e: AbortFlowException) {
             // complete
@@ -249,6 +249,6 @@ public fun <T1, T2, R> Flow<T1>.zip(other: Flow<T2>, transform: suspend (T1, T2)
 // Channel has any type due to onReceiveOrNull. This will be fixed after receiveOrClosed
 private fun CoroutineScope.asChannel(flow: Flow<*>): ReceiveChannel<Any> = produce {
     flow.collect { value ->
-        channel.send(value ?: NullSurrogate)
+        channel.send(value ?: NULL)
     }
 }

kotlinx-coroutines-core/common/src/flow/terminal/Reduce.kt

@@ -18,18 +18,18 @@ import kotlin.jvm.*
  */
 @FlowPreview
 public suspend fun <S, T : S> Flow<T>.reduce(operation: suspend (accumulator: S, value: T) -> S): S {
-    var accumulator: Any? = NullSurrogate
+    var accumulator: Any? = NULL
 
     collect { value ->
-        accumulator = if (accumulator !== NullSurrogate) {
+        accumulator = if (accumulator !== NULL) {
             @Suppress("UNCHECKED_CAST")
             operation(accumulator as S, value)
         } else {
             value
         }
     }
 
-    if (accumulator === NullSurrogate) throw UnsupportedOperationException("Empty flow can't be reduced")
+    if (accumulator === NULL) throw UnsupportedOperationException("Empty flow can't be reduced")
     @Suppress("UNCHECKED_CAST")
     return accumulator as S
 }
@@ -56,13 +56,13 @@ public suspend inline fun <T, R> Flow<T>.fold(
  */
 @FlowPreview
 public suspend fun <T> Flow<T>.single(): T {
-    var result: Any? = NullSurrogate
+    var result: Any? = NULL
     collect { value ->
-        if (result !== NullSurrogate) error("Expected only one element")
+        if (result !== NULL) error("Expected only one element")
         result = value
     }
 
-    if (result === NullSurrogate) throw NoSuchElementException("Expected at least one element")
+    if (result === NULL) throw NoSuchElementException("Expected at least one element")
     @Suppress("UNCHECKED_CAST")
     return result as T
 }

kotlinx-coroutines-core/common/src/internal/Symbol.kt

@@ -11,4 +11,7 @@ package kotlinx.coroutines.internal
  */
 internal class Symbol(val symbol: String) {
     override fun toString(): String = symbol
+
+    @Suppress("UNCHECKED_CAST", "NOTHING_TO_INLINE")
+    inline fun <T> unbox(value: Any?): T = if (value === this) null as T else value as T
 }