From 1678b637a6cec1176d795d96483e7caf562a506d Mon Sep 17 00:00:00 2001
From: InsanusMokrassar <ovsyannikov.alexey95@gmail.com>
Date: Mon, 13 Jan 2020 10:34:35 +0600
Subject: [PATCH] fillup readme with asFlow info and add one more test

---
 README.md                                     | 26 ++++++++++++++
 .../insanusmokrassar/krontab/utils/FailJob.kt |  8 +++++
 .../krontab/utils/SchedulerFlow.kt            | 35 +++++++++++++++++++
 3 files changed, 69 insertions(+)
 create mode 100644 src/commonTest/kotlin/com/insanusmokrassar/krontab/utils/FailJob.kt

diff --git a/README.md b/README.md
index ec1b0ba..1852a64 100644
--- a/README.md
+++ b/README.md
@@ -11,6 +11,7 @@ runtime of applications.
 | [ How to use: Including in project ](#including-in-project) |
 | [ How to use: Config from string ](#config-from-string) |
 | [ How to use: Config via builder (DSL preview) ](#config-via-builder) |
+| [ How to use: KronScheduler as a Flow ](#KronScheduler-as-a-Flow) |
 
 ## How to use
 
@@ -124,3 +125,28 @@ kronScheduler.doInfinity {
 ```
 
 All of these examples will do the same things: print `Called` message every five seconds.
+
+### KronScheduler as a Flow
+
+Any `KronScheduler`can e converted to a `Flow<DateTime` using extension `asFlow`:
+
+```kotlin
+val kronScheduler = buildSchedule {
+    seconds {
+        0 every 1
+    }
+}
+
+val flow = kronScheduler.asFlow()
+```
+
+So, in this case any operations related to flow are available and it is expected tt th will work correctly. For example,
+it is possible to use this flow with `takeWhile`:
+
+```kotlin
+flow.takeWhile {
+    condition()
+}.collect {
+    action()
+}
+```
diff --git a/src/commonTest/kotlin/com/insanusmokrassar/krontab/utils/FailJob.kt b/src/commonTest/kotlin/com/insanusmokrassar/krontab/utils/FailJob.kt
new file mode 100644
index 0000000..43530fe
--- /dev/null
+++ b/src/commonTest/kotlin/com/insanusmokrassar/krontab/utils/FailJob.kt
@@ -0,0 +1,8 @@
+package com.insanusmokrassar.krontab.utils
+
+import kotlinx.coroutines.*
+
+fun CoroutineScope.createFailJob(forTimeMillis: Long) = launch {
+    delay(forTimeMillis)
+    throw IllegalStateException("This job must be completed at before this exception happen")
+}
diff --git a/src/commonTest/kotlin/com/insanusmokrassar/krontab/utils/SchedulerFlow.kt b/src/commonTest/kotlin/com/insanusmokrassar/krontab/utils/SchedulerFlow.kt
index 29eca3b..2bb913f 100644
--- a/src/commonTest/kotlin/com/insanusmokrassar/krontab/utils/SchedulerFlow.kt
+++ b/src/commonTest/kotlin/com/insanusmokrassar/krontab/utils/SchedulerFlow.kt
@@ -1,6 +1,7 @@
 package com.insanusmokrassar.krontab.utils
 
 import com.insanusmokrassar.krontab.builder.buildSchedule
+import com.soywiz.klock.DateTime
 import kotlinx.coroutines.*
 import kotlinx.coroutines.flow.collect
 import kotlinx.coroutines.flow.takeWhile
@@ -31,4 +32,38 @@ class SchedulerFlowTests {
             assertEquals(mustBeCollected, collected)
         }
     }
+
+    @Test
+    fun testThatFlowIsCorrectlyWorkEverySecondWithMuchOfEmitters() {
+        val kronScheduler = buildSchedule {
+            seconds {
+                0 every 1
+            }
+        }
+
+        val flow = kronScheduler.asFlow()
+
+        runTest {
+            val testsCount = 10
+            val failJob = it.createFailJob((testsCount * 2) * 1000L)
+            val mustBeCollected = 10
+            val answers = (0 until testsCount).map { _ ->
+                it.async {
+                    var collected = 0
+                    flow.takeWhile {
+                        collected < mustBeCollected
+                    }.collect {
+                        collected++
+                    }
+                    collected
+                }
+            }.awaitAll()
+
+            failJob.cancel()
+
+            answers.forEach {
+                assertEquals(mustBeCollected, it)
+            }
+        }
+    }
 }