Adds Stream.callback

arrow-fx-coroutines/src/main/kotlin/arrow/fx/coroutines/predef.kt

@@ -1,8 +1,11 @@
 package arrow.fx.coroutines
 
-inline infix fun <A, B, C> ((A) -> B).andThen(crossinline f: (B) -> C): (A) -> C =
+internal inline infix fun <A, B, C> ((A) -> B).andThen(crossinline f: (B) -> C): (A) -> C =
   { a -> f(this(a)) }
 
+internal inline infix fun <A, B, C> (suspend (A) -> B).andThen(crossinline f: suspend (B) -> C): suspend (A) -> C =
+  { a: A -> f(this(a)) }
+
 infix fun <A> A.prependTo(fa: Iterable<A>): List<A> =
   listOf(this) + fa
 

arrow-fx-coroutines/src/main/kotlin/arrow/fx/coroutines/stream/ParJoin.kt

@@ -16,7 +16,7 @@ import arrow.fx.coroutines.stream.concurrent.SignallingAtomic
 import arrow.fx.coroutines.uncancellable
 
 /**
- * Nondeterministically merges a stream of streams (`outer`) in to a single stream,
+ * Non-deterministically merges a stream of streams (`outer`) in to a single stream,
  * opening at most `maxOpen` streams at any point in time.
  *
  * The outer stream is evaluated and each resulting inner stream is run concurrently,
@@ -67,7 +67,7 @@ suspend fun <O> stop(
   outputQ.enqueue1(None)
 }
 
-suspend fun <O> decrementRunning(
+internal suspend fun <O> decrementRunning(
   done: SignallingAtomic<Option<Option<Throwable>>>,
   outputQ: NoneTerminatedQueue<Chunk<O>>,
   running: SignallingAtomic<Int>
@@ -77,7 +77,7 @@ suspend fun <O> decrementRunning(
     Pair(now, (if (now == 0) suspend { stop(done, outputQ, None) } else suspend { Unit }))
   }.invoke()
 
-suspend fun incrementRunning(running: SignallingAtomic<Int>): Unit =
+internal suspend fun incrementRunning(running: SignallingAtomic<Int>): Unit =
   running.update { it + 1 }
 
 // runs inner stream, each stream is forked. terminates when killSignal is true if fails will enq in queue failure
@@ -121,7 +121,7 @@ internal suspend fun <O> runInner(
   }
 
 // runs the outer stream, interrupts when kill == true, and then decrements the `running`
-suspend fun <O> Stream<Stream<O>>.runOuter(
+internal suspend fun <O> Stream<Stream<O>>.runOuter(
   done: SignallingAtomic<Option<Option<Throwable>>>,
   outputQ: NoneTerminatedQueue<Chunk<O>>,
   running: SignallingAtomic<Int>,
@@ -146,9 +146,17 @@ suspend fun <O> Stream<Stream<O>>.runOuter(
 
 // awaits when all streams (outer + inner) finished,
 // and then collects result of the stream (outer + inner) execution
-suspend fun signalResult(done: SignallingAtomic<Option<Option<Throwable>>>): Unit =
+internal suspend fun signalResult(done: SignallingAtomic<Option<Option<Throwable>>>): Unit =
   done.get().flatten().fold({ Unit }, { throw it })
 
+/**
+ * Merges both Streams into an Stream of A and B represented by Either<A, B>.
+ * This operation is equivalent to a normal merge but for different types.
+ */
+fun <A, B> Stream<A>.either(other: Stream<B>): Stream<Either<A, B>> =
+  Stream(this.map { Either.Left(it) }, other.map { Either.Right(it) })
+    .parJoin(2)
+
 fun <O> Stream<Stream<O>>.parJoin(maxOpen: Int): Stream<O> {
   require(maxOpen > 0) { "maxOpen must be > 0, was: $maxOpen" }
   return Stream.effect {
@@ -180,6 +188,6 @@ fun <O> Stream<Stream<O>>.parJoin(maxOpen: Int): Stream<O> {
   }.flatten()
 }
 
-/** Like [parJoin] but races all inner streams simultaneously. */
+/** Like [parJoin] but races all inner streams simultaneously without limit. */
 fun <O> Stream<Stream<O>>.parJoinUnbounded(): Stream<O> =
   parJoin(Int.MAX_VALUE)

arrow-fx-coroutines/src/main/kotlin/arrow/fx/coroutines/stream/Stream.kt

@@ -1037,7 +1037,7 @@ inline fun <A> StreamOf<A>.fix(): Stream<A> =
   }
 
   /**
-   * Determinsitically zips elements, terminating when the end of either branch is reached naturally.
+   * Deterministically zips elements, terminating when the end of either branch is reached naturally.
    *
    * ```kotlin:ank:playground
    * import arrow.fx.coroutines.stream.*
@@ -1069,7 +1069,7 @@ inline fun <A> StreamOf<A>.fix(): Stream<A> =
     zipWith(that) { x, _ -> x }
 
   /**
-   * Determinsitically zips elements using the specified function,
+   * Deterministically zips elements using the specified function,
    * terminating when the end of either branch is reached naturally.
    *
    * ```kotlin:ank:playground
@@ -1405,13 +1405,13 @@ inline fun <A> StreamOf<A>.fix(): Stream<A> =
    * Run the supplied effectful action at the end of this stream, regardless of how the stream terminates.
    */
   fun onFinalize(f: suspend () -> Unit): Stream<O> =
