Understanding ZChannels (zio-streams 2.x)

pipeline.scala

object Experiment extends ZIOAppDefault {
  /**
  Collect elements and emit them in single chunks of 100
  */
  val aggregate: ZPipeline[Any, Nothing, Int, Long] = {
    def go[Err](acc: Long): ZChannel[Any, Err, Chunk[Int], Any, Err, Chunk[Long], Any] =
      ZChannel.readWith[Any, Err, Chunk[Int], Any, Err, Chunk[Long], Any](
        in = { inChunk =>
          val next = acc + inChunk.sum
          if (next > 100L) {
            val rem = next - 100L
            ZChannel.write(Chunk.single(100L)) *> go(rem)
          } else go(next)
        },
        error = err => ZChannel.fail(err),
        done = done => ZChannel.write(Chunk.single(acc)) *> ZChannel.succeed(done)
      )

    go(0L).toPipeline[Int, Long]
  }

  override def run: ZIO[ZIOAppArgs with Scope, Any, Any] =
    ZStream
      .iterate(0)(_ + 1)
      .take(50)
      .via(aggregate)
      .debug("aggregator>")
      .runCollect
      .debug("collect>")
}

stream.scala

object LearningChannel extends ZIOAppDefault {

  // Producer of elements
  val countUp: UStream[Int] = {
    def go(current: Int): ZChannel[Any, Any, Any, Any, Nothing, Chunk[Int], Unit] =
      ZChannel.write(Chunk.single(current)) *> go(current + 1)

    go(0).toStream
  }

  override def run: ZIO[ZIOAppArgs with Scope, Any, Any] =
    countUp
      .throttleShape(1, 100.millis)(_.size.toLong)
      .debug("writer>")
      .runDrain
}