Stream.kt

package arrow.fx.coroutines.stream

import arrow.core.Either
import arrow.core.Either.Left
import arrow.core.Either.Right
import arrow.core.None
import arrow.core.Option
import arrow.core.Some
import arrow.core.extensions.list.foldable.foldLeft
import arrow.core.identity
import arrow.fx.coroutines.ComputationPool
import arrow.fx.coroutines.ExitCase
import arrow.fx.coroutines.Fiber
import arrow.fx.coroutines.ForkAndForget
import arrow.fx.coroutines.Promise
import arrow.fx.coroutines.Resource
import arrow.fx.coroutines.forkAndForget
import arrow.fx.coroutines.guaranteeCase
import arrow.fx.coroutines.stream.concurrent.Signal
import arrow.typeclasses.Monoid
import arrow.typeclasses.Semigroup
import kotlinx.coroutines.delay
import java.util.concurrent.TimeoutException
import kotlin.coroutines.CoroutineContext
import kotlin.random.Random
import kotlin.time.Duration

class ForStream private constructor() {
  companion object
}

typealias StreamOf<A> = arrow.Kind<ForStream, A>

inline fun <A> StreamOf<A>.fix(): Stream<A> =
  this as Stream<A>

@Deprecated("Deprecated in favor of Flow")
/*inline*/ class Stream<out O> internal constructor(internal val asPull: Pull<O, Unit>) : StreamOf<O> {

  /**
   * Creates a stream whose elements are generated by applying [f] to each output of
   * the source stream and concatenated all of the results.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream(1, 2, 3)
   *     .flatMap { i -> Stream(i, i, i) }
   *     .toList()
   *     .let(::println) // [1, 1, 1, 2, 2, 2, 3, 3, 3]
   * //sampleEnd
   * ```
   */
  fun <B> flatMap(f: (O) -> Stream<B>): Stream<B> =
    Stream(asPull.flatMapOutput { o -> f(o).asPull })

  /**
   * Returns a stream of [O] values wrapped in [Either.Right] until the first error, which is emitted wrapped in [Either.Left].
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream(1, 2, 3)
   *     .append { Stream.raiseError(RuntimeException("Boom!")) }
   *     .append { Stream(4, 5, 6) }
   *     .attempt()
   *     .toList()
   *     .let(::println) // [Right(b=1), Right(b=2), Right(b=3), Left(a=java.lang.RuntimeException: Boom!)]
   * //sampleEnd
   * ```
   */
  fun attempt(): Stream<Either<Throwable, O>> =
    map { Right(it) }
      .handleErrorWith { e -> just(Left(e)) }

  /**
   * Retries on failure, returning a stream of attempts that can
   * be manipulated with standard stream operations such as `take`,
   * `collectFirst` and `interruptWhen`.
   *
   * Note: The resulting stream does *not* automatically halt at the
   * first successful attempt. Also see `retry`.
   */
  fun attempts(delays: Stream<Duration>): Stream<Either<Throwable, O>> =
    attempt().append { delays.flatMap { d -> sleep_(d).append { attempt() } } }

  /**
   * Returns a [Pull] that writes all output to the [Pull].
   */
  fun asPull(): Pull<O, Unit> = asPull

  /**
   * Behaves like the identity function, but requests `n` elements at a time from the input.
   *
   *```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit {
   *   val buf = mutableListOf<String>()
   *
   *   Stream.range(0..100)
   *     .effectTap { i -> buf.add(">$i") }
   *     .buffer(4)
   *     .effectTap { i -> buf.add("<$i") }
   *     .take(10)
   *     .toList()
   *     .let(::println) // [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
   *
   *   println(buf.joinToString()) // [>0, >1, >2, >3, <0, <1, <2, <3, >4, >5, >6, >7, <4, <5, <6, <7, >8, >9, >10, >11, <8, <9]
   * }
   * //sampleEnd
   * ```
   */
  fun buffer(n: Int): Stream<O> =
    asPull.repeat { pull ->
      pull.unconsN(n, allowFewer = true).flatMap { unconsN ->
        when (unconsN) {
          null -> Pull.just(null)
          else -> Pull.output(unconsN.head).map { unconsN.tail }
        }
      }
    }.stream()

  /**
   * Behaves like the identity stream, but emits no output until the source is exhausted.
   *
   *```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit {
   *   val buf = mutableListOf<String>()
   *
   *   Stream.range(0..10)
   *     .effectTap { i -> buf.add(">$i") }
   *     .bufferAll()
   *     .effectTap { i -> buf.add("<$i") }
   *     .take(4)
   *     .toList()
   *     .let(::println) // [0, 1, 2, 3]
   *
   *   println(buf.joinToString()) // [>0, >1, >2, >3, >4, >5, >6, >7, >8, >9, >10, <0, <1, <2, <3]
   * }
   * //sampleEnd
   * ```
   */
  fun bufferAll(): Stream<O> =
    bufferBy { true }

  /**
   * Behaves like the identity stream, but requests elements from its
   * input in blocks that end whenever the predicate switches from true to false.
   *
   *```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit {
   *   val buf = mutableListOf<String>()
   *
   *   Stream.range(0 until 10)
   *     .effectTap { i -> buf.add(">$i") }
   *     .bufferBy { it % 2 == 0 }
   *     .effectTap { i -> buf.add("<$i") }
   *     .toList()
   *     .let(::println) // [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
   *
   *   println(buf.joinToString()) // [>0, >1, <0, <1, >2, >3, <2, <3, >4, >5, <4, <5, >6, >7, <6, <7, >8, >9, <8, <9]
   * }
   * //sampleEnd
   * ```
   */
  fun bufferBy(f: (O) -> Boolean): Stream<O> {
    fun go(buffer: List<Chunk<O>>, last: Boolean, s: Pull<O, Unit>): Pull<O, Unit> =
      s.unconsOrNull().flatMap { uncons ->
        when (uncons) {
          null ->
            buffer.reversed().foldLeft(Pull.just(Unit) as Pull<O, Unit>) { acc, c ->
              acc.flatMap { Pull.output(c) }
            }
          else -> {
            val (out, buf, newLast) =
              uncons.head.fold(Triple(emptyList<Chunk<O>>(), emptyList<O>(), last)) { (out, buf, last), i ->
                val cur = f(i)
                if (!cur && last) Triple(Chunk.iterable(buf + i) prependTo out, emptyList(), cur)
                else Triple(out, buf + i, cur)
              }

            if (out.isEmpty()) go(Chunk.iterable(buf) prependTo buffer, newLast, uncons.tail)
            else {
              val outBuffer = buffer.reversed().fold(Pull.just(Unit) as Pull<O, Unit>) { acc, c ->
                acc.flatMap { Pull.output(c) }
              }
              val outAll = out.reversed().fold(outBuffer) { acc, c -> acc.flatMap { Pull.output(c) } }
              outAll.flatMap { go(listOf(Chunk.iterable(buf)), newLast, uncons.tail) }
            }
          }
        }
      }

    return go(emptyList(), false, asPull).stream()
  }

  /**
   * Skips the first element that matches the predicate.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream.range(1..10)
   *     .delete { it % 2 == 0 }
   *     .toList()
   *     .let(::println) // [1, 3, 4, 5, 6, 7, 8, 9, 10]
   * //sampleEnd
   * ```
   */
  fun delete(p: (O) -> Boolean): Stream<O> =
    asPull
      .takeWhile { o -> !p(o) }
      .flatMap { it.drop(1) }
      .stream()

  /**
   * Drops [n] elements of the input, then echoes the rest.
   * Alias for [drop].
   **/
  fun drop(n: Int): Stream<O> =
    drop(n.toLong())

  /**
   * Drops [n] elements of the input, then echoes the rest.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream.range(0..10)
   *     .drop(5)
   *     .toList()
   *     .let(::println) // [5, 6, 7, 8, 9, 10]
   * //sampleEnd
   * ```
   */
  fun drop(n: Long): Stream<O> =
    asPull.drop(n).stream()

  /**
   * Emits all elements of the input except the first one.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream.range(0..5)
   *     .tail()
   *     .toList()
   *     .let(::println) // [1, 2, 3, 4, 5]
   * //sampleEnd
   * ```
   */
  fun tail(): Stream<O> =
    drop(1)

  fun dropLast(n: Int): Stream<O> =
    dropLast(n.toLong())

  /**
   * Outputs all but the last `n` elements of the input.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream.range(0..10)
   *     .dropLast(5)
   *     .toList()
   *     .let(::println) // [0, 1, 2, 3, 4]
   * //sampleEnd
   * ```
   */
  fun dropLast(n: Long): Stream<O> =
    if (n <= 0) this
    else {
      fun go(acc: Chunk.Queue<O>, pull: Pull<O, Unit>): Pull<O, Unit> =
        pull.unconsOrNull().flatMap { uncons ->
          when (uncons) {
            null -> Pull.done
            else -> {
              val all = acc.enqueue(uncons.head)
              all
                .dropLast(n.toInt())
                .chunks
                .fold(Pull.done<O>()) { acc, c ->
                  acc.flatMap { Pull.output(c) }
                }.flatMap { go(all.takeLast(n.toInt()), uncons.tail) }
            }
          }
        }

      go(Chunk.Queue.empty(), asPull).stream()
    }

  /**
   * Drops elements from the head of this stream until the supplied predicate returns false.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream.range(0..10)
   *     .dropWhile { it != 4 }
   *     .toList()
   *     .let(::println) // [4, 5, 6, 7, 8, 9, 10]
   * //sampleEnd
   * ```
   */
  fun dropWhile(p: (O) -> Boolean): Stream<O> =
    asPull.dropWhile(p).stream()

  /**
   * Like [dropWhile], but drops the first value which tests false.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream.range(0..10)
   *     .dropThrough { it != 4 }
   *     .toList()
   *     .let(::println) // [5, 6, 7, 8, 9, 10]
   * //sampleEnd
   * ```
   */
  fun dropThrough(p: (O) -> Boolean): Stream<O> =
    asPull.dropThrough(p).stream()

  /**
   * Emits the first [n] elements of this stream.
   * Alias for [take].
   */
  fun take(n: Int): Stream<O> =
    asPull.take(n.toLong()).void().stream()

  /**
   * Emits the first [n] elements of this stream.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream.range(0..1000)
   *     .take(5)
   *     .toList()
   *     .let(::println) // [0, 1, 2, 3, 4]
   * //sampleEnd
   * ```
   */
  fun take(n: Long): Stream<O> =
    asPull.take(n).void().stream()

  /**
   * Emits the last `n` elements of the input.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream.empty<Int>()
   *     .takeLastOrNull(5)
   *     .toList()
   *     .let(::println) // [null]
   * //sampleEnd
   * ```
   */
  fun takeLastOrNull(n: Int): Stream<O?> =
    asPull
      .takeLast(n)
      .flatMap { chunkQueue ->
        if (chunkQueue.isEmpty()) Pull.output1(null)
        else chunkQueue.chunks.fold(Pull.done<O>()) { acc, c ->
          acc.flatMap { Pull.output(c) }
        }
      }.stream()

  /**
   * Emits the last `n` elements of the input.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream.range(0..1000)
   *     .takeLast(5)
   *     .toList()
   *     .let(::println) // [996, 997, 998, 999, 1000]
   * //sampleEnd
   * ```
   */
  fun takeLast(n: Int): Stream<O> =
    asPull
      .takeLast(n)
      .flatMap { chunkQueue ->
        chunkQueue.chunks.fold(Pull.done<O>()) { acc, c ->
          acc.flatMap { Pull.output(c) }
        }
      }.stream()

  /**
   * Emits the first element of the stream for which the predicate is defined, or is empty.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream(null, 1, 2, null, 3)
   *     .first { it != null }
   *     .toList()
   *     .let(::println) // [1]
   * //sampleEnd
   * ```
   */
  fun first(f: (O) -> Boolean): Stream<O> =
    asPull
      .firstOrNull(f)
      .flatMap { uncons ->
        uncons?.head?.takeIf(f)?.let(Pull.Companion::output1) ?: Pull.done
      }.stream()

  /** alias for first **/
  fun find(f: (O) -> Boolean): Stream<O> =
    first(f)

  /**
   * Emits `true` as soon as a matching element is received, else `false` if no input matches.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit {
   *   Stream.range(0..10).exists { it == 4 }
   *     .toList().let(::println) //[true]
   *
   *   Stream.range(0..10).exists { it == 11 }
   *     .toList().let(::println) //[false]
   * }
   * //sampleEnd
   * ```
   *
   * @return Either a singleton stream, or a `never` stream.
   *  - If `this` is a finite stream, the result is a singleton stream, with after yielding one single value.
   *    If `this` is empty, that value is the `mempty` of the instance of `Monoid`.
   *  - If `this` is a non-terminating stream, and no matter if it yields any value, then the result is
   *    equivalent to the `Stream.never`: it never terminates nor yields any value.
   */
  fun exists(f: (O) -> Boolean): Stream<Boolean> =
    asPull
      .firstOrNull(f)
      .flatMap { uncons ->
        uncons?.head?.takeIf(f)?.let { Pull.output1(true) } ?: Pull.output1(false)
      }.stream()

  /**
   * Emits the longest prefix of the input for which all elements test true according to [f].
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream.range(0..1000)
   *     .takeWhile { it != 5 }
   *     .toList()
   *     .let(::println) // [0, 1, 2, 3, 4]
   * //sampleEnd
   * ```
   */
  fun takeWhile(p: (O) -> Boolean): Stream<O> =
    asPull.takeWhile(p).void().stream()

  /**
   * Like [takeWhile], but emits the first value which tests false.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream.range(0..1000)
   *     .takeThrough { it != 5 }
   *     .toList()
   *     .let(::println) // [0, 1, 2, 3, 4, 5]
   * //sampleEnd
   * ```
   */
  fun takeThrough(p: (O) -> Boolean): Stream<O> =
    asPull.takeThrough(p).void().stream()

  /**
   * Repeat this stream an infinite number of times.
   * `s.repeat() == s.append { s.append { s.append { ... } }`
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream(1,2,3)
   *     .repeat()
   *     .take(8)
   *     .toList()
   *     .let(::println) // [1, 2, 3, 1, 2, 3, 1, 2]
   * //sampleEnd
   * ```
   */
  fun repeat(): Stream<O> =
    this.append { repeat() }

  /**
   * Repeat this stream a given number of times.
   *
   * `s.repeatN(n) == s ++ s ++ s ++ ... (n times)`
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream(1,2,3)
   *     .repeatN(3)
   *     .toList()
   *     .let(::println) // [1, 2, 3, 1, 2, 3, 1, 2, 3]
   * //sampleEnd
   * ```
   */
  fun repeatN(n: Long): Stream<O> =
    if (n >= 0) this
    else this.append { repeatN(n - 1) }

  /**
   * Alias for `flatMap { o -> Stream.effect { f(o) } }`.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream(1, 2, 3)
   *     .effectMap { print(it) }
   *     .drain() // 123
   * //sampleEnd
   * ```
   */
  fun <B> effectMap(f: suspend (O) -> B): Stream<B> =
    flatMap { o -> effect { f(o) } }

  /**
   * Alias for `effectMap { o -> f(o); o }`.
   * Useful if you want to attach log function
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream(1, 2, 3)
   *     .effectTap { print(it) } // 123
   *     .toList()
   *     .let(::println) // [1, 2, 3]
   * //sampleEnd
   * ```
   */
  fun effectTap(f: suspend (O) -> Unit): Stream<O> =
    flatMap { o -> effect { f(o); o } }

  /**
   * Like [mapAccumulate], but accepts a function returning an suspend fun.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.*
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream(1,2,3,4)
   *     .effectMapAccumulate(0) { acc, i ->
   *       sleep((i * 10).milliseconds)
   *       Pair(i, acc + i)
   *     }
   *     .toList()
   *     .let(::println) // [(1,1), (2,3), (3,5), (4,7)]
   * //sampleEnd
   * ```
   */
  fun <S, O2> effectMapAccumulate(s: S, f: suspend (S, O) -> Pair<S, O2>): Stream<Pair<S, O2>> {
    fun go(s: S, pull: Pull<O, Unit>): Pull<Pair<S, O2>, Unit> =
      pull.uncons1OrNull().flatMap { uncons1 ->
        when (uncons1) {
          null -> Pull.done
          else -> {
            Pull.effect { f(s, uncons1.head) }.flatMap { (newS, o) ->
              Pull.output1(Pair(newS, o)).flatMap { go(newS, uncons1.tail) }
            }
          }
        }
      }

    return go(s, asPull).stream()
  }

  /**
   * Applies the specified pure function to each input and emits the result.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream("Hello", "World!")
   *     .map(String::length)
   *     .toList().let(::println) // [5, 6]
   * //sampleEnd
   * ```
   */
  fun <B> map(f: (O) -> B): Stream<B> =
    Stream(asPull.mapOutput(f))

  /**
   * Applies the specified pure function to each chunk in this stream.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream(1, 2, 3).append { Stream(4, 5, 6) }
   *     .mapChunks { ch -> ch.map { it + 1 } }
   *     .toList().let(::println) // [2, 3, 4, 5, 6, 7]
   * //sampleEnd
   * ```
   */
  fun <B> mapChunks(f: (Chunk<O>) -> Chunk<B>): Stream<B> =
    asPull.repeat { p ->
      p.unconsOrNull().flatMap { uncons ->
        when (uncons) {
          null -> Pull.just(null)
          else -> Pull.output(f(uncons.head)).map { uncons.tail }
        }
      }
    }.stream()

  /**
   * Maps a running total according to `S` and the input with the function `f`.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream("Hello", "World")
   *   .mapAccumulate(0) { l, s -> Pair(l + s.length, s.first()) }
   *   .toList()
   *   .let(::println) //[(5,H), (10,W)]
   * //sampleEnd
   * ```
   */
  fun <S, O2> mapAccumulate(init: S, f: (S, O) -> Pair<S, O2>): Stream<Pair<S, O2>> =
    scanChunks(init) { acc, c ->
      c.mapAccumulate(acc, { s: S, o: O ->
        val (newS, newO) = f(s, o)
        Pair(newS, Pair(newS, newO))
      })
    }

  /**
   * Converts the input to a stream of 1-element chunks.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream(1,2,3).append { Stream(4,5,6) }
   *   .unchunk()
   *   .chunks()
   *   .toList()
   *   .let(::println) //[Chunk(1), Chunk(2), Chunk(3), Chunk(4), Chunk(5), Chunk(6)]
   * //sampleEnd
   * ```
   */
  fun unchunk(): Stream<O> =
    asPull.repeat { pull ->
      pull.uncons1OrNull().flatMap { uncons1 ->
        when (uncons1) {
          null -> Pull.just(null)
          else -> Pull.output1(uncons1.head).map { uncons1.tail }
        }
      }
    }.stream()

  /**
   * Outputs all chunks from the source stream.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream(1).append { Stream(2, 3).append { Stream(4, 5, 6) } }
   *   .chunks()
   *   .toList()
   *   .let(::println) //[Chunk(1), Chunk(2, 3), Chunk(4, 5, 6)]
   * //sampleEnd
   * ```
   */
  fun chunks(): Stream<Chunk<O>> =
    asPull.repeat { pull ->
      pull.unconsOrNull().flatMap { uncons ->
        when (uncons) {
          null -> Pull.just(null)
          else -> Pull.output1(uncons.head).map { uncons.tail }
        }
      }
    }.stream()

  /**
   * Outputs chunk with a limited maximum size, splitting as necessary.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream(1).append { Stream(2, 3).append { Stream(4, 5, 6) } }
   *   .chunkLimit(2)
   *   .toList()
   *   .let(::println) //[Chunk(1), Chunk(2, 3), Chunk(4, 5), Chunk(6)]
   * //sampleEnd
   * ```
   */
  fun chunkLimit(n: Int): Stream<Chunk<O>> =
    asPull.repeat { pull ->
      pull.unconsLimitOrNull(n).flatMap { uncons ->
        when (uncons) {
          null -> Pull.just(null)
          else -> Pull.output1(uncons.head).map { uncons.tail }
        }
      }
    }.stream()

  /**
   * Outputs chunks of size larger than N
   *
   * Chunks from the source stream are split as necessary.
   *
   * If `allowFewerTotal` is true,
   * if the stream is smaller than N, should the elements be included
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream(1, 2).append { Stream(3, 4).append { Stream(5, 6, 7) } }
   *   .chunkMin(3)
   *   .toList()
   *   .let(::println) //[Chunk(1, 2, 3, 4), Chunk(5, 6, 7)]
   * //sampleEnd
   * ```
   */
  fun chunkMin(n: Int, allowFewerTotal: Boolean = true): Stream<Chunk<O>> {
    fun <A> go(nextChunk: Chunk.Queue<A>, s: Pull<A, Unit>): Pull<Chunk<A>, Unit> =
      s.unconsOrNull().flatMap { uncons ->
        when (uncons) {
          null ->
            if (allowFewerTotal && nextChunk.size > 0) Pull.output1(nextChunk.toChunk())
            else Pull.done
          else -> {
            val next = nextChunk.enqueue(uncons.head)
            if (next.size >= n) Pull.output1(next.toChunk()).flatMap { go(Chunk.Queue.empty(), uncons.tail) }
            else go(next, uncons.tail)
          }
        }
      }

    return asPull.unconsOrNull().flatMap { uncons ->
      when (uncons) {
        null -> Pull.done
        else -> {
          if (uncons.head.size() >= n) Pull.output1(uncons.head).flatMap { go(Chunk.Queue.empty(), uncons.tail) }
          else go(Chunk.Queue(uncons.head), uncons.tail)
        }
      }
    }.stream()
  }

  /**
   * Outputs chunks of size `n`.
   *
   * Chunks from the source stream are split as necessary.
   * If `allowFewer` is true, the last chunk that is emitted may have less than `n` elements, like shown below.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream(1,2,3)
   *   .repeat()
   *   .chunkN(2)
   *   .take(6)
   *   .toList()
   *   .let(::println) //[Chunk(1, 2), Chunk(3, 1), Chunk(2, 3), Chunk(1, 2), Chunk(3, 1), Chunk(2)]
   * //sampleEnd
   * ```
   */
  fun chunkN(n: Int, allowFewer: Boolean = true) =
    asPull.repeat { pull: Pull<O, Unit> ->
      pull.unconsN(n, allowFewer).flatMap { uncons ->
        when (uncons) {
          null -> Pull.just(null)
          else -> Pull.output1(uncons.head).map { uncons.tail }
        }
      }
    }.stream()

  /**
   * Emits only inputs which match the supplied predicate.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream.range(0..10)
   *     .filter { it % 2 == 0 }
   *     .toList()
   *     .let(::println) //[0, 2, 4, 6, 8]
   * //sampleEnd
   * ```
   */
  fun filter(p: (O) -> Boolean): Stream<O> =
    mapChunks { it.filter(p) }

  fun <B> mapNotNull(p: (O) -> B?): Stream<B> =
    mapChunks { it.mapNotNull(p) }

  fun <O2> mapFilter(p: (O) -> O2?): Stream<O2> =
    mapNotNull(p)

  /**
   * Like `filter`, but the predicate `f` depends on the previously emitted and
   * current elements.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream(1, -1, 2, -2, 3, -3, 4, -4)
   *   .filterWithPrevious { previous, current -> previous < current }
   *   .toList()
   *   .let(::println) //[1, 2, 3, 4]
   * //sampleEnd
   * ```
   */
  fun filterWithPrevious(f: (O, O) -> Boolean): Stream<O> {
    fun go(last: O, s: Pull<O, Unit>): Pull<O, Unit> =
      s.unconsOrNull().flatMap { uncons ->
        when (uncons) {
          null -> Pull.done
          else -> {
            // Check if we can emit this chunk unmodified
            val (allPass, newLast) = uncons.head.foldLeft(Pair(true, last)) { (acc, last), o ->
              Pair(acc && f(last, o), o)
            }
            if (allPass) {
              Pull.output(uncons.head).flatMap { go(newLast, uncons.tail) }
            } else {
              val (acc, newLast) = uncons.head.foldLeft(Pair(emptyList<O>(), last)) { (acc, last), o ->
                if (f(last, o)) Pair(acc + o, o)
                else Pair(acc, last)
              }
              Pull.output(Chunk.iterable(acc)).flatMap { go(newLast, uncons.tail) }
            }
          }
        }
      }

    return asPull.uncons1OrNull().flatMap { uncons ->
      when (uncons) {
        null -> Pull.done
        else -> Pull.output1(uncons.head).flatMap { go(uncons.head, uncons.tail) }
      }
    }.stream()
  }

  /**
   * Folds all inputs using an initial value `z` and supplied binary operator,
   * and emits a single element stream.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream(1, 2, 3, 4, 5)
   *   .fold(0) { acc, b -> acc + b }
   *   .toList()
   *   .let(::println) //[15]
   * //sampleEnd
   * ```
   */
  fun <B> fold(initial: B, f: (B, O) -> B): Stream<B> =
    asPull.fold(initial, f)
      .flatMap(Pull.Companion::output1)
      .stream()

  /**
   * Alias for `map(f).foldMonoid`.
   *
   * ```kotlin:ank:playground
   * import arrow.core.extensions.monoid
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream(1, 1, 1, 1, 1)
   *   .foldMap(Int.monoid()) { it + 1 }
   *   .toList()
   *   .let(::println) //[10]
   * //sampleEnd
   * ```
   */
  fun <O2> foldMap(MO2: Monoid<O2>, f: (O) -> O2): Stream<O2> =
    MO2.run {
      fold(empty()) { acc, o -> acc.combine(f(o)) }
    }

  /**
   * Like [scan], but accepts a suspending function.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream(1,2,3,4)
   *   .effectScan(0) { acc,i -> acc + i }
   *   .toList()
   *   .let(::println) //[0, 1, 3, 6, 10]
   * //sampleEnd
   * ```
   */
  fun <O2> effectScan(z: O2, f: suspend (O2, O) -> O2): Stream<O2> {
    fun go(z: O2, s: Pull<O, Unit>): Pull<O2, Unit> =
      s.uncons1OrNull().flatMap { uncons1 ->
        when (uncons1) {
          null -> Pull.done
          else -> Pull.effect { f(z, uncons1.head) }.flatMap { newO2 ->
            Pull.output1(newO2).flatMap { go(newO2, uncons1.tail) }
          }
        }
      }

    return asPull.uncons1OrNull().flatMap { uncons1 ->
      when (uncons1) {
        null -> Pull.output1(z)
        else -> Pull.effect { f(z, uncons1.head) }.flatMap { newO2 ->
          Pull.output(Chunk(z, newO2)).flatMap { go(newO2, uncons1.tail) }
        }
      }
    }.stream()
  }

  /**
   * Deterministically zips elements, terminating when the end of either branch is reached naturally.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream(1, 2, 3).zip(Stream(4, 5, 6, 7))
   *   .toList()
   *   .let(::println) //[(1,4), (2,5), (3,6)]
   * //sampleEnd
   * ```
   */
  fun <B> zip(that: Stream<B>): Stream<Pair<O, B>> =
    zipWith(that, ::Pair)

  /**
   * Like `zip`, but selects the right values only.
   * Useful with timed streams, the example below will emit a number every 100 milliseconds.
   */
  fun <B> zipRight(that: Stream<B>): Stream<B> =
    zipWith(that) { _, y -> y }

  /**
   * Like `zip`, but selects the left values only.
   * Useful with timed streams, the example below will emit a number every 100 milliseconds.
   */
  fun <B> zipLeft(that: Stream<B>): Stream<O> =
    zipWith(that) { x, _ -> x }

  /**
   * Deterministically zips elements using the specified function,
   * terminating when the end of either branch is reached naturally.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream(1, 2, 3)
   *   .zipWith(Stream(4, 5, 6, 7)) { acc, b -> acc + b }
   *   .toList()
   *   .let(::println) //[5, 7, 9]
   * //sampleEnd
   * ```
   */
  fun <B, C> zipWith(other: Stream<B>, f: (O, B) -> C): Stream<C> =
    asPull.zipWith_(
      other.asPull,
      { Pull.done },
      { Pull.done },
      f
    ).stream()

  /**
   * Zips the elements of the input stream with its indices, and returns the new stream.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream("The", "quick", "brown", "fox")
   *   .zipWithIndex()
   *   .toList()
   *   .let(::println) //[(The,0), (quick,1), (brown,2), (fox,3)]
   * //sampleEnd
   * ```
   */
  fun zipWithIndex(): Stream<Pair<O, Long>> =
    this.scanChunks(0L) { index, c ->
      var idx = index
      val out = c.map { o ->
        val r = Pair(o, idx)
        idx += 1
        r
      }

      Pair(idx, out)
    }

  /**
   * Zips each element of this stream with the next element wrapped into `Some`.
   * The last element is zipped with `None`.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream("The", "quick", "brown", "fox")
   *   .zipWithNext()
   *   .toList()
   *   .let(::println) //[(The,quick), (quick,brown), (brown,fox), (fox,null)]
   * //sampleEnd
   * ```
   */
  fun zipWithNext(): Stream<Pair<O, O?>> {
    fun go(last: O, s: Pull<O, Unit>): Pull<Pair<O, O?>, Unit> =
      s.unconsOrNull().flatMap { uncons ->
        when (uncons) {
          null -> Pull.output1(Pair(last, null))
          else -> {
            val (newLast, out) = uncons.head.mapAccumulate(last) { prev, next ->
              Pair(next, Pair(prev, next))
            }
            Pull.output(out).flatMap { go(newLast, uncons.tail) }
          }
        }
      }

    return asPull.uncons1OrNull().flatMap { uncons1 ->
      when (uncons1) {
        null -> Pull.done
        else -> go(uncons1.head, uncons1.tail)
      }
    }.stream()
  }

  /**
   * Zips each element of this stream with the previous element wrapped into `Some`.
   * The first element is zipped with `None`.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream("The", "quick", "brown", "fox")
   *   .zipWithPrevious()
   *   .toList()
   *   .let(::println) //[(null,The), (The,quick), (quick,brown), (brown,fox)]
   * //sampleEnd
   * ```
   */
  fun zipWithPrevious(): Stream<Pair<O?, O>> =
    mapAccumulate<O?, Pair<O?, O>>(null) { prev, next ->
      Pair(next, Pair(prev, next))
    }.map { it.second }

  /**
   * Zips each element of this stream with its previous and next element wrapped into `Some`.
   * The first element is zipped with `None` as the previous element,
   * the last element is zipped with `None` as the next element.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream("The", "quick", "brown", "fox")
   *   .zipWithPreviousAndNext()
   *   .toList()
   *   .let(::println) //[(null,The, quick), (The, quick,brown), (quick,brown,fox), (brown,fox,null)]
   * //sampleEnd
   * ```
   */
  fun zipWithPreviousAndNext(): Stream<Triple<O?, O, O?>> =
    zipWithPrevious().zipWithNext().map {
      Triple(it.first.first, it.first.second, it.second?.second)
    }

  /**
   * Zips the input with a running total according to `S`, up to but not including the current element. Thus the initial
   * `z` value is the first emitted to the output:
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream("uno", "dos", "tres", "cuatro")
   *   .zipWithScan(0) { acc, b -> acc + b.length }
   *   .toList()
   *   .let(::println) //[(uno,0), (dos,3), (tres,6), (cuatro,10)]
   * //sampleEnd
   * ```
   *
   * @see [zipWithScan1]
   */
  fun <B> zipWithScan(z: B, f: (B, O) -> B): Stream<Pair<O, B>> =
    mapAccumulate(z) { s, o ->
      val s2 = f(s, o)
      Pair(s2, Pair(o, s))
    }.map { it.second }

  /**
   * Zips the input with a running total according to `S`, including the current element. Thus the initial
   * `z` value is the first emitted to the output:
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream("uno", "dos", "tres", "cuatro")
   *   .zipWithScan1(0) { acc, b -> acc + b.length }
   *   .toList()
   *   .let(::println) //[(uno,3), (dos,6), (tres,10), (cuatro,16)]
   * //sampleEnd
   * ```
   *
   * @see [zipWithScan]
   */
  fun <B> zipWithScan1(z: B, f: (B, O) -> B): Stream<Pair<O, B>> =
    mapAccumulate(z) { s, o ->
      val s2 = f(s, o)
      Pair(s2, Pair(o, s2))
    }.map { it.second }

  /**
   * Emits `false` and halts as soon as a non-matching element is received; or
   * emits a single `true` value if it reaches the stream end and every input before that matches the predicate;
   * or hangs without emitting values if the input is infinite and all inputs match the predicate.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream(1, 2, 3, 4, 5)
   *   .forall { it < 10 }
   *   .toList()
   *   .let(::println) //[true]
   * //sampleEnd
   * ```
   *
   * @return Either a singleton or a never stream:
   * - '''If''' `this` yields an element `x` for which `¬ p(x)`, '''then'''
   *   a singleton stream with the value `false`. Pulling from the resultg
   *   performs all the effects needed until reaching the counterexample `x`.
   * - If `this` is a finite stream with no counterexamples of `p`, '''then''' a singleton stream with the `true` value.
   *   Pulling from the it will perform all effects of `this`.
   * - If `this` is an infinite stream and all its the elements satisfy  `p`, then the result
   *   is a `never` stream. Pulling from that stream will pull all effects from `this`.
   */
  fun forall(p: (O) -> Boolean): Stream<Boolean> =
    asPull.forall(p).flatMap(Pull.Companion::output1).stream()

  /**
   * Removes all output values from this stream.
   *
   * Often used with `merge` to run one side of the merge for its effect
   * while getting outputs from the opposite side of the merge.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream(1, 2, 3, 4)
   *   .void()
   *   .toList().let(::println) //[]
   * //sampleEnd
   * ```
   */
  fun void(): Stream<Nothing> =
    mapChunks { Chunk.empty<Nothing>() }

  /**
   * Behaves like the identity function but halts the stream on an error and does not return the error.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit =
   *   Stream(1,2,3).append { Stream.raiseError<Int>(RuntimeException()).append { Stream(4, 5, 6) } }
   *     .mask()
   *     .toList().let(::println) //[1, 2, 3]
   * //sampleEnd
   * ```
   */
  fun mask(): Stream<O> =
    handleErrorWith { empty() }

  /**
   * Returns a stream that when run, sleeps for duration `d` and then pulls from this stream.
   *
   * Alias for `sleep(d) ++ this`.
   */
  fun delayBy(d: Duration): Stream<O> =
    sleep_(d).append { this }

  /**
   * Runs the supplied stream in the background as elements from this stream are pulled.
   *
   * The resulting stream terminates upon termination of this stream. The background stream will
   * be interrupted at that point. Early termination of [other] does not terminate the resulting stream.
   *
   * Any errors that occur in either `this` or [other] stream result in the overall stream terminating
   * with an error.
   *
   * Upon finalization, the resulting stream will interrupt the background stream and wait for it to be
   * finalized.
   *
   * ```kotlin:ank:playground
   * import arrow.fx.coroutines.stream.concurrent.SignallingAtomic
   * import arrow.fx.coroutines.stream.*
   *
   * //sampleStart
   * suspend fun main(): Unit {
   *   val data = Stream.range(1..10)
   *   val signalling = SignallingAtomic(0)
   *
   *   signalling.discrete()
   *     .concurrently(data.effectMap { signalling.set(it) })
   *     .takeWhile { it < 9 }
   *     .toList()
   *     .let(::println) //[0, 1, 2, 3, 4, 5, 6, 7, 8]
   * }
   * //sampleEnd
   * ```
   */
  fun <O2> concurrently(other: Stream<O2>, ctx: CoroutineContext = ComputationPool): Stream<O> =
    effect { Pair(Promise<Unit>(), Promise<Either<Throwable, Unit>>()) }
      .flatMap { (interrupt, doneR) ->
        bracket(
          { ForkAndForget(ctx) { concurrentlyRunR(other, interrupt, doneR) } },
          {
            interrupt.complete(Unit)
            doneR.get().fold({ throw it }, ::identity)
          }
        ).flatMap { this.interruptWhen { Either.catch { interrupt.get() } } }
      }

  private suspend fun <O> concurrentlyRunR(
    other: Stream<O>,
    interrupt: Promise<Unit>,
    done: Promise<Either<Throwable, Unit>>
  ): Unit {
    val r = Either.catch {
      other
        .interruptWhen { Either.catch { interrupt.get() } }
        .drain()
    }

    done.complete(r)

    when (r) { // interrupt only if this failed otherwise give change to `this` to finalize
      is Left -> interrupt.complete(Unit)
      else -> Unit
    }
  }

  /**
   * 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 <B> either(ctx: CoroutineContext = ComputationPool, other: Stream<B>): Stream<Either<O, B>> =
    Stream(this.map { Left(it) }, other.map { Right(it) })
      .parJoin(2, ctx)

  /**
   * Starts this stream and cancels it as finalization of the returned stream.
   */
  fun spawn(ctx: CoroutineContext = ComputationPool): Stream<Fiber<Unit>> =
    supervise(ctx) { drain() }

  /**
   * 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() }).flatMap { this }

  /**
   * Like [onFinalize] but provides the reason for finalization as an `ExitCase`.
   */
  fun onFinalizeCase(f: suspend (ExitCase) -> Unit): Stream<O> =
    bracketCase({ Unit }) { _, ec -> f(ec) }.flatMap { this }

  /**
   * Interrupts the stream, when `haltOnSignal` finishes its evaluation.
   */
  fun interruptWhen(haltOnSignal: suspend () -> Either<Throwable, Unit>): Stream<O> =
    getScope.flatMap { scope ->
      supervise {
        val e = haltOnSignal.invoke().ignoreCancellation()
        scope.interrupt(e)
      }.flatMap { this }
    }.interruptScope()

  /**
   * Creates a scope that may be interrupted by calling scope#interrupt.
   */
  fun interruptScope(): Stream<O> =
    Stream(asPull.interruptScope())

  /**
   * Introduces an explicit scope.
   *
   * Scopes are normally introduced automatically, when using `bracket` or similar
   * operations that acquire resources and run finalizers. Manual scope introduction
   * is useful when using [bracketWeak]/[bracketCaseWeak], where no scope
   * is introduced.
   */
  fun scope(): Stream<O> =
    Stream(asPull.scope())

  /** Alias for `interruptWhen(haltWhenTrue.discrete)`. */
  fun interruptWhen(haltWhenTrue: Signal<Boolean>): Stream<O> =
    interruptWhen(haltWhenTrue.discrete())

  /**
   * Let through the `s2` branch as long as the `s1` branch is `false`,
   * listening asynchronously for the left branch to become `true`.
   * This halts as soon as either branch halts.
   *
   * Consider using the overload that takes a `Signal`, `Deferred` or `F[Either[Throwable, Unit]]`.
   */
  fun interruptWhen(haltWhenTrue: Stream<Boolean>): Stream<O> =
    effect {
      Triple(Promise<Unit>(), Promise<Either<Throwable, Unit>>(), Promise<Unit>())
    }.flatMap { (interruptL, doneR, interruptR) ->
      bracket(
        // Launch `haltWhenTrue` in a separate `Fiber`.
        { ForkAndForget { runR(haltWhenTrue, interruptR, doneR, interruptL) } },
        {
          interruptR.complete(Unit) // Interrupt `haltWhenTrue`
          doneR.get().fold({ throw it }, ::identity) // Signal exception if occurred
        }
      ) // After `haltWhenTrue` `Fiber` started return
        .flatMap {
          this@Stream.interruptWhen { Right(interruptL.get()) }
        }
    }

  private suspend fun runR(
    haltWhenTrue: Stream<Boolean>,
    interruptR: Promise<Unit>,
    doneR: Promise<Either<Throwable, Unit>>,
    interruptL: Promise<Unit>
  ): Unit =
    guaranteeCase({
      haltWhenTrue
        .takeWhile(Boolean::not)
        .interruptWhen { Right(interruptR.get()) }
        .drain()
    }, { ex ->
      val r = when (ex) {
        is ExitCase.Failure -> Left(ex.failure)
        else -> Right(Unit)
      }
      doneR.complete(r) // complete result promise
      interruptL.complete(Unit) // interrupt right with `Unit`
    })

  /** Fails this stream with a [TimeoutException] if it does not complete within given `timeout`. */
  fun timeout(timeout: Duration): Stream<O> =
    interruptWhen {
      sleep(timeout)
      Left(TimeoutException("Timed out after $timeout"))
    }

  /**
   * Interrupts this stream after the specified duration has passed.
   */
  fun interruptAfter(duration: Duration): Stream<O> =
    interruptWhen { Right(delay(duration)) }

  /**
   * Transforms this stream using the given `Pipe`.
   */
  fun <B> through(pipe: Pipe<O, B>): Stream<B> =
    pipe(this)

  /** Transforms this stream and `s2` using the given `Pipe2`. */
  fun <O2, B> through(s2: Stream<O2>, pipe2: Pipe2<O, O2, B>): Stream<B> =
    pipe2(this, s2)

  companion object {

    /** Empty [Stream]. */
    fun <O> empty(): Stream<O> = EmptyStream

    val unit: Stream<Unit> = Stream(Unit)

    /** Creates a [Stream] that, when run, fails with the supplied exception [err]. **/
    fun <O> raiseError(err: Throwable): Stream<O> =
      Stream(Pull.raiseError(err))

    /** Lifts a [Chunk] into a [Stream] */
    fun <A> chunk(ch: Chunk<A>): Stream<A> =
      Stream(Pull.output(ch))

    /**
     * A single-element `Stream` that waits for the duration `d` before emitting unit.
     */
    fun sleep(d: Duration): Stream<Unit> =
      effect { delay(d) }

    /**
     * Alias for `sleep(d).void`. Often used in conjunction with [append] (i.e., `sleep_(..).append { s }`) as a more
     * performant version of `sleep(..).flatMap { s }`.
     */
    fun sleep_(d: Duration): Stream<Nothing> =
      sleep(d).void()

    /**
     * Creates a single element stream that gets its value by evaluating the supplied effect.
     * If the effect fails, the returned stream fails.
     *
     * ```kotlin:ank:playground
     * import arrow.core.Either
     * import arrow.fx.coroutines.*
     * import arrow.fx.coroutines.stream.*
     *
     * //sampleEnd
     * suspend fun main(): Unit {
     *   Stream.effect { 10 }
     *     .toList()
     *     .let(::println) // [10]
     *
     *   Either.catch {
     *     Stream.effect { throw RuntimeException() }
     *       .toList()
     *   }.let(::println) // Left(java.lang.RuntimeException)
     * }
     * //sampleStart
     * ```
     */
    fun <O> effect(fo: suspend () -> O): Stream<O> =
      Stream(Pull.effect(fo).flatMap(Pull.Companion::output1))

    /**
     * Creates a stream that evaluates the supplied `fa` for its effect, discarding the output value.
     * As a result, the returned stream emits no elements and hence has output type `INothing`.
     *
     * Alias for `effect(fa).drain`.
     *
     * ```kotlin:ank:playground
     * import arrow.fx.coroutines.stream.*
     *
     * //sampleEnd
     * suspend fun main(): Unit =
     *   Stream.effect_ { println("Ran") }
     *     .toList()
     *     .let(::println) // []
     * //sampleStart
     * ```
     */
    fun effect_(fa: suspend () -> Unit): Stream<Nothing> =
      Stream(Pull.effect(fa).void())

    /** like `effect` but resulting chunk is flatten efficiently **/
    fun <O> effectUnChunk(fo: suspend () -> Chunk<O>): Stream<O> =
      Stream(Pull.effect(fo).flatMap { c -> Pull.output(c) })

    /**
     * Creates an infinite pure stream that always returns the supplied value.
     *
     * Elements are emitted in finite chunks with `chunkSize` number of elements.
     *
     * ```kotlin:ank:playground
     * import arrow.fx.coroutines.stream.*
     *
     * //sampleEnd
     * suspend fun main(): Unit {
     *   Stream.constant(0)
     *     .take(5)
     *     .toList()
     *     .let(::println) // [0, 0, 0, 0, 0]
     * }
     * //sampleStart
     * ```
     */
    fun <O> constant(o: O, chunkSize: Int = 256): Stream<O> =
      chunk(Chunk(chunkSize) { o }).repeat()

    /**
     * Lifts an effect that generates a stream in to a stream. Alias for `effect(f).flatMap(_)`.
     *
     * ```kotlin:ank:playground
     * import arrow.fx.coroutines.stream.*
     *
     * //sampleEnd
     * suspend fun main(): Unit =
     *   Stream.force { Stream(1, 2, 3) }
     *     .toList()
     *     .let(::println) // [1, 2, 3]
     * //sampleStart
     * ```
     */
    fun <A> force(f: suspend () -> Stream<A>): Stream<A> =
      effect(f).flatMap { identity(it) }

    /**
     * Like `emits`, but works for any class that extends `Iterable`
     */
    fun <O> iterable(os: Iterable<O>): Stream<O> =
      Stream.chunk(Chunk.iterable(os))

    /**
     * An infinite `Stream` that repeatedly applies a given function to a start value.
     * [initial] is the first value emitted, followed by `f(initial)`, then `f(f(initial))`, and so on.
     *
     * ```kotlin:ank:playground
     * import arrow.fx.coroutines.stream.*
     *
     * //sampleEnd
     * suspend fun main(): Unit =
     *   Stream.iterateEffect(0) { it + 1 }
     *     .take(10)
     *     .toList()
     *     .let(::println) // [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
     * //sampleStart
     * ```
     */
    fun <A> iterate(initial: A, f: (A) -> A): Stream<A> =
      just(initial).append { iterate(f(initial), f) }

    /** Like [iterate], but takes an suspend function. */
    fun <A> iterateEffect(start: A, f: suspend (A) -> A): Stream<A> =
      just(start).append {
        effect { f(start) }
          .flatMap { a -> iterateEffect(a, f) }
      }

    /**
     * Creates a stream by successively applying [f] until a `null` is returned, emitting
     * each output [O] and using each output [S] as input to the next invocation of [f].
     *
     * ```kotlin:ank:playground
     * import arrow.fx.coroutines.stream.*
     *
     * //sampleEnd
     * suspend fun main(): Unit =
     *   Stream.unfold(0) { i -> if (i < 5) Pair(i, i + 1) else null }
     *     .toList()
     *     .let(::println) //[0, 1, 2, 3, 4]
     * //sampleStart
     * ```
     */
    fun <S, O> unfold(s: S, f: (S) -> Pair<O, S>?): Stream<O> {
      fun loop(s: S): Stream<O> = f(s).let { pair ->
        if (pair != null) {
          just(pair.first).append { loop(pair.second) }
        } else empty()
      }

      return defer { loop(s) }
    }

    /** Like [unfold], but takes an suspend function. */
    fun <S, O> unfoldEffect(s: S, f: suspend (S) -> Pair<O, S>?): Stream<O> {
      fun loop(s: S): Stream<O> =
        effect { f(s) }.flatMap { pair ->
          if (pair != null) {
            just(pair.first).append { loop(pair.second) }
          } else empty()
        }

      return defer { loop(s) }
    }

    /**
     * Like [unfold] but each invocation of `f` provides a chunk of output.
     *
     * ```kotlin:ank:playground
     * import arrow.fx.coroutines.stream.*
     *
     * //sampleStart
     * suspend fun main(): Unit =
     *   Stream.unfoldChunk(0) { i ->
     *     if (i < 5) Pair(Chunk(i) { i }, i + 1)
     *     else null
     *   }
     *     .toList()
     *     .let(::println) //[1, 2, 2, 3, 3, 3, 4, 4, 4, 4]
     * //sampleEnd
     * ```
     */
    fun <S, O> unfoldChunk(s: S, f: (S) -> Pair<Chunk<O>, S>?): Stream<O> {
      fun loop(s: S): Stream<O> = f(s).let { pair ->
        if (pair != null) {
          chunk(pair.first).append { loop(pair.second) }
        } else empty()
      }

      return defer { loop(s) }
    }

    /** Like [unfoldChunk], but takes an effectful function. */
    fun <S, O> unfoldChunkEffect(s: S, f: suspend (S) -> Pair<Chunk<O>, S>?): Stream<O> {
      fun loop(s: S): Stream<O> =
        effect { f(s) }.flatMap { next ->
          when (next) {
            null -> empty()
            else -> chunk(next.first).append { loop(next.second) }
          }
        }

      return defer { loop(s) }
    }

    /**
     * Creates a singleton stream that emits the supplied value.
     *
     * ```kotlin:ank:playground
     * import arrow.fx.coroutines.stream.*
     *
     * //sampleStart
     * suspend fun main(): Unit =
     *   Stream.just(1)
     *     .toList()
     *     .let(::println) //[0]
     * //sampleEnd
     * ```
     */
    fun <O> just(o: O): Stream<O> =
      Stream(Pull.output1(o))

    /**
     * Returns a stream that evaluates [s] each time the stream is used,
     * allowing use of a mutable value in stream computations.
     *
     * Note: it's generally easier to reason about such computations using effectful
     * values. That is, allocate the mutable value in an effect and then use
     * `Stream.effect(fa).flatMap { a -> ??? }`.
     */
    fun <O> defer(s: () -> Stream<O>): Stream<O> =
      Stream(Pull.defer { s.invoke().asPull })

    /**
     * Creates a stream that emits a resource allocated by an effect, ensuring the resource is
     * eventually released regardless of how the stream is used.
     *
     * A typical use case for bracket is working with files or network sockets. The resource effect
     * opens a file and returns a reference to it. One can then flatMap on the returned Stream to access
     *  the file, e.g to read bytes and transform them in to some stream of elements
     * (e.g., bytes, strings, lines, etc.).
     * The `release` action then closes the file once the result Stream terminates, even in case of interruption
     * or errors.
     *
     * @param acquire resource to acquire at start of stream
     * @param release function which returns an effect that releases the resource
     */
    fun <R> bracket(acquire: suspend () -> R, release: suspend (R) -> Unit): Stream<R> =
      bracketCase(acquire) { r, _ -> release(r) }

    internal fun <R> bracketWeak(acquire: suspend () -> R, release: suspend (R) -> Unit): Stream<R> =
      bracketCaseWeak(acquire) { r, _ -> release(r) }

    /**
     * Like [bracket] but the release action is passed an [ExitCase].
     *
     * [ExitCase.Cancelled] is passed to the release action in the event of either stream interruption or
     * overall compiled effect cancellation.
     */
    fun <R> bracketCase(acquire: suspend () -> R, release: suspend (R, ExitCase) -> Unit): Stream<R> =
      bracketCaseWeak(acquire, release).scope()

    internal fun <R> bracketCaseWeak(acquire: suspend () -> R, release: suspend (R, ExitCase) -> Unit): Stream<R> =
      Stream(Pull.Eval.Acquire(acquire, release).flatMap(Pull.Companion::output1))

    /**
     * Converts the supplied resource in to a singleton stream.
     * Gives the same guarantees as using [bracketCase].
     */
    fun <O> resource(r: Resource<O>): Stream<O> =
      resourceWeak(r).scope()

    /**
     * Like [resource] but does not introduce a scope, allowing finalization to occur after
     * subsequent appends or other scope-preserving transformations.
     *
     * Scopes can be manually introduced via [scope] if desired.
     */
    internal fun <O> resourceWeak(r: Resource<O>): Stream<O> =
      when (r) {
        is Resource.Allocate ->
          Stream.bracketCaseWeak(r.acquire) { a, e -> r.release(a, e) }
        is Resource.Bind<*, O> -> resourceWeak(r.source).flatMap { o ->
          resourceWeak((r.f as (Any?) -> Resource<O>).invoke(o))
        }
        is Resource.Defer -> effect(r.resource).flatMap { resourceWeak(it) }
      }

    /**
     * Starts the supplied task and cancels it as finalization of the returned stream.
     */
    fun <A> supervise(ctx: CoroutineContext = ComputationPool, fa: suspend () -> A): Stream<Fiber<A>> =
      bracket(acquire = { fa.forkAndForget(ctx) }, release = { it.cancel() })

    /**
     * Lazily produce the range `[start, stopExclusive)`. If you want to produce
     * the sequence in one chunk, instead of lazily, use `emits(start until stopExclusive)`.
     *
     * ```kotlin:ank:playground
     * import arrow.fx.coroutines.stream.*
     *
     * //sampleStart
     * suspend fun main(): Unit =
     *   Stream.range(0..20 step 2)
     *     .toList().let(::println) // [0, 2, 4, 6, 8, 10, 12, 14, 16, 18, 20]
     * //sampleEnd
     * ```
     */
    fun range(range: IntProgression): Stream<Int> =
      unfold(range.first) { i ->
        if (i in range) Pair(i, i + range.step)
        else null
      }

    /**
     * Lazily produce the range `[start, stopExclusive)`. If you want to produce
     * the sequence in one chunk, instead of lazily, use `emits(start until stopExclusive)`.
     *
     * ```kotlin:ank:playground
     * import arrow.fx.coroutines.stream.*
     *
     * //sampleStart
     * suspend fun main(): Unit =
     *   Stream.range(0L..15 step 3)
     *     .toList().let(::println) // [0, 3, 6, 9, 12, 15]
     * //sampleEnd
     * ```
     */
    fun range(range: LongProgression): Stream<Long> =
      unfold(range.first) { i ->
        if (i in range) Pair(i, i + range.step)
        else null
      }

    /**
     * Lazily produce the range `[start, stopExclusive)`. If you want to produce
     * the sequence in one chunk, instead of lazily, use `emits(start until stopExclusive)`.
     *
     * ```kotlin:ank:playground
     * import arrow.fx.coroutines.stream.*
     *
     * //sampleStart
     * suspend fun main(): Unit =
     *   Stream.range('a'..'z' step 2)
     *     .toList()
     *     .let(::println) // [a, c, e, g, i, k, m, o, q, s, u, w, y]
     * //sampleEnd
     * ```
     */
    fun range(range: CharProgression): Stream<Char> =
      unfold(range.first) { i ->
        if (i in range) Pair(i, i + range.step)
        else null
      }

    /**
     * Gets the current scope, allowing manual leasing or interruption.
     * This is a low-level method and generally should not be used by user code.
     */
    val getScope: Stream<Scope> =
      Stream(Pull.getScope.flatMap(Pull.Companion::output1))

    /**
     * Creates a [Stream] that `never` returns any values
     *
     * ```kotlin:ank:playground
     * import arrow.fx.coroutines.*
     * import arrow.fx.coroutines.stream.*
     *
     * //sampleStart
     * suspend fun main(): Unit =
     *   timeOutOrNull(1.seconds) {
     *     Stream.never<Int>().drain()
     *   }.let(::println)
     * //sampleEnd
     * ```
     */
    fun <A> never(): Stream<A> =
      effect { arrow.fx.coroutines.never<A>() }

    /**
     * Creates a pure stream that emits the supplied values.
     *
     * ```kotlin:ank:playground
     * import arrow.fx.coroutines.stream.*
     *
     * //sampleStart
     * suspend fun main(): Unit =
     *   Stream.emits(1, 2, 3)
     *     .toList()
     *     .let(::println) // [1, 2, 3]
     * //sampleEnd
     *```
     */
    fun <A> emits(vararg aas: A): Stream<A> =
      when (aas.size) {
        0 -> empty()
        1 -> just(aas[0])
        else -> Stream(Pull.output(Chunk.array(aas)))
      }

    /**
     * Creates a pure stream that emits the supplied values.
     * Alias for [emits].
     */
    operator fun <A> invoke(vararg aas: A): Stream<A> =
      when (aas.size) {
        0 -> empty()
        1 -> just(aas[0])
        else -> Stream(Pull.output(Chunk.array(aas)))
      }

    /**
     * Creates a random stream of integers using a random seed.
     */
    fun random(seed: Int): Stream<Int> =
      effect { Random(seed) }
        .flatMap { r -> effect { r.nextInt() }.repeat() }
  }

  override fun toString(): String =
    "$Stream($asPull)"
}

