Leave only public ticker function, introduce TickerMode enum

Author: elizarov

binary-compatibility-validator/reference-public-api/kotlinx-coroutines-core.txt

@@ -849,12 +849,15 @@ public final class kotlinx/coroutines/experimental/channels/SubscriptionReceiveC
 }
 
 public final class kotlinx/coroutines/experimental/channels/TickerChannelsKt {
-	public static final fun adjustingTicker (JLjava/util/concurrent/TimeUnit;JLkotlin/coroutines/experimental/CoroutineContext;)Lkotlinx/coroutines/experimental/channels/ReceiveChannel;
-	public static synthetic fun adjustingTicker$default (JLjava/util/concurrent/TimeUnit;JLkotlin/coroutines/experimental/CoroutineContext;ILjava/lang/Object;)Lkotlinx/coroutines/experimental/channels/ReceiveChannel;
-	public static final fun fixedTicker (JLjava/util/concurrent/TimeUnit;JLkotlin/coroutines/experimental/CoroutineContext;)Lkotlinx/coroutines/experimental/channels/ReceiveChannel;
-	public static synthetic fun fixedTicker$default (JLjava/util/concurrent/TimeUnit;JLkotlin/coroutines/experimental/CoroutineContext;ILjava/lang/Object;)Lkotlinx/coroutines/experimental/channels/ReceiveChannel;
-	public static final fun ticker (JLjava/util/concurrent/TimeUnit;JLkotlin/coroutines/experimental/CoroutineContext;Z)Lkotlinx/coroutines/experimental/channels/ReceiveChannel;
-	public static synthetic fun ticker$default (JLjava/util/concurrent/TimeUnit;JLkotlin/coroutines/experimental/CoroutineContext;ZILjava/lang/Object;)Lkotlinx/coroutines/experimental/channels/ReceiveChannel;
+	public static final fun ticker (JLjava/util/concurrent/TimeUnit;JLkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/channels/TickerMode;)Lkotlinx/coroutines/experimental/channels/ReceiveChannel;
+	public static synthetic fun ticker$default (JLjava/util/concurrent/TimeUnit;JLkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/channels/TickerMode;ILjava/lang/Object;)Lkotlinx/coroutines/experimental/channels/ReceiveChannel;
+}
+
+public final class kotlinx/coroutines/experimental/channels/TickerMode : java/lang/Enum {
+	public static final field FIXED_DELAY Lkotlinx/coroutines/experimental/channels/TickerMode;
+	public static final field FIXED_PERIOD Lkotlinx/coroutines/experimental/channels/TickerMode;
+	public static fun valueOf (Ljava/lang/String;)Lkotlinx/coroutines/experimental/channels/TickerMode;
+	public static fun values ()[Lkotlinx/coroutines/experimental/channels/TickerMode;
 }
 
 public final class kotlinx/coroutines/experimental/intrinsics/CancellableKt {

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

@@ -4,118 +4,103 @@ import kotlinx.coroutines.experimental.*
 import kotlinx.coroutines.experimental.timeunit.*
 import kotlin.coroutines.experimental.*
 
-
 /**
- * Creates rendezvous channel which produces the first item after the given initial delay and subsequent items with the
- * given delay between them. Backpressure is guaranteed by [RendezvousChannel].
- *
- * This channel stops producing elements immediately after [ReceiveChannel.cancel] invocation.
- * **Note** producer to this channel is dispatched via [Unconfined] dispatcher and started eagerly
- *
- * @param initialDelay delay after which the first item will be produced
- * @param delay delay between each element
- * @param unit unit of time that applies to [initialDelay] and [delay]
- * @param coroutineContext context of the producing coroutine
- * @param fixedPeriod whether producer should use fixed delay or should attempt to adjust delay when consumer cannot keep up
+ * Mode for [ticker] function.
  */
-public fun ticker(
-    delay: Long,
-    unit: TimeUnit = TimeUnit.MILLISECONDS,
-    initialDelay: Long = delay,
-    coroutineContext: CoroutineContext = Unconfined,
-    fixedPeriod: Boolean = false
-): ReceiveChannel<Unit> {
-    return if (fixedPeriod) fixedTicker(delay, unit, initialDelay, coroutineContext)
-    else adjustingTicker(delay, unit, initialDelay, coroutineContext)
+enum class TickerMode {
+    /**
+     * Adjust delay to maintain fixed period if consumer cannot keep up or is otherwise slow.
+     * **This is a default mode.**
+     *
+     * ```
+     * val channel = ticker(delay = 100)
+     * delay(350) // 250 ms late
+     * println(channel.poll()) // prints Unit
+     * println(channel.poll()) // prints null
+     *
+     * delay(50)
+     * println(channel.poll()) // prints Unit, delay was adjusted
+     * delay(50)
+     * println(channel.poll()) // prints null, we'are not late relatively to previous element
+     * ```
+     */
+    FIXED_PERIOD,
+
+    /**
+     * Maintains fixed delay between produced elements if consumer cannot keep up or it otherwise slow.
+     */
+    FIXED_DELAY
 }
 
 /**
- * Creates rendezvous channel which produces the first item after the given initial delay and subsequent items after
- * given delay.
+ * Creates a channel that produces the first item after the given initial delay and subsequent items with the
+ * given delay between them.
  *
- * Producer to resulting channel tries to adjust delay if consumer cannot keep up:
- * ```
- * val channel = adjustingTicker(delay = 100)
- * delay(350) // 250 ms late
- * println(channel.poll()) // prints Unit
- * println(channel.poll()) // prints null
- *
- * delay(50)
- * println(channel.poll() // prints Unit, delay was adjusted
- * delay(50)
- * println(channel.poll() // prints null, we'are not late relatively to previous element
- * ```
+ * The resulting channel is a [rendezvous channel][RendezvousChannel]. When receiver from this channel does not keep
+ * up with receiving the elements from this channel, they are not being being send due to backpressure. The actual
+ * timing behavior of ticker in this case is controlled by [mode] parameter which
+ * is set to [TickerMode.FIXED_PERIOD] by default. See [TickerMode] for other details.
  *
  * This channel stops producing elements immediately after [ReceiveChannel.cancel] invocation.
- * **Note** producer to this channel is dispatched via [Unconfined] dispatcher and started eagerly
  *
- * @param initialDelay delay after which the first item will be produced
- * @param delay delay between each elements
- * @param unit unit of time that applies to [initialDelay] and [delay]
- * @param coroutineContext context of the producing coroutine
+ * **Note** producer to this channel is dispatched via [Unconfined] dispatcher by default and started eagerly.
+ *
+ * @param delay delay between each element.
+ * @param unit unit of time that applies to [initialDelay] and [delay] (in milliseconds by default).
+ * @param initialDelay delay after which the first element will be produced (it is equal to [delay] by default).
+ * @param context context of the producing coroutine.
+ * @param mode specifies behavior when elements are not received ([FIXED_PERIOD][TickerMode.FIXED_PERIOD] by default).
  */
-public fun adjustingTicker(
+public fun ticker(
     delay: Long,
     unit: TimeUnit = TimeUnit.MILLISECONDS,
     initialDelay: Long = delay,
-    coroutineContext: CoroutineContext = Unconfined
+    context: CoroutineContext = EmptyCoroutineContext,
+    mode: TickerMode = TickerMode.FIXED_PERIOD
 ): ReceiveChannel<Unit> {
     require(delay >= 0) { "Expected non-negative delay, but has $delay" }
     require(initialDelay >= 0) { "Expected non-negative initial delay, but has $initialDelay" }
-
-    val result = RendezvousChannel<Unit>()
-    launch(coroutineContext) {
-        delay(initialDelay, unit)
-
-        val delayNs = unit.toNanos(delay)
-        var deadline = timeSource.nanoTime()
-        while (true) {
-            deadline += delayNs
-            result.send(Unit)
-            val now = timeSource.nanoTime()
-            val nextDelay = (deadline - now).coerceAtLeast(0)
-            if (nextDelay == 0L && delayNs != 0L) {
-                val adjustedDelay = delayNs - (now - deadline) % delayNs
-                deadline = now + adjustedDelay
-                delay(adjustedDelay, java.util.concurrent.TimeUnit.NANOSECONDS)
-            } else {
-                delay(nextDelay, java.util.concurrent.TimeUnit.NANOSECONDS)
-            }
+    return produce(Unconfined + context, capacity = 0) {
+        when(mode) {
+            TickerMode.FIXED_PERIOD -> fixedPeriodTicker(delay, unit, initialDelay, channel)
+            TickerMode.FIXED_DELAY -> fixedDelayTicker(delay, unit, initialDelay, channel)
         }
     }
-
-    return result
 }
 
-/**
- * Creates rendezvous channel which produces the first item after the given initial delay and subsequent items after
- * given delay **after** consumption of the previous item.
- *
- * This channel stops producing elements immediately after [ReceiveChannel.cancel] invocation.
- * **Note** producer to this channel is dispatched via [Unconfined] dispatcher and started eagerly
- *
- * @param initialDelay delay after which the first item will be produced
- * @param delay delay between each elements
- * @param unit unit of time that applies to [initialDelay] and [delay]
- * @param coroutineContext context of the producing coroutine
- */
-public fun fixedTicker(
+private suspend fun fixedPeriodTicker(
     delay: Long,
-    unit: TimeUnit = TimeUnit.MILLISECONDS,
-    initialDelay: Long = delay,
-    coroutineContext: CoroutineContext = Unconfined
-): ReceiveChannel<Unit> {
-    require(delay >= 0) { "Expected non-negative delay, but has $delay" }
-    require(initialDelay >= 0) { "Expected non-negative initial delay, but has $initialDelay" }
-
-    val result = RendezvousChannel<Unit>()
-    launch(coroutineContext) {
-        delay(initialDelay, unit)
-        while (true) {
-            result.send(Unit)
-            delay(delay, unit)
+    unit: TimeUnit,
+    initialDelay: Long,
+    channel: SendChannel<Unit>
+) {
+    var deadline = timeSource.nanoTime() + unit.toNanos(initialDelay)
+    delay(initialDelay, unit)
+    val delayNs = unit.toNanos(delay)
+    while (true) {
+        deadline += delayNs
+        channel.send(Unit)
+        val now = timeSource.nanoTime()
+        val nextDelay = (deadline - now).coerceAtLeast(0)
+        if (nextDelay == 0L && delayNs != 0L) {
+            val adjustedDelay = delayNs - (now - deadline) % delayNs
+            deadline = now + adjustedDelay
+            delay(adjustedDelay, java.util.concurrent.TimeUnit.NANOSECONDS)
+        } else {
+            delay(nextDelay, java.util.concurrent.TimeUnit.NANOSECONDS)
         }
     }
+}
 
-    return result
+private suspend fun fixedDelayTicker(
+    delay: Long,
+    unit: TimeUnit,
+    initialDelay: Long,
+    channel: SendChannel<Unit>
+) {
+    delay(initialDelay, unit)
+    while (true) {
+        channel.send(Unit)
+        delay(delay, unit)
+    }
 }

core/kotlinx-coroutines-core/src/test/kotlin/guide/example-channel-10.kt

@@ -21,24 +21,24 @@ import kotlinx.coroutines.experimental.*
 import kotlinx.coroutines.experimental.channels.*
 
 fun main(args: Array<String>) = runBlocking<Unit> {
- val tickerChannel = ticker(delay = 100, initialDelay = 0) // create ticker channel
+    val tickerChannel = ticker(delay = 100, initialDelay = 0) // create ticker channel
     var nextElement = withTimeoutOrNull(1) { tickerChannel.receive() }
     println("Initial element is available immediately: $nextElement") // Initial delay hasn't passed yet
 
     nextElement = withTimeoutOrNull(50) { tickerChannel.receive() } // All subsequent elements has 100ms delay
     println("Next element is not ready in 50 ms: $nextElement")
 
-    nextElement = withTimeoutOrNull(51) { tickerChannel.receive() }
+    nextElement = withTimeoutOrNull(60) { tickerChannel.receive() }
     println("Next element is ready in 100 ms: $nextElement")
 
     // Emulate large consumption delays
-    println("Consumer pause in 150ms")
+    println("Consumer pauses for 150ms")
     delay(150)
     // Next element is available immediately
     nextElement = withTimeoutOrNull(1) { tickerChannel.receive() }
     println("Next element is available immediately after large consumer delay: $nextElement")
     // Note that the pause between `receive` calls is taken into account and next element arrives faster
-    nextElement = withTimeoutOrNull(60) { tickerChannel.receive() } // 60 instead of 50 to mitigate scheduler delays
+    nextElement = withTimeoutOrNull(60) { tickerChannel.receive() } 
     println("Next element is ready in 50ms after consumer pause in 150ms: $nextElement")
 
     tickerChannel.cancel() // indicate that no more elements are needed

core/kotlinx-coroutines-core/src/test/kotlin/guide/test/GuideTest.kt

@@ -367,7 +367,7 @@ class GuideTest {
             "Initial element is available immediately: kotlin.Unit",
             "Next element is not ready in 50 ms: null",
             "Next element is ready in 100 ms: kotlin.Unit",
-            "Consumer pause in 150ms",
+            "Consumer pauses for 150ms",
             "Next element is available immediately after large consumer delay: kotlin.Unit",
             "Next element is ready in 50ms after consumer pause in 150ms: kotlin.Unit"
         )

core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/channels/TickerChannelCommonTest.kt

@@ -9,7 +9,6 @@ import kotlin.test.*
 
 @RunWith(Parameterized::class)
 class TimerChannelCommonTest(private val channelFactory: Channel) : TestBase() {
-
     companion object {
         @Parameterized.Parameters(name = "{0}")
         @JvmStatic
@@ -18,12 +17,14 @@ class TimerChannelCommonTest(private val channelFactory: Channel) : TestBase() {
     }
 
     enum class Channel {
-        DELAY {
-            override fun invoke(delay: Long, initialDelay: Long) = adjustingTicker(delay, initialDelay = initialDelay)
+        FIXED_PERIOD {
+            override fun invoke(delay: Long, initialDelay: Long) =
+                ticker(delay, initialDelay = initialDelay, mode = TickerMode.FIXED_PERIOD)
         },
 
         FIXED_DELAY {
-            override fun invoke(delay: Long, initialDelay: Long) = fixedTicker(delay, initialDelay = initialDelay)
+            override fun invoke(delay: Long, initialDelay: Long) =
+                ticker(delay, initialDelay = initialDelay, mode = TickerMode.FIXED_DELAY)
         };
 
         abstract operator fun invoke(delay: Long, initialDelay: Long = 0): ReceiveChannel<Unit>

core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/channels/TickerChannelTest.kt

@@ -4,10 +4,9 @@ import kotlinx.coroutines.experimental.*
 import org.junit.*
 
 class TickerChannelTest : TestBase() {
-
     @Test
     fun testFixedDelayChannelBackpressure() = runTest {
-        val delayChannel = fixedTicker(delay = 100, initialDelay = 0)
+        val delayChannel = ticker(delay = 100, initialDelay = 0, mode = TickerMode.FIXED_DELAY)
         delayChannel.checkNotEmpty()
         delayChannel.checkEmpty()
 

coroutines-guide.md

@@ -1669,36 +1669,36 @@ The first four elements are added to the buffer and the sender suspends when try
 
 ### Ticker channels
 
-Ticker channel is a special rendezvous channel, which produces `Unit` every time given delay passes since last consumption from this channel.
-Though it may seem to be useless standalone, it is a useful building block to create complex time-based [produce] operators, using this channel as one of [select] clauses and performing "timeout" action in its [onReceive][ReceiveChannel.onReceive].
+Ticker channel is a special rendezvous channel that produces `Unit` every time given delay passes since last consumption from this channel.
+Though it may seem to be useless standalone, it is a useful building block to create complex time-based [produce] 
+pipelines and operators that do windowing and other time-dependend processing.
+Ticker channel can be used in [select] to perform "on tick" action.
 
-To create such channel, use factory method [ticker] and to indicate that no further elements are needed use [ReceiveChannel.cancel] method on it.
+To create such channel use a factory method [ticker]. 
+To indicate that no further elements are needed use [ReceiveChannel.cancel] method on it.
 
 Now let's see how it works in practice:
-<!--- INCLUDE  
-import kotlinx.coroutines.experimental.channels.*
--->
 
 ```kotlin
 fun main(args: Array<String>) = runBlocking<Unit> {
- val tickerChannel = ticker(delay = 100, initialDelay = 0) // create ticker channel
+    val tickerChannel = ticker(delay = 100, initialDelay = 0) // create ticker channel
     var nextElement = withTimeoutOrNull(1) { tickerChannel.receive() }
     println("Initial element is available immediately: $nextElement") // Initial delay hasn't passed yet
 
     nextElement = withTimeoutOrNull(50) { tickerChannel.receive() } // All subsequent elements has 100ms delay
     println("Next element is not ready in 50 ms: $nextElement")
 
-    nextElement = withTimeoutOrNull(51) { tickerChannel.receive() }
+    nextElement = withTimeoutOrNull(60) { tickerChannel.receive() }
     println("Next element is ready in 100 ms: $nextElement")
 
     // Emulate large consumption delays
-    println("Consumer pause in 150ms")
+    println("Consumer pauses for 150ms")
     delay(150)
     // Next element is available immediately
     nextElement = withTimeoutOrNull(1) { tickerChannel.receive() }
     println("Next element is available immediately after large consumer delay: $nextElement")
     // Note that the pause between `receive` calls is taken into account and next element arrives faster
-    nextElement = withTimeoutOrNull(60) { tickerChannel.receive() } // 60 instead of 50 to mitigate scheduler delays
+    nextElement = withTimeoutOrNull(60) { tickerChannel.receive() } 
     println("Next element is ready in 50ms after consumer pause in 150ms: $nextElement")
 
     tickerChannel.cancel() // indicate that no more elements are needed
@@ -1713,16 +1713,18 @@ It prints following lines:
 Initial element is available immediately: kotlin.Unit
 Next element is not ready in 50 ms: null
 Next element is ready in 100 ms: kotlin.Unit
-Consumer pause in 150ms
+Consumer pauses for 150ms
 Next element is available immediately after large consumer delay: kotlin.Unit
 Next element is ready in 50ms after consumer pause in 150ms: kotlin.Unit
 ```
 
 <!--- TEST -->
 
-Note that [ticker] is aware of possible consumer pauses and adapts next element delay if a pause occurs. 
-[fixedTicker] doesn't do so and produces elements with fixed delay after consumption, but both have built-in backpressure. 
-via [RendezvousChannel]
+Note that [ticker] is aware of possible consumer pauses and, by default, adjusts next produced element 
+delay if a pause occurs, trying to maintain a fixed rate of produced elements.
+ 
+Optionally, a `mode` parameters equal to [TickerMode.FIXED_DELAY] can be specified to maintain a fixed
+delay between elements.  
 
 ### Channels are fair
 
@@ -2502,12 +2504,10 @@ Channel was closed
 [produce]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/produce.html
 [consumeEach]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/consume-each.html
 [Channel()]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-channel.html
-[ReceiveChannel.onReceive]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-receive-channel/on-receive.html
 [ticker]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/ticker.html
 [ReceiveChannel.cancel]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-receive-channel/cancel.html
-[fixedTicker]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/fixed-ticker.html
-[RendezvousChannel]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-rendezvous-channel/index.html
 [actor]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/actor.html
+[ReceiveChannel.onReceive]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-receive-channel/on-receive.html
 [ReceiveChannel.onReceiveOrNull]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-receive-channel/on-receive-or-null.html
 [SendChannel.onSend]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-send-channel/on-send.html
 <!--- INDEX kotlinx.coroutines.experimental.selects -->