Merge remote-tracking branch 'origin/develop'

Author: elizarov

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

@@ -16,17 +16,11 @@
 
 package kotlinx.coroutines.experimental.channels
 
-import kotlinx.coroutines.experimental.CancellableContinuation
-import kotlinx.coroutines.experimental.DisposableHandle
+import kotlinx.coroutines.experimental.*
 import kotlinx.coroutines.experimental.internal.*
-import kotlinx.coroutines.experimental.intrinsics.startCoroutineUndispatched
-import kotlinx.coroutines.experimental.removeOnCancel
-import kotlinx.coroutines.experimental.selects.ALREADY_SELECTED
-import kotlinx.coroutines.experimental.selects.SelectClause1
-import kotlinx.coroutines.experimental.selects.SelectClause2
-import kotlinx.coroutines.experimental.selects.SelectInstance
-import kotlinx.coroutines.experimental.suspendAtomicCancellableCoroutine
-import kotlin.coroutines.experimental.startCoroutine
+import kotlinx.coroutines.experimental.intrinsics.*
+import kotlinx.coroutines.experimental.selects.*
+import kotlin.coroutines.experimental.*
 
 /**
  * Abstract send channel. It is a base class for all send channel implementations.
@@ -374,6 +368,37 @@ public abstract class AbstractSendChannel<E> : SendChannel<E> {
         }
     }
 
+    // ------ debug ------
+
+    public override fun toString() =
+        "$classSimpleName@$hexAddress{$queueDebugStateString}$bufferDebugString"
+
+    private val queueDebugStateString: String
+        get() {
+            val head = queue.next
+            if (head === queue) return "EmptyQueue"
+            var result = when (head) {
+                is Closed<*> -> head.toString()
+                is Receive<*> -> "ReceiveQueued"
+                is Send -> "SendQueued"
+                else -> "UNEXPECTED:$head" // should not happen
+            }
+            val tail = queue.prev
+            if (tail !== head) {
+                result += ",queueSize=${countQueueSize()}"
+                if (tail is Closed<*>) result += ",closedForSend=$tail"
+            }
+            return result
+        }
+
+    private fun countQueueSize(): Int {
+        var size = 0
+        queue.forEach<LockFreeLinkedListNode> { size++ }
+        return size
+    }
+
+    protected open val bufferDebugString: String get() = ""
+
     // ------ private ------
 
     private class SendSelect<E, R>(

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

@@ -16,11 +16,10 @@
 
 package kotlinx.coroutines.experimental.channels
 
-import kotlinx.coroutines.experimental.selects.ALREADY_SELECTED
-import kotlinx.coroutines.experimental.selects.SelectInstance
-import java.util.concurrent.CopyOnWriteArrayList
-import java.util.concurrent.locks.ReentrantLock
-import kotlin.concurrent.withLock
+import kotlinx.coroutines.experimental.selects.*
+import java.util.concurrent.*
+import java.util.concurrent.locks.*
+import kotlin.concurrent.*
 
 /**
  * Broadcast channel with array buffer of a fixed [capacity].
@@ -353,4 +352,9 @@ class ArrayBroadcastChannel<E>(
             return result
         }
     }
+
+    // ------ debug ------
+
+    override val bufferDebugString: String
+        get() = "(buffer:capacity=${buffer.size},size=$size)"
 }

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

@@ -16,10 +16,9 @@
 
 package kotlinx.coroutines.experimental.channels
 
-import kotlinx.coroutines.experimental.selects.ALREADY_SELECTED
-import kotlinx.coroutines.experimental.selects.SelectInstance
-import java.util.concurrent.locks.ReentrantLock
-import kotlin.concurrent.withLock
+import kotlinx.coroutines.experimental.selects.*
+import java.util.concurrent.locks.*
+import kotlin.concurrent.*
 
 /**
  * Channel with array buffer of a fixed [capacity].
@@ -245,4 +244,9 @@ public open class ArrayChannel<E>(
         // then clean all queued senders
         super.cleanupSendQueueOnCancel()
     }
+
+    // ------ debug ------
+
+    override val bufferDebugString: String
+        get() = "(buffer:capacity=${buffer.size},size=$size)"
 }

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

@@ -577,8 +577,19 @@ public inline suspend fun <E, C : SendChannel<E>> ReceiveChannel<E>.filterIndexe
  * This function [consumes][consume] all elements of the original [ReceiveChannel].
  */
 // todo: mark predicate with crossinline modifier when it is supported: https://youtrack.jetbrains.com/issue/KT-19159
-public fun <E> ReceiveChannel<E>.filterNot(predicate: suspend (E) -> Boolean): ReceiveChannel<E> =
-    filter { !predicate(it) }
+public fun <E> ReceiveChannel<E>.filterNot(context: CoroutineContext = Unconfined, predicate: suspend (E) -> Boolean): ReceiveChannel<E> =
+    filter(context) { !predicate(it) }
+
+/**
+ * Returns a channel containing all elements not matching the given [predicate].
+ *
+ * The operation is _intermediate_ and _stateless_.
+ * This function [consumes][consume] all elements of the original [ReceiveChannel].
+ *
+ * @suppress **Deprecated**: For binary compatibility only
+ */
+@Deprecated("For binary compatibility only", level = DeprecationLevel.HIDDEN)
+public fun <E> ReceiveChannel<E>.filterNot(predicate: suspend (E) -> Boolean): ReceiveChannel<E> = filterNot(predicate = predicate)
 
 /**
  * Returns a channel containing all elements that are not `null`.