/**
 * 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.
 *
 * ```kotlin:ank:playground
 * import arrow.fx.coroutines.stream.*
 *
 * //sampleStart
 * suspend fun main(): Unit =
 *   Stream(1,2,3)
 *     .zipAll(Stream(4,5,6,7), 0,0)
 *     .toList().let(::println) // [(1,4), (2,5), (3,6), (0,7)]
 * //sampleEnd
 *```
 */
fun <O, B> Stream<O>.zipAll(that: Stream<B>, pad1: O, pad2: B): Stream<Pair<O, B>> =
  zipAllWith(that, pad1, pad2, ::Pair)

/**
 * 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.
 *
 * ```kotlin:ank:playground
 * import arrow.fx.coroutines.stream.*
 *
 * //sampleStart
 * suspend fun main(): Unit =
 *   Stream(1,2,3)
 *     .zipAllWith(Stream(4,5,6,7), 0, 0) { acc, b -> acc + b }
 *     .toList().let(::println) // [5, 7, 9, 7]
 * //sampleEnd
 *```
 */
fun <A, B, C> Stream<A>.zipAllWith(
  that: Stream<B>,
  pad1: A,
  pad2: B,
  f: (A, B) -> C
): Stream<C> {
  fun cont1(z: Either<Pair<Chunk<A>, Pull<A, Unit>>, Pull<A, Unit>>): Pull<C, Unit> {
    fun contLeft(s: Pull<A, Unit>): Pull<C, Unit> =
      s.unconsOrNull().flatMap { uncons ->
        when (uncons) {
          null -> Pull.done
          else ->
            Pull.output(uncons.head.map { o -> f(o, pad2) }).flatMap { contLeft(uncons.tail) }
        }
      }

    return when (z) {
      is Left ->
        Pull.output(z.a.first.map { o -> f(o, pad2) }).flatMap { contLeft(z.a.second) }
      is Right -> contLeft(z.b)
    }
  }

  fun cont2(z: Either<Pair<Chunk<B>, Pull<B, Unit>>, Pull<B, Unit>>): Pull<C, Unit> {
    fun contRight(s: Pull<B, Unit>): Pull<C, Unit> =
      s.unconsOrNull().flatMap { uncons ->
        when (uncons) {
          null -> Pull.done
          else ->
            Pull.output(uncons.head.map { o2 -> f(pad1, o2) }).flatMap { contRight(uncons.tail) }
        }
      }

    return when (z) {
      is Left -> Pull.output(z.a.first.map { o2 -> f(pad1, o2) }).flatMap { contRight(z.a.second) }
      is Right -> contRight(z.b)
    }
  }

  return asPull.zipWith_(that.asPull, ::cont1, ::cont2, f).stream()
}