-    bracket({ Unit }, { f.invoke() }).flatMap { this }
+    bracket({ Unit }, { f() }).flatMap { this }
 
   /**
    * Like [onFinalize] but provides the reason for finalization as an `ExitCase`.
    */
   fun onFinalizeCase(f: suspend (ExitCase) -> Unit): Stream<O> =
-    Stream.bracketCase({ Unit }) { _, ec -> f(ec) }.flatMap { this }
+    bracketCase({ Unit }) { _, ec -> f(ec) }.flatMap { this }
 
   /**
    * Interrupts the stream, when `haltOnSignal` finishes its evaluation.
@@ -1433,9 +1433,9 @@ inline fun <A> StreamOf<A>.fix(): Stream<A> =
   /**
    * Introduces an explicit scope.
    *
-   * Scopes are normally introduced automatically, when using `bracket` or xsimilar
+   * Scopes are normally introduced automatically, when using `bracket` or similar
    * operations that acquire resources and run finalizers. Manual scope introduction
-   * is useful when using [onFinalizeWeak]/[onFinalizeCaseWeak], where no scope
+   * is useful when using [bracketWeak]/[bracketCaseWeak], where no scope
    * is introduced.
    */
   fun scope(): Stream<O> =
@@ -1971,7 +1971,7 @@ inline fun <A> StreamOf<A>.fix(): Stream<A> =
 }
 
 /**
- * Determinsitically zips elements, terminating when the ends of both branches
+ * Deterministically zips elements, terminating when the ends of both branches
  * are reached naturally, padding the left branch with `pad1` and padding the right branch
  * with `pad2` as necessary.
  *
@@ -1991,7 +1991,7 @@ fun <O, B> Stream<O>.zipAll(that: Stream<B>, pad1: O, pad2: B): Stream<Pair<O, B
   zipAllWith(that, pad1, pad2, ::Pair)
 
 /**
- * Determinsitically zips elements with the specified function, terminating
+ * Deterministically zips elements with the specified function, terminating
  * when the ends of both branches are reached naturally, padding the left
  * branch with `pad1` and padding the right branch with `pad2` as necessary.
  *
@@ -2172,13 +2172,32 @@ fun <O : Any> Stream<O?>.filterNull(): Stream<O> =
  * //sampleEnd
  * ```
  */