/**
 * Run `s2` after `this`, regardless of errors during `this`, then reraise any errors encountered during `this`.
 *
 * Note: this should *not* be used for resource cleanup! Use `bracket` or `onFinalize` instead.
 *
 * ```kotlin:ank:playground
 * import arrow.fx.coroutines.stream.*
 *
 * //sampleStart
 * suspend fun main(): Unit =
 *   Stream(1, 2, 3).onComplete { Stream(4, 5) }
 *     .toList().let(::println) // [1, 2, 3, 4, 5]
 * //sampleEnd
 *```
 */
fun <O> Stream<O>.onComplete(s2: () -> Stream<O>): Stream<O> =
  handleErrorWith { e -> s2.invoke().append { Stream.raiseError(e) } }
    .append(s2)

/**
 * Repartitions the input with the function `f`. On each step `f` is applied
 * to the input and all elements but the last of the resulting sequence
 * are emitted. The last element is then appended to the next input using the
 * Semigroup `S`.
 *
 * ```kotlin:ank:playground
 * import arrow.fx.coroutines.stream.*
 * import arrow.core.extensions.semigroup
 *
 * //sampleStart
 * suspend fun main(): Unit =
 *   Stream("Hel", "l", "o Wor", "ld")
 *     .repartition(String.semigroup()) { s -> Chunk.iterable(s.split(" ")) }
 *     .toList()
 *     .let(::println) //[Hello, World]
 * //sampleEnd
 * ```
 */
fun <O> Stream<O>.repartition(S: Semigroup<O>, f: (O) -> Chunk<O>): Stream<O> =
  S.run {
    asPull.scanChunks<O, O?, O?>(null) { carry, chunk ->
      val (out, res) = chunk.scanLeftCarry(Pair(Chunk.empty<O>(), carry)) { (_, carry), o ->
        val newO = carry?.combine(o) ?: o
        val partitions = f(newO)

        when (val size = partitions.size()) {
          0 -> Pair(partitions, null)
          1 -> Pair(Chunk.empty(), partitions.lastOrNull())
          else -> Pair(partitions.take(size - 1), partitions.lastOrNull())
        }
      }

      Pair(res.second, out.flatMap { (o, _) -> o })
    }.flatMap {
      it?.let(Pull.Companion::output1) ?: Pull.done
    }.stream()
  }

/**
 * Filters any `null`.
 *
 * ```kotlin:ank:playground
 * import arrow.core.Some
 * import arrow.core.None
 * import arrow.fx.coroutines.stream.*
 *
 * //sampleStart
 * suspend fun main(): Unit =
 *   Stream(1, 2, null, 3, None)
 *     .filterNotNull()
 *     .toList()
 *     .let(::println) //[1, 2, 3]
 * //sampleEnd
 * ```
 */
fun <O : Any> Stream<O?>.filterNotNull(): Stream<O> =
  mapFilter { it }

/**
 * Halts the input stream at the first `null`.
 *
 * ```kotlin:ank:playground
 * import arrow.fx.coroutines.stream.*
 *
 * //sampleStart
 * suspend fun main(): Unit =
 *   Stream(1, 2, null, 3, null)
 *     .terminateOnNull()
 *     .toList()
 *     .let(::println) //[1, 2]
 * //sampleEnd
 * ```
 */
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 }
 *     .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(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 }
          }
        }
      }
    }
  }.stream()

/**
 * Filters any [arrow.core.None].
 *
 * ```kotlin:ank:playground
 * import arrow.core.Some
 * import arrow.core.None
 * import arrow.fx.coroutines.stream.*
 *
 * //sampleStart
 * suspend fun main(): Unit =
 *   Stream(Some(1), Some(2), None, Some(3), None)
 *     .filterOption()
 *     .toList()
 *     .let(::println) //[1, 2, 3]
 * //sampleEnd
 * ```
 */
fun <O> Stream<Option<O>>.filterOption(): Stream<O> =
  mapFilter { it.orNull() }

/**
 * Halts the input stream at the first [arrow.core.None].
 *
 * ```kotlin:ank:playground
 * import arrow.core.Some
 * import arrow.core.None
 * import arrow.fx.coroutines.stream.*
 *
 * //sampleStart
 * suspend fun main(): Unit =
 *   Stream(Some(1), Some(2), None, Some(3), None)
 *     .terminateOnNone()
 *     .toList()
 *     .let(::println) //[1, 2]
 * //sampleEnd
 * ```
 */