-fun <O : Any> Stream<O?>.terminateOnNull(): Stream<O> =
+fun <O> Stream<O>.terminateOnNull(): Stream<O> =
+  terminateOn { it == null }
+
+/**
+ * Halts the input stream when the condition is true.
+ *
+ * ```kotlin:ank:playground
+ * import arrow.fx.coroutines.stream.*
+ *
+ * //sampleStart
+ * suspend fun main(): Unit =
+ *   Stream(1, 2, 3, 4)
+ *     .terminateOn { it == 3 }
+ *     .compile()
+ *     .toList()
+ *     .let(::println) //[1, 2]
+ * //sampleEnd
+ * ```
+ */
+fun <O> Stream<O>.terminateOn(terminator: (O) -> Boolean): Stream<O> =
   asPull.repeat { pull ->
     pull.unconsOrNull().flatMap { uncons ->
       when (uncons) {
         null -> Pull.just(null)
         else -> {
-          when (val idx = uncons.head.indexOfFirst { it == null }) {
+          when (val idx = uncons.head.indexOfFirst(terminator)) {
             0 -> Pull.just(null)
             null -> Pull.output<O>(uncons.head.map { it!! }).map { uncons.tail }
             else -> Pull.output(uncons.head.take(idx).map { it!! }).map { null }
@@ -2243,6 +2262,10 @@ fun <O> Stream<Option<O>>.terminateOnNone(): Stream<O> =
     }
   }.stream()
 
+/**
+ * Wraps the inner values in Option as Some and adds an None at the end of the Stream to signal completion.
+ * Note that this doesn't actually limit an infinite stream.
+ */
 fun <O> Stream<O>.noneTerminate(): Stream<Option<O>> =
   map { Some(it) }.append { Stream.just(None) }
 
@@ -2258,7 +2281,7 @@ fun <O> emptyStream(): Stream<O> =
   Stream.empty()
 
 /**
- * Determinsitically interleaves elements, starting on the left, terminating when the end of either branch is reached naturally.
+ * Deterministically interleaves elements, starting on the left, terminating when the end of either branch is reached naturally.
  *
  * ```kotlin:ank:playground
  * import arrow.fx.coroutines.stream.*
@@ -2277,7 +2300,7 @@ fun <O> Stream<O>.interleave(that: Stream<O>): Stream<O> =
   zip(that).flatMap { (o1, o2) -> Stream(o1, o2) }
 
 /**
- * Determinsitically interleaves elements, starting on the left, terminating when the ends of both branches are reached naturally.
+ * Deterministically interleaves elements, starting on the left, terminating when the ends of both branches are reached naturally.
  *
  * ```kotlin:ank:playground
  * import arrow.fx.coroutines.stream.*

arrow-fx-coroutines/src/main/kotlin/arrow/fx/coroutines/stream/callbacks.kt

@@ -0,0 +1,123 @@
+package arrow.fx.coroutines.stream
+
+import arrow.core.Either
+import arrow.fx.coroutines.CancelToken
+import arrow.fx.coroutines.ExitCase
+import arrow.fx.coroutines.ForkConnected
+import arrow.fx.coroutines.UnsafePromise
+import arrow.fx.coroutines.guaranteeCase
+import arrow.fx.coroutines.stream.concurrent.Queue
+import kotlin.coroutines.Continuation
+import kotlin.coroutines.EmptyCoroutineContext
+import kotlin.coroutines.startCoroutine
+import kotlin.experimental.ExperimentalTypeInference
+
+private object END
+
+interface EmitterSyntax<A> {
+  fun onCancel(cancelF: suspend () -> Unit): Unit
+  fun emit(a: A): Unit
+  fun emit(chunk: Chunk<A>): Unit
+  fun emit(iterable: Iterable<A>): Unit
+  fun emit(vararg aas: A): Unit
+  fun end(): Unit
+}
+
+/**
+ * Creates a Stream from the given suspended block callback, allowing to emit, set cancel effects and end the emission.
+ *
+ * ```kotlin:ank:playground
+ * import arrow.fx.coroutines.stream.*
+ *
+ * //sampleStart
+ * suspend fun main(): Unit =
+ *   Stream.callback {
+ *       onCancel { /* cancel something */ }
+ *       emit(1)
+ *       emit(2, 3, 4)
+ *       end()
+ *     }
+ *     .compile()
+ *     .toList()
+ *     .let(::println) //[1, 2, 3, 4]
+ * //sampleEnd
+ * ```
+ *
+ * Note that if neither `end()` nor other limit operators such as `take(N)` are called,
+ * then the Stream will never end.
+ *
+ * Cancellation or errors might happen any time during emission, so it's recommended to call `onCancel` as early as
+ * possible, otherwise it's not guaranteed to be called.
+ */
+// @OptIn(ExperimentalTypeInference::class) in 1.3.70
+@UseExperimental(ExperimentalTypeInference::class)
+fun <A> Stream.Companion.callback(@BuilderInference f: suspend EmitterSyntax<A>.() -> Unit): Stream<A> =
+  force {
+    val q = Queue.unbounded<Any?>()
+    val error = UnsafePromise<Throwable>()
+    val cancel = UnsafePromise<CancelToken>()
+
+    ForkConnected { emitterCallback(f, cancel, error, q) }
+
+    (q.dequeue()
+      .interruptWhen { Either.Left(error.join()) }
+      .terminateOn { it === END } as Stream<Chunk<A>>)
+      .flatMap(::chunk)
+      .onFinalizeCase {
+        when (it) {
+          is ExitCase.Cancelled -> {
+            when (val r = cancel.tryGet()) {
+              null -> Unit // This means the user didn't set the onCancel or set it too late
+              else -> r.getOrNull()!!.invoke()
+            }
+          }
+        }
+      }
+  }
+
+private suspend fun <A> emitterCallback(
+  f: suspend EmitterSyntax<A>.() -> Unit,
+  cancel: UnsafePromise<CancelToken>,
+  error: UnsafePromise<Throwable>,
+  q: Queue<Any?>
+): Unit {
+  val cb = { ch: Any? ->
+    suspend {
+      q.enqueue1(ch)
+    }.startCoroutine(Continuation(EmptyCoroutineContext) { r ->
+      r.fold({ Unit }, { e -> error.complete(Result.success(e)) })
+    })
+  }
+
+  val emitter = object : EmitterSyntax<A> {
+    override fun emit(a: A) {
+      emit(Chunk.just(a))
+    }
+
+    override fun emit(chunk: Chunk<A>) {
+      cb(chunk)
+    }
+
+    override fun emit(iterable: Iterable<A>) {
+      cb(Chunk.iterable(iterable))
+    }
+
+    override fun emit(vararg aas: A) {
+      cb(Chunk(*aas))
+    }
+
+    override fun end() {
+      cb(END)
+    }
+
+    override fun onCancel(cancelF: suspend () -> Unit) {
+      cancel.complete(Result.success(CancelToken(cancelF)))
+    }
+  }
+
+  guaranteeCase({ emitter.f() }, { exit ->
+    when (exit) {
+      is ExitCase.Failure -> error.complete(Result.success(exit.failure))
+    }
+  })
+}

arrow-fx-coroutines/src/main/kotlin/arrow/fx/coroutines/stream/concurrent/Queue.kt

@@ -134,8 +134,8 @@ interface Queue<A> : Enqueue<A>, Dequeue1<A>, Dequeue<A> {
     suspend fun <A> bounded(maxSize: Int): Queue<A> =
       fromStrategy(Strategy.boundedFifo(maxSize))
 
-    /** Creates a FILO queue with the specified size bound. */
-    suspend fun <A> boundedLife(maxSize: Int): Queue<A> =
+    /** Creates a LIFO queue with the specified size bound. */
+    suspend fun <A> boundedLifo(maxSize: Int): Queue<A> =
       fromStrategy(Strategy.boundedLifo(maxSize))
 
     /** Creates a queue which stores the last `maxSize` enqueued elements and which never blocks on enqueue. */

arrow-fx-coroutines/src/test/kotlin/arrow/fx/coroutines/stream/BracketTest.kt

@@ -20,7 +20,6 @@ import io.kotest.property.arbitrary.int
 import io.kotest.property.arbitrary.list
 import io.kotest.property.arbitrary.long
 import io.kotest.property.arbitrary.string
-import io.kotest.property.checkAll
 
 class BracketTest : StreamSpec(spec = {
 

arrow-fx-coroutines/src/test/kotlin/arrow/fx/coroutines/stream/StreamTest.kt

@@ -1,5 +1,7 @@
 package arrow.fx.coroutines.stream
 
+import arrow.core.None
+import arrow.core.Some
 import arrow.core.extensions.list.foldable.combineAll
 import arrow.core.extensions.list.foldable.foldMap
 import arrow.core.extensions.monoid
@@ -527,4 +529,25 @@ class StreamTest : StreamSpec(spec = {
         .toList() shouldBe List(n) { i }
     }
   }
+
+  "terminateOn" {
+    Stream(1, 2, 3, 4)
+      .terminateOn { it % 3 == 0 }
+      .compile()
+      .toList() shouldBe listOf(1, 2)
+  }
+
+  "terminateOnNull" {
+    Stream(1, 2, null, 4)
+      .terminateOnNull()
+      .compile()
+      .toList() shouldBe listOf(1, 2)
+  }
+
+  "terminateOnNone" {
+    Stream(Some(1), Some(2), None, Some(4))
+      .terminateOnNone()
+      .compile()
+      .toList() shouldBe listOf(1, 2)
+  }
 })