fun <O> Stream<Option<O>>.terminateOnNone(): Stream<O> =
  asPull.repeat { pull ->
    pull.unconsOrNull().flatMap { uncons ->
      when (uncons) {
        null -> Pull.just(null)
        else -> {
          when (val idx = uncons.head.indexOfFirst(Option<O>::isEmpty)) {
            0 -> Pull.just(null)
            null -> Pull.output<O>(uncons.head.map { it.orNull()!! }).map { uncons.tail }
            else -> Pull.output(uncons.head.take(idx).map { it.orNull()!! }).map { null }
          }
        }
      }
    }
  }.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) }

private val EmptyStream: Stream<Nothing> = Stream(Pull.done)

fun <O> emptyStream(): Stream<O> =
  Stream.empty()

/**
 * 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.*
 *
 * //sampleStart
 * suspend fun main(): Unit =
 *   Stream(1, 2, 3)
 *     .interleave(Stream(4, 5, 6, 7))
 *     .toList()
 *     .let(::println) //[1, 4, 2, 5, 3, 6]
 * //sampleEnd
 * ```
 */
fun <O> Stream<O>.interleave(that: Stream<O>): Stream<O> =
  zip(that).flatMap { (o1, o2) -> Stream(o1, o2) }

/**
 * 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.*
 *
 * //sampleStart
 * suspend fun main(): Unit =
 *   Stream(1, 2, 3)
 *     .interleaveAll(Stream(4, 5, 6, 7))
 *     .toList()
 *     .let(::println) //[1, 4, 2, 5, 3, 6, 7]
 * //sampleEnd
 * ```
 */
fun <O> Stream<O>.interleaveAll(that: Stream<O>): Stream<O> =
  map { Option(it) }
    .zipAll(that.map { Option(it) }, None, None)
    .flatMap { (opt1, opt2) ->
      Stream.chunk(Chunk.iterable(opt1.toList() + opt2.toList()))
    }

/**
 * Emits the specified separator between every pair of elements in the source stream.
 *
 * ```kotlin:ank:playground
 * import arrow.fx.coroutines.stream.*
 *
 * //sampleStart
 * suspend fun main(): Unit =
 *   Stream(1, 2, 3, 4, 5)
 *     .intersperse(0)
 *     .toList()
 *     .let(::println) //[1, 0, 2, 0, 3, 0, 4, 0, 5]
 * //sampleEnd
 * ```
 */
fun <O> Stream<O>.intersperse(separator: O): Stream<O> =
  asPull.echo1().flatMap { pull ->
    when (pull) {
      null -> Pull.done
      else -> pull.repeat { pull2 ->
        pull2.unconsOrNull().flatMap { uncons ->
          when (uncons) {
            null -> Pull.just(null)
            else -> Pull.output(uncons.head.intersperse(separator))
              .flatMap { Pull.just(uncons.tail) }
          }
        }
      }
    }
  }.stream()

/**
 * Lazily appends [s2] to the end of this stream.
 *
 * ```kotlin:ank:playground
 * import arrow.fx.coroutines.stream.*
 *
 * //sampleStart
 * suspend fun main(): Unit =
 *   Stream(1, 2, 3).append { Stream(4, 5, 6) }
 *     .toList().let(::println) // [1, 2, 3, 4, 5, 6]
 * //sampleEnd
 *```
 */
fun <O> Stream<O>.append(s2: () -> Stream<O>): Stream<O> =
  Stream(asPull.append { s2.invoke().asPull })

/**
 * Prepends a [Chunk] onto the front of this stream.
 *
 * ```kotlin:ank:playground
 * import arrow.fx.coroutines.stream.*
 *
 * //sampleStart
 * suspend fun main(): Unit =
 *   (Chunk(-1, 0) prependTo Stream(1, 2, 3))
 *   .toList().let(::println) // [-1, 0, 1, 2, 3]
 * //sampleEnd
 * ```
 */
infix fun <O> Chunk<O>.prependTo(s: Stream<O>): Stream<O> =
  s.cons(this)

/**
 * Prepends a [Chunk] onto the front of this stream.
 *
 * ```kotlin:ank:playground
 * import arrow.fx.coroutines.stream.*
 *
 * //sampleStart
 * suspend fun main(): Unit =
 *   Stream(1, 2, 3).cons(Chunk(-1, 0))
 *     .toList().let(::println) // [-1, 0, 1, 2, 3]
 * //sampleEnd
 * ```
 */
fun <O> Stream<O>.cons(c: Chunk<O>): Stream<O> =
  if (c.isEmpty()) this else Stream.chunk(c).append { this }

/**
 * Prepends a value onto the front of this stream.
 *
 * ```kotlin:ank:playground
 * import arrow.fx.coroutines.stream.*
 *
 * //sampleStart
 * suspend fun main(): Unit =
 *   (0 prependTo Stream(1, 2, 3))
 *   .toList().let(::println) // [0, 1, 2, 3]
 * //sampleEnd
 * ```
 */
infix fun <O> O.prependTo(s: Stream<O>): Stream<O> =
  Stream.just(this).append { s }

/**
 * Prepends a value onto the front of this stream.
 *
 * ```kotlin:ank:playground
 * import arrow.fx.coroutines.stream.*
 *
 * //sampleStart
 * suspend fun main(): Unit =
 *   Stream(1, 2, 3).cons1(0)
 *     .toList()
 *   .  let(::println) // [0, 1, 2, 3]
 * //sampleEnd
 * ```
 */
fun <O> Stream<O>.cons1(o: O): Stream<O> =
  Stream.just(o).append { this }

/**
 * If `this` [Stream] terminates with [Stream.raiseError]`, invoke [h] and continue with result.
 *
 * ```kotlin:ank:playground
 * import arrow.fx.coroutines.stream.*
 *
 * //sampleStart
 * suspend fun main(): Unit =
 *   Stream(1, 2, 3)
 *     .append { Stream.raiseError(RuntimeException()) }
 *     .handleErrorWith { _: Throwable -> Stream.just(0) }
 *     .toList()
 *     .let(::println) // [1, 2, 3, 0]
 * //sampleEnd
 * ```
 */
fun <O> Stream<O>.handleErrorWith(h: (Throwable) -> Stream<O>): Stream<O> =
  Stream(asPull.scope().handleErrorWith { e -> h(e).asPull })

/** Flattens a stream of streams in to a single stream by concatenating each stream. */
fun <O> Stream<Stream<O>>.flatten(): Stream<O> =
  Stream(asPull.flatMapOutput { o -> o.asPull })

/**
 * Folds all inputs using the supplied operator [f], and emits a single-element stream,
 * or the empty stream if the input is empty, or the never stream if the input is non-terminating.
 *
 * ```kotlin:ank:playground
 * import arrow.fx.coroutines.stream.*
 *
 * //sampleStart
 * suspend fun main(): Unit =
 *   Stream(1, 2, 3, 4, 5)
 *     .fold1 { a, b -> a + b }
 *     .toList()
 *     .let(::println) // [15]
 * //sampleEnd
 * ```
 */
fun <O> Stream<O>.fold1(f: (O, O) -> O): Stream<O> =
  asPull.fold1(f).flatMap { o ->
    o?.let(Pull.Companion::output1) ?: Pull.done
  }.stream()

fun <O> Stream<O>.reduce(f: (O, O) -> O): Stream<O> =
  fold1(f)

/**
 * Reduces this stream with the Semigroup for `O`.
 */
fun <O> Stream<O>.reduceSemigroup(S: Semigroup<O>): Stream<O> =
  S.run { reduce { o, o2 -> o.combine(o2) } }

/**
 * Folds this stream with the monoid for `O`.
 *
 * @return Either a singleton stream or a `never` stream:
 *  - If `this` is a finite stream, the result is a singleton stream.
 *    If `this` is empty, that value is the `mempty` of the instance of `Monoid`.
 *  - If `this` is a non-terminating stream, and no matter if it yields any value, then the result is
 *    equivalent to the `Stream.never`: it never terminates nor yields any value.
 *
 * ```kotlin:ank:playground
 * import arrow.core.extensions.monoid
 * import arrow.fx.coroutines.stream.*
 *
 * //sampleStart
 * suspend fun main(): Unit =
 *   Stream(1, 2, 3, 4, 5)
 *     .foldMonoid(Int.monoid())
 *     .toList()
 *     .let(::println) // [15]
 * //sampleEnd
 * ```
 */
fun <O> Stream<O>.foldMonoid(MO: Monoid<O>): Stream<O> =
  MO.run {
    fold(empty()) { acc, b -> acc.combine(b) }
  }

/**
 * Left fold which outputs all intermediate results.
 *
 * ```kotlin:ank:playground
 * import arrow.fx.coroutines.stream.*
 *
 * //sampleStart
 * suspend fun main(): Unit =
 *   Stream(1,2,3,4)
 *   .scan(0) { a, b -> a + b }
 *   .toList()
 *   .let(::println) // [0, 1, 3, 6, 10]
 * //sampleEnd
 * ```
 *
 * More generally:
 *   `Stream().scan(z, f) == Stream(z)`
 *   `Stream(x1).scan(z, f) == Stream(z, f(z,x1))`
 *   `Stream(x1,x2).scan(z, f) == Stream(z, f(z,x1), f(f(z,x1),x2))`
 *   etc
 */
fun <O, O2> Stream<O>.scan(init: O2, f: (O2, O) -> O2): Stream<O2> =
  Pull.output1(init).flatMap { asPull.scan_(init, f) }.stream()

/**
 * Folds this stream with the monoid for `O` while emitting all intermediate results.
 *
 * ```kotlin:ank:playground
 * import arrow.fx.coroutines.stream.*
 * import arrow.core.extensions.monoid
 *
 * //sampleStart
 * suspend fun main(): Unit =
 *   Stream(1, 2, 3, 4)
 *     .scanMonoid(Int.monoid())
 *     .toList()
 *     .let(::println) //[0, 1, 3, 6, 10]
 * //sampleEnd
 * ```
 */
fun <O> Stream<O>.scanMonoid(MO: Monoid<O>): Stream<O> =
  MO.run { scan(empty()) { a, b -> a.combine(b) } }

/**
 * Alias for `map(f).scanMonoid`.
 *
 * ```kotlin:ank:playground
 * import arrow.fx.coroutines.stream.*
 * import arrow.core.extensions.monoid
 *
 * //sampleStart
 * suspend fun main(): Unit =
 *   Stream("a", "aa", "aaa", "aaaa")
 *     .scanMap(Int.monoid()) { it.length }
 *     .toList()
 *     .let(::println) //[0, 1, 3, 6, 10]
 * //sampleEnd
 * ```
 */
fun <O, O2> Stream<O>.scanMap(MO2: Monoid<O2>, f: (O) -> O2): Stream<O2> =
  MO2.run { scan(empty()) { a, b -> a.combine(f(b)) } }

private fun <O, O2> Pull<O, Unit>.scan_(init: O2, f: (O2, O) -> O2): Pull<O2, Unit> =
  unconsOrNull().flatMap { uncons ->
    when (uncons) {
      null -> Pull.done
      else -> {
        val (out, carry) = uncons.head.scanLeftCarry(init, f)
        Pull.output(out).flatMap { uncons.tail.scan_(carry, f) }
      }
    }
  }

/**
 * Like `[scan]`, but uses the first element of the stream as the seed.
 *
 * ```kotlin:ank:playground
 * import arrow.fx.coroutines.stream.*
 *
 * //sampleStart
 * suspend fun main(): Unit =
 *   Stream(1, 2, 3, 4)
 *    .scan1 { a, b -> a + b }
 *    .toList()
 *    .let(::println) //[1, 3, 6, 10]
 * //sampleEnd
 * ```
 */
fun <O> Stream<O>.scan1(f: (O, O) -> O): Stream<O> =
  asPull.uncons1OrNull().flatMap { uncons1 ->
    when (uncons1) {
      null -> Pull.done
      else -> {
        Pull.output1(uncons1.head).flatMap { uncons1.tail.scan_(uncons1.head, f) }
      }
    }
  }.stream()

/**
 * Like `scan` but `f` is applied to each chunk of the source stream.
 * The resulting chunk is emitted and the result of the chunk is used in the
 * next invocation of `f`.
 *
 * Many stateful pipes can be implemented efficiently (i.e., supporting fusion) with this method.
 */
fun <O, S, O2> Stream<O>.scanChunks(
  init: S,
  f: (S, Chunk<O>) -> Pair<S, Chunk<O2>>
): Stream<O2> =
  scanChunksOpt(init) { s -> { c: Chunk<O> -> f(s, c) } }

/**
 * More general version of `scanChunks` where the current state (i.e., `S`) can be inspected
 * to determine if another chunk should be pulled or if the stream should terminate.
 * Termination is signaled by returning `null` from `f`. Otherwise, a function which consumes
 * the next chunk.
 *
 * ```kotlin:ank:playground
 * import arrow.fx.coroutines.stream.*
 *
 * //sampleStart
 * fun <O> Stream<O>.take(n: Int): Stream<O> =
 *   scanChunksOpt<O, Int, O>(n) { n ->
 *     if (n <= 0) null
 *     else { chunk: Chunk<O> ->
 *       if (chunk.size() < n) Pair(n - chunk.size(), chunk)
 *       else Pair(0, chunk.take(n))
 *     }
 *   }
 *
 * suspend fun main(): Unit =
 *   Stream.range(0..100).take(5)
 *     .toList().let(::println) // [0, 1, 2, 3, 4]
 * //sampleEnd
 * ```
 */
fun <O, S, O2> Stream<O>.scanChunksOpt(
  init: S,
  f: (S) -> ((Chunk<O>) -> Pair<S, Chunk<O2>>)?
): Stream<O2> =
  asPull.scanChunksOpt(init, f).void().stream()

private fun <A> getNextIterator(i: Iterator<A>): Pair<A, Iterator<A>>? =
  if (i.hasNext()) Pair(i.next(), i)
  else null

private fun <A, B, C> Pull<A, Unit>.zipWith_(
  other: Pull<B, Unit>,
  padA: ZipWithCont<A, C>,
  padB: ZipWithCont<B, C>,
  f: (A, B) -> C
): Pull<C, Unit> {
  fun go(leg1: StepLeg<A>, leg2: StepLeg<B>): Pull<C, Unit> {
    val l1h = leg1.head
    val l2h = leg2.head

    val out = l1h.zipWith(l2h, f)

    return Pull.output(out).flatMap {
      if (l1h.size() > l2h.size()) {
        val extra1 = l1h.drop(l2h.size())
        leg2.stepLeg().flatMap { step ->
          when (step) {
            null -> padA(Left(Pair(extra1, leg1.pull())))
            else -> go(leg1.setHead(extra1), step)
          }
        }
      } else {
        val extra2 = l2h.drop(l1h.size())
        leg1.stepLeg().flatMap { step ->
          when (step) {
            null -> padB(Left(Pair(extra2, leg2.pull())))
            else -> go(step, leg2.setHead(extra2))
          }
        }
      }
    }
  }

  return stepLeg().flatMap { step ->
    when (step) {
      null -> padB(Right(other))
      else -> other.stepLeg().flatMap { step2 ->
        when (step2) {
          null -> padA(Left(Pair(step.head, step.pull())))
          else -> go(step, step2)
        }
      }
    }
  }
}

typealias ZipWithCont<I, O> =
  (Either<Pair<Chunk<I>, Pull<I, Unit>>, Pull<I, Unit>>) -> Pull<O, Unit>

/** `Monoid` instance for `Stream`. */
fun <O> Stream.Companion.monoid(): Monoid<Stream<O>> =
  object : Monoid<Stream<O>> {
    override fun empty(): Stream<O> =
      Stream.empty()

    override fun Stream<O>.combine(b: Stream<O>): Stream<O> =
      this@combine.append { b }
  }