Skip to main content

How to Run Streams Without Blocking

In this guide, we total the price of a coffee order by asking a price service for each item, without parking a thread while we wait. By the end, you will have a pipeline written once against Stream and Async that runs on the JVM and on Scala.js, and you will know where it is safe to block, how its errors and cancellation behave, and how to read a browser or Node ReadableStream such as a fetch response body as a stream.

We start from the blocking terminal that causes the trouble, swap it for its Async twin, add async work per element with runForeachAsync and runFoldAsync, and compose the result. Then we look at errors and cancellation, run the same pipeline on Scala.js, and read a ReadableStream.

The Problem​

A blocking terminal such as Stream#runCollect or Stream#runFold makes one promise: when it returns, the stream is finished and the answer is in your hand. On the JVM, the promise is kept by parking the calling thread until the stream completes. If the stream does a remote call per element, the thread sits idle for every call. On Scala.js the promise cannot be kept at all, because the single thread that would wait is the one that has to run the callbacks that deliver the answer.

For example, here is a price total over a list of items, where each price comes from a call that returns a Future. A caller that wants one number back has nothing to do but wait for every price in turn:

import scala.concurrent.{Await, Future}
import scala.concurrent.duration._

def totalBlocking(items: List[String], price: String => Future[Long]): Long =
  items.foldLeft(0L)((sum, item) => sum + Await.result(price(item), 5.seconds))

This parks the calling thread once per item, one item at a time, so a hundred 100 ms lookups hold a thread for ten seconds doing nothing. A web server that does this per request runs out of threads long before it runs out of CPU. A blocking stream terminal such as Stream#runCollect or Stream#runFold has the same cost, and on Scala.js it does not even compile, because those terminals do not exist there.

In this guide, we replace each blocking call with an asynchronous one that returns immediately with an Async and let the caller decide when, and whether, to wait.

Prerequisites​

To add the streams module to your project, include one of these dependencies in your build.sbt. The async module comes with it:

// Scala 2.13 and Scala 3.3 to 3.8
libraryDependencies += "dev.zio" %% "zio-blocks-streams" % "0.0.56"

// Scala 3.9 and newer (also needs scalacOptions += "-experimental"): the same module with the zio-blocks-next- prefix
libraryDependencies += "dev.zio" %% "zio-blocks-next-streams" % "0.0.56"

For a Scala.js or cross-built project, use %%% instead of %%. How to Add ZIO Blocks to Your Build explains the two artifact names. Streams need JDK 21 at runtime on the JVM.

This guide assumes you have worked through Your First Streaming Pipeline and Getting Started with Async. To follow along, we need a stream of order lines and a price service. MenuPrices.priceInCents answers after a 100 ms delay on another thread, the way a remote call would, and fails for an item the menu does not list. The prices are sample data:

import zio.blocks.async._
import zio.blocks.chunk.Chunk
import zio.blocks.streams._

import java.util.concurrent.CompletableFuture
import java.util.concurrent.TimeUnit.MILLISECONDS

final case class OrderLine(item: String, table: Int)

object MenuPrices {
  val order: List[OrderLine] = List(
    OrderLine("Espresso", 4),
    OrderLine("Flat White", 7),
    OrderLine("Croissant", 4)
  )

  private val cents: Map[String, Long] = Map(
    "Espresso"   -> 350L,
    "Flat White" -> 480L,
    "Croissant"  -> 420L
  )

  def priceInCents(item: String): Async[Long] =
    Async.fromCompletionStage(
      CompletableFuture.supplyAsync(
        () => cents.getOrElse(item, throw new NoSuchElementException(s"No price for $item")),
        CompletableFuture.delayedExecutor(100, MILLISECONDS)
      )
    )
}

priceInCents wraps a CompletableFuture with Async.fromCompletionStage. How to Wrap Callback and Future-Based APIs in Async covers that bridge and its siblings for Future, JavaScript promises and callbacks.

Swapping a Blocking Terminal for Its Async Twin​

Every blocking terminal has a twin whose name ends in Async: Stream#runCollect becomes Stream#runCollectAsync, Stream#runDrain becomes Stream#runDrainAsync, and Stream#runFold becomes Stream#runFoldAsync. The twin returns Async[Either[E, Z]]: the same Either the blocking terminal returns, wrapped in an Async that completes later.

To collect the price of every line in the order:

val prices: Stream[Nothing, Long] =
  Stream.fromIterator(MenuPrices.order.iterator).mapAsync(line => MenuPrices.priceInCents(line.item))

val pending: Async[Either[Nothing, Chunk[Long]]] = prices.runCollectAsync

Creating pending takes no time, because the terminal returns at once and the work happens when something drives the Async. To get the answer, we drive it with Async#block:

println(pending.block)
// Right(Chunk(350,480,420))

Running the same stream with the blocking runCollect and with runCollectAsync and timing the two calls shows the difference:

runCollect returned after 434 ms with Right(Chunk(350,480,420))
runCollectAsync returned after 1 ms, nothing has run yet
the result is Right(Chunk(350,480,420))

The blocking call holds its thread for the three lookups; the async call hands back an Async immediately. The availability table in Platform Differences lists the replacement for every blocking member.

warning

An async terminal does not make a blocking source non-blocking. The Async runs on whatever thread drives it, so a source whose pull blocks, such as an iterator that sleeps, a file read or Stream.fromInputStream, still blocks that thread. In a probe with an iterator that sleeps 200 ms per element, runCollectAsync returns after about 21 ms, and Async#block then takes about 654 ms because the sleeping runs inside it. Only sources and callbacks that report completion through an Async, like mapAsync over MenuPrices.priceInCents, wait without holding a thread.

Doing Async Work for Each Element​

mapAsync lets a stage return an Async, but often we only want the final answer. runForeachAsync and runFoldAsync take a step function that returns an Async and run it one element at a time, so no thread waits for the price of an item.

runForeachAsync runs an effect for each element and returns Async[Either[E, Unit]]. To print each order line with its price:

val order: Stream[Nothing, OrderLine] = Stream.fromIterator(MenuPrices.order.iterator)

val printed: Async[Either[Nothing, Unit]] =
  order.runForeachAsync { line =>
    MenuPrices.priceInCents(line.item).map(cents => println(s"${line.item} for table ${line.table}: $cents cents"))
  }

To run printed and print one line per order line:

printed.block
// Espresso for table 4: 350 cents
// Flat White for table 7: 480 cents
// Croissant for table 4: 420 cents
// res2: Either[Nothing, Unit] = Right(())

runFoldAsync threads an accumulator through the stream. To total the order:

val total: Async[Either[Nothing, Long]] =
  order.runFoldAsync(0L)((sum, line) => MenuPrices.priceInCents(line.item).map(sum + _))

To print the total:

println(total.block)
// Right(1250)

runFoldAsync has an overload for each primitive accumulator, Double, Float, Int and Long, and a generic one. The compiler picks the overload from the static type of the zero, which is why we wrote 0L and not 0. The primitive overloads keep the accumulator unboxed, so a fold over a primitive total does not allocate a box per step. The Float overload has no blocking twin.

The step function runs for one element at a time. Each lookup finishes before the next one starts, so this total takes about 300 ms for three order lines. To overlap the lookups, use mapParAsync on the stream; Asynchronous Stream Execution covers the bounded-concurrency operator.

Using the Result Without Waiting​

An Async is a value you pass around, so most of the time we do not block at all. We write the pipeline as a function that returns an Async, and let the caller compose it:

def totalInCents(items: Stream[Nothing, String]): Async[Either[Nothing, Long]] =
  items.runFoldAsync(0L)((sum, item) => MenuPrices.priceInCents(item).map(sum + _))

Async#map transforms the result when it arrives, with no thread waiting in between:

val items: Stream[Nothing, String] = Stream.fromIterator(MenuPrices.order.iterator.map(_.item))

val label: Async[Either[Nothing, String]] =
  totalInCents(items).map(_.map(cents => f"The order costs ${cents / 100.0}%.2f"))

To print the label:

println(label.block)
// Right(The order costs 12.50)

Code that has not moved to Async usually wants a Future. On the JVM, Async#toFuture takes an implicit ExecutionContext and returns one. On Scala.js, Async#toJsPromise returns a js.Promise. To hand the label to a Future-based caller:

import scala.concurrent.{Await, Future}
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration._

val future: Future[Either[Nothing, String]] = label.toFuture
println(Await.result(future, 5.seconds))

To run the Async in the background and keep a handle that you can join or cancel, use Async#start, which returns an Async.Running. Getting Started with Async covers it.

tip

Block only at the outermost edge of a JVM program, such as the end of main, where there is nothing left to do but wait. A call to Async#block anywhere else holds a thread, which is what we set out to avoid. On Scala.js, Async#block returns a value that is already complete and throws IllegalStateException when it would have to wait, so there is no edge to block at. The host drives the Async, as the next sections show.

Errors and Cancellation​

A stream has two kinds of failure, and the Async keeps them apart. A typed error, the E of Stream[E, A], arrives as Left inside the result. A defect, an exception thrown by user code or a failed Async inside a step, fails the Async itself.

A typed error lands in the Either:

val typed: Stream[String, Long] = Stream.fail("the price service is down")
// typed: Stream[String, Long] = Stream.fail(...)
println(typed.runFoldAsync(0L)((sum, cents) => Async.succeed(sum + cents)).block)
// Left(the price service is down)

A defect fails the Async. We total the order with an item the menu does not list, and look at it through Async#either, then through Async#block:

val unlisted: Async[Either[Nothing, Long]] =
  order.runFoldAsync(0L) { (sum, line) =>
    val item = if (line.item == "Flat White") "Flat White XL" else line.item
    MenuPrices.priceInCents(item).map(sum + _)
  }

To see the defect as an Either and as a thrown exception:

println(unlisted.either.block)
// Left(java.util.NoSuchElementException: No price for Flat White XL)
println(scala.util.Try(unlisted.block))
// Failure(java.util.NoSuchElementException: No price for Flat White XL)

Async#either moves the failure into an Either[Throwable, ...] value, and Async#block rethrows it, so one try or Try at the edge covers every defect.

Cancelling a running terminal stops the pulls and runs the stream's finalizers. To see it, we drain an endless stream of lookups with a finalizer, start it, and cancel it after 350 ms:

import java.util.concurrent.atomic.AtomicInteger

val pulls     = new AtomicInteger
val finalized = new AtomicInteger

val endless: Stream[Nothing, Long] =
  Stream
    .fromIterator(Iterator.continually(MenuPrices.order.head))
    .mapAsync { line =>
      pulls.incrementAndGet()
      MenuPrices.priceInCents(line.item)
    }
    .ensuring { finalized.incrementAndGet(); () }

val running = endless.runDrainAsync.start
Thread.sleep(350)
running.cancel()
val pullsAtCancel = pulls.get()
Thread.sleep(300)
println(s"pulls at cancel: $pullsAtCancel, pulls 300 ms later: ${pulls.get()}, finalizer ran: ${finalized.get()}")

Running it prints:

pulls at cancel: 4, pulls 300 ms later: 4, finalizer ran: 1

No pull starts after the cancel, and the finalizer runs exactly once. One thing the cancel does not do is reach inside the lookup that was in flight: a lookup built from a Future, CompletionStage or JavaScript promise keeps running after you stop waiting for it. How to Wrap Callback and Future-Based APIs in Async explains why and how to carry a cancel handle through.

Running the Same Pipeline on Scala.js​

Everything above used only the *Async terminals, Async#map and Async.fromCompletionStage, so only one piece is JVM-specific: the price lookup. The pipeline itself, OrderTotalJs.totalInCents, does not change if we pass the lookup in as a parameter. The Scala.js code in the rest of this guide is not compiled by the docs build; each snippet runs on Node 24 with Scala.js 1.22 on Scala 2.13.18 and 3.9.0.

On Scala.js, the lookup wraps a js.Promise with Async.fromJsPromise:

// Scala.js code
import zio.blocks.async._
import zio.blocks.streams._

import scala.scalajs.js

object OrderTotalJs {
  private val cents: Map[String, Long] = Map(
    "Espresso"   -> 350L,
    "Flat White" -> 480L,
    "Croissant"  -> 420L
  )

  def priceInCents(item: String): Async[Long] =
    Async.fromJsPromise(new js.Promise[Long]((resolve, reject) => {
      js.timers.setTimeout(100) {
        cents.get(item) match {
          case Some(price) => resolve(price)
          case None        => reject(new NoSuchElementException(s"No price for $item"))
        }
      }
      ()
    }))

  def totalInCents(items: Stream[Nothing, String], price: String => Async[Long]): Async[Either[Nothing, Long]] =
    items.runFoldAsync(0L)((sum, item) => price(item).map(sum + _))

  def main(args: Array[String]): Unit = {
    val items = Stream("Espresso", "Flat White", "Croissant")

    totalInCents(items, priceInCents).map(total => println(s"The total is $total")).toJsPromise

    println("main has returned; the total prints when the timers fire")
  }
}

Running it under Node prints:

main has returned; the total prints when the timers fire
The total is Right(1250)

main returns before the first timer fires, which is the point: nothing waits. Async#toJsPromise hands the Async to the JavaScript event loop, so Node stays alive until the promise settles. In a browser, you would keep the Async or the js.Promise and update the page in the callback.

Two things differ from the JVM, and the compiler or the runtime tells you at once. A blocking terminal does not exist on Scala.js, so the call fails to compile:

value runCollect is not a member of zio.blocks.streams.Stream[Nothing, Int]

Async#block does exist in shared source, so it cannot be removed. It returns a value that is already complete and throws when it would have to wait:

A ready value: 1
A pending value: Async.block: suspension did not complete synchronously and JavaScript cannot block. Drive the Pollable from a non-blocking entry point instead.

Reading a ReadableStream as a Source​

In a browser or Node 18 and later, a fetch response body is a ReadableStream of bytes. ReadableStreamReaders.fromReadableStream wraps one as a Reader.AsyncReader[Byte], and Stream.fromReader turns the reader into a Stream[E, Byte]. The adapter is Scala.js-only and byte-only: every chunk the JavaScript stream delivers is a Uint8Array, and the stream you get has element type Byte. It calls read() on the JavaScript stream only when the zio-blocks stream pulls for more bytes, and has at most one read in flight, so a slow consumer slows the reads rather than piling up chunks.

To fetch an order list, one item per line, and total its prices with the same OrderTotalJs.totalInCents, we first read the bytes and then decode them. The URL is a data: URL, so the example needs no network. Node and the browsers fetch it like any other:

// Scala.js code
import zio.blocks.async._
import zio.blocks.chunk.Chunk
import zio.blocks.streams._

import scala.scalajs.js

object OrderListJs {
  private val orderUrl =
    "data:text/plain,Espresso%0AFlat%20White%0ACroissant%0A"

  def itemsIn(bytes: Chunk[Byte]): Stream[Nothing, String] =
    Stream.fromIterator(new String(bytes.toArray, "UTF-8").split("\n").iterator.filter(_.nonEmpty))

  def main(args: Array[String]): Unit = {
    val response = js.Dynamic.global.fetch(orderUrl).asInstanceOf[js.Promise[js.Dynamic]]

    val total: Async[Either[Nothing, Long]] =
      Async.fromJsPromise(response).flatMap { resp =>
        val body = resp.body.asInstanceOf[ReadableStreamReaders.ReadableStream]
        val bytes: Stream[Nothing, Byte] =
          Stream.fromReader[Nothing, Byte](ReadableStreamReaders.fromReadableStream(body))

        bytes.runCollectAsync.flatMap {
          case Right(all) => OrderTotalJs.totalInCents(itemsIn(all), OrderTotalJs.priceInCents)
          case Left(e)    => Async.succeed(Left(e))
        }
      }

    total.map(result => println(s"The order list costs $result")).toJsPromise
    ()
  }
}
The order list costs Right(1250)

ReadableStreamReaders declares its own small @js.native facades, ReadableStreamReaders.ReadableStream among them, so we cast the body of the untyped fetch result to that facade. The factory takes the stream and calls getReader() itself.

ReadableStreamReaders.fromReadableStream owns the stream. When the zio-blocks stream closes early, for example under take(3), the adapter cancels the JavaScript stream and releases its lock. To see it, we give the stream a cancel hook and take three of its five bytes:

// Scala.js code
import java.util.concurrent.atomic.AtomicBoolean

import zio.blocks.async._
import zio.blocks.streams._

import scala.scalajs.js
import scala.scalajs.js.typedarray.Uint8Array

object EarlyCloseJs {
  def main(args: Array[String]): Unit = {
    val cancelled = new AtomicBoolean(false)

    val source = js.Dynamic.literal(
      start = { (controller: js.Dynamic) =>
        controller.enqueue(new Uint8Array(js.Array[Short](1, 2, 3, 4, 5)))
      }: js.Function1[js.Dynamic, Unit],
      cancel = { (reason: js.Any) =>
        cancelled.set(true)
      }: js.Function1[js.Any, Unit]
    )
    val stream = js.Dynamic
      .newInstance(js.Dynamic.global.ReadableStream)(source)
      .asInstanceOf[ReadableStreamReaders.ReadableStream]

    Stream
      .fromReader[Nothing, Byte](ReadableStreamReaders.fromReadableStream(stream))
      .take(3)
      .runCollectAsync
      .map(result => println(s"take(3) gave $result, stream cancelled: ${cancelled.get()}"))
      .toJsPromise
    ()
  }
}

Running it prints:

take(3) gave Right(Chunk(1,2,3)), stream cancelled: true

If the stream is not yours to cancel, use ReadableStreamReaders.fromReadableStreamUnmanaged, which releases the lock and never cancels, so the code that creates the stream can keep using it. Reader documents both factories.

warning

In 0.0.56, a read that the JavaScript stream rejects, for example a dropped connection, does not arrive as Left in the Either the way the JVM channel adapter's failures do. It fails the Async itself with an internal StreamError, whose message is empty, so handle it with Async#either or a catch at the edge. This is a known bug in 0.0.56:

The read failed: Left(zio.blocks.streams.internal.StreamError)

That line is the result of runCollectAsync.either on a stream whose second read is rejected with new js.Error("network reset").

Putting It Together​

The complete example puts the pipeline behind a PriceLookup function, so the same receipt could take the Scala.js lookup. It uses the OrderLine case class and the MenuPrices price service from the support file MenuPrices.scala, which Running the Examples embeds first. It prints a line per order line with runForeachAsync, totals the order with runFoldAsync, joins the two with Async#flatMap, and blocks once, at the end of main. A second run adds an item the menu does not list and looks at the failure through Async#either. The complete example, embedded from its companion file, follows:

streams-examples/src/main/scala/runstreamswithoutblocking/CompleteExample.scala
package runstreamswithoutblocking

import zio.blocks.async._
import zio.blocks.streams.Stream

/**
 * How to Run Streams Without Blocking — Putting It Together
 *
 * One pipeline written against a price lookup that returns an `Async`: it
 * prints a line per order line and totals the order without ever parking a
 * thread. `main` blocks once, at the very end.
 *
 * Run with: cd streams-examples && sbt "runMain
 * runstreamswithoutblocking.CompleteExample"
 */
object CompleteExample {
  type PriceLookup = String => Async[Long]

  def receipt(order: List[OrderLine], price: PriceLookup): Async[Either[Nothing, Long]] = {
    val lines: Stream[Nothing, OrderLine] = Stream.fromIterator(order.iterator)

    val printed: Async[Either[Nothing, Unit]] =
      lines.runForeachAsync { line =>
        price(line.item).map(cents => println(s"${line.item} for table ${line.table}: $cents cents"))
      }

    val total: Async[Either[Nothing, Long]] =
      lines.runFoldAsync(0L)((sum, line) => price(line.item).map(sum + _))

    printed.flatMap(_ => total)
  }

  def main(args: Array[String]): Unit = {
    val ok: Async[Either[Nothing, Long]] = receipt(MenuPrices.order, MenuPrices.priceInCents)
    println(s"Total for the order: ${ok.block} cents")

    // An item the menu does not list fails the Async itself, so `either` is the way to look at it.
    val unlisted = OrderLine("Cold Brew", 9)
    val failing  = receipt(MenuPrices.order :+ unlisted, MenuPrices.priceInCents)
    println(s"With an unlisted item: ${failing.either.block}")
  }
}

Running the Examples​

Every JVM snippet in this guide has a standalone companion file in the streams-examples build. Clone the repository and move into that build:

git clone https://github.com/zio/zio-blocks.git
cd zio-blocks/streams-examples

Every example uses the OrderLine case class and the MenuPrices price service:

The Price Service

The price service and the OrderLine case class:

streams-examples/src/main/scala/runstreamswithoutblocking/MenuPrices.scala
package runstreamswithoutblocking

import zio.blocks.async._

import java.util.concurrent.CompletableFuture
import java.util.concurrent.TimeUnit.MILLISECONDS

final case class OrderLine(item: String, table: Int)

/**
 * A fictional price service for a coffee shop menu. `priceInCents` answers
 * after a 100 ms delay on another thread, the way a remote call would, and
 * fails for an item the menu does not list. Prices are sample data.
 */
object MenuPrices {
  val order: List[OrderLine] = List(
    OrderLine("Espresso", 4),
    OrderLine("Flat White", 7),
    OrderLine("Croissant", 4)
  )

  private val cents: Map[String, Long] = Map(
    "Espresso"   -> 350L,
    "Flat White" -> 480L,
    "Croissant"  -> 420L
  )

  def priceInCents(item: String): Async[Long] =
    Async.fromCompletionStage(
      CompletableFuture.supplyAsync(
        () => cents.getOrElse(item, throw new NoSuchElementException(s"No price for $item")),
        CompletableFuture.delayedExecutor(100, MILLISECONDS)
      )
    )
}
Swap a Blocking Terminal for Its Async Twin

The terminal swap, timed against the blocking call:

streams-examples/src/main/scala/runstreamswithoutblocking/Concept1SwapTerminalExample.scala
package runstreamswithoutblocking

import zio.blocks.async._
import zio.blocks.chunk.Chunk
import zio.blocks.streams.Stream

/**
 * How to Run Streams Without Blocking — Swap a Blocking Terminal for Its Async
 * Twin
 *
 * Runs one stream of price lookups with `runCollect` and with
 * `runCollectAsync`, and shows when each call returns.
 *
 * Run with: cd streams-examples && sbt "runMain
 * runstreamswithoutblocking.Concept1SwapTerminalExample"
 */
object Concept1SwapTerminalExample {
  def main(args: Array[String]): Unit = {
    val prices: Stream[Nothing, Long] =
      Stream.fromIterator(MenuPrices.order.iterator).mapAsync(line => MenuPrices.priceInCents(line.item))

    val blockedAt = System.nanoTime()
    val blocking  = prices.runCollect
    println(s"runCollect returned after ${elapsedMillis(blockedAt)} ms with $blocking")

    val startedAt                                    = System.nanoTime()
    val pending: Async[Either[Nothing, Chunk[Long]]] = prices.runCollectAsync
    println(s"runCollectAsync returned after ${elapsedMillis(startedAt)} ms, nothing has run yet")
    println(s"the result is ${pending.block}")
  }

  private def elapsedMillis(since: Long): Long = (System.nanoTime() - since) / 1000000
}

Run it with:

sbt "runMain runstreamswithoutblocking.Concept1SwapTerminalExample"
Do Async Work for Each Element

Per-element async work with runForeachAsync and runFoldAsync:

streams-examples/src/main/scala/runstreamswithoutblocking/Concept2PerElementExample.scala
package runstreamswithoutblocking

import zio.blocks.async._
import zio.blocks.streams.Stream

/**
 * How to Run Streams Without Blocking — Do Async Work for Each Element
 *
 * Totals the price of every line of an order with `runFoldAsync`, and prints
 * each price as it arrives with `runForeachAsync`. The step function returns an
 * `Async`, so no thread waits for a price.
 *
 * Run with: cd streams-examples && sbt "runMain
 * runstreamswithoutblocking.Concept2PerElementExample"
 */
object Concept2PerElementExample {
  def main(args: Array[String]): Unit = {
    val order: Stream[Nothing, OrderLine] = Stream.fromIterator(MenuPrices.order.iterator)

    val printed: Async[Either[Nothing, Unit]] =
      order.runForeachAsync { line =>
        MenuPrices.priceInCents(line.item).map(cents => println(s"${line.item} for table ${line.table}: $cents cents"))
      }

    val total: Async[Either[Nothing, Long]] =
      order.runFoldAsync(0L)((sum, line) => MenuPrices.priceInCents(line.item).map(sum + _))

    println(s"printed: ${printed.block}")
    println(s"total: ${total.block} cents")
  }
}

Run it with:

sbt "runMain runstreamswithoutblocking.Concept2PerElementExample"
Use the Result Without Waiting

Composing the result and handing it to a Future:

streams-examples/src/main/scala/runstreamswithoutblocking/Concept3UsingResultExample.scala
package runstreamswithoutblocking

import zio.blocks.async._
import zio.blocks.streams.Stream

import scala.concurrent.Await
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration._

/**
 * How to Run Streams Without Blocking — Use the Result Without Waiting
 *
 * Composes the result of a terminal with `map`, hands it to code that expects a
 * `Future`, and blocks only at the end of `main`.
 *
 * Run with: cd streams-examples && sbt "runMain
 * runstreamswithoutblocking.Concept3UsingResultExample"
 */
object Concept3UsingResultExample {
  def totalInCents(items: Stream[Nothing, String]): Async[Either[Nothing, Long]] =
    items.runFoldAsync(0L)((sum, item) => MenuPrices.priceInCents(item).map(sum + _))

  def main(args: Array[String]): Unit = {
    val items: Stream[Nothing, String] = Stream.fromIterator(MenuPrices.order.iterator.map(_.item))

    // `map` on the Async: no thread waits, the function runs when the total is ready.
    val label: Async[Either[Nothing, String]] =
      totalInCents(items).map(_.map(cents => f"The order costs ${cents / 100.0}%.2f"))

    // Hand the same Async to code that wants a Future.
    val future = label.toFuture
    println(s"Future completed yet: ${future.isCompleted}")

    // The one blocking call: the end of main, which has nothing left to do.
    println(Await.result(future, 5.seconds))
    println(label.block)
  }
}

Run it with:

sbt "runMain runstreamswithoutblocking.Concept3UsingResultExample"
Errors and Cancellation

Typed errors, defects and cancellation:

streams-examples/src/main/scala/runstreamswithoutblocking/Concept4ErrorsAndCancellationExample.scala
package runstreamswithoutblocking

import zio.blocks.async._
import zio.blocks.streams.Stream

import java.util.concurrent.atomic.AtomicInteger

/**
 * How to Run Streams Without Blocking — Errors and Cancellation
 *
 * Shows where a typed error, a defect and a cancel end up, and that cancelling
 * a running terminal stops the pulls and runs the stream's finalizers.
 *
 * Run with: cd streams-examples && sbt "runMain
 * runstreamswithoutblocking.Concept4ErrorsAndCancellationExample"
 */
object Concept4ErrorsAndCancellationExample {
  def main(args: Array[String]): Unit = {
    // A typed error lands in the Either.
    val typed: Stream[String, Long] = Stream.fail("the price service is down")
    println(s"typed error: ${typed.runFoldAsync(0L)((sum, cents) => Async.succeed(sum + cents)).block}")

    // A defect lands in the Async's own failure channel.
    val lines: Stream[Nothing, OrderLine] = Stream.fromIterator(MenuPrices.order.iterator)
    val unlisted                          = lines.runFoldAsync(0L) { (sum, line) =>
      MenuPrices.priceInCents(if (line.item == "Flat White") "Flat White XL" else line.item).map(sum + _)
    }
    println(s"defect, as an Either: ${unlisted.either.block}")
    println(s"defect, through block: ${scala.util.Try(unlisted.block)}")

    // Cancel a long run: the pulls stop and the finalizer runs.
    val pulls                          = new AtomicInteger
    val finalized                      = new AtomicInteger
    val endless: Stream[Nothing, Long] =
      Stream
        .fromIterator(Iterator.continually(MenuPrices.order.head))
        .mapAsync { line =>
          pulls.incrementAndGet()
          MenuPrices.priceInCents(line.item)
        }
        .ensuring { finalized.incrementAndGet(); () }

    val running = endless.runDrainAsync.start
    Thread.sleep(350)
    running.cancel()
    val pullsAtCancel = pulls.get()
    Thread.sleep(300)
    println(s"pulls at cancel: $pullsAtCancel, pulls 300 ms later: ${pulls.get()}, finalizer ran: ${finalized.get()}")
  }
}

Run it with:

sbt "runMain runstreamswithoutblocking.Concept4ErrorsAndCancellationExample"
Complete Example: An Order Receipt

The complete order receipt:

streams-examples/src/main/scala/runstreamswithoutblocking/CompleteExample.scala
package runstreamswithoutblocking

import zio.blocks.async._
import zio.blocks.streams.Stream

/**
 * How to Run Streams Without Blocking — Putting It Together
 *
 * One pipeline written against a price lookup that returns an `Async`: it
 * prints a line per order line and totals the order without ever parking a
 * thread. `main` blocks once, at the very end.
 *
 * Run with: cd streams-examples && sbt "runMain
 * runstreamswithoutblocking.CompleteExample"
 */
object CompleteExample {
  type PriceLookup = String => Async[Long]

  def receipt(order: List[OrderLine], price: PriceLookup): Async[Either[Nothing, Long]] = {
    val lines: Stream[Nothing, OrderLine] = Stream.fromIterator(order.iterator)

    val printed: Async[Either[Nothing, Unit]] =
      lines.runForeachAsync { line =>
        price(line.item).map(cents => println(s"${line.item} for table ${line.table}: $cents cents"))
      }

    val total: Async[Either[Nothing, Long]] =
      lines.runFoldAsync(0L)((sum, line) => price(line.item).map(sum + _))

    printed.flatMap(_ => total)
  }

  def main(args: Array[String]): Unit = {
    val ok: Async[Either[Nothing, Long]] = receipt(MenuPrices.order, MenuPrices.priceInCents)
    println(s"Total for the order: ${ok.block} cents")

    // An item the menu does not list fails the Async itself, so `either` is the way to look at it.
    val unlisted = OrderLine("Cold Brew", 9)
    val failing  = receipt(MenuPrices.order :+ unlisted, MenuPrices.priceInCents)
    println(s"With an unlisted item: ${failing.either.block}")
  }
}

Run it with:

sbt "runMain runstreamswithoutblocking.CompleteExample"

To check that every example compiles without running any of them, run:

sbt compile

The streams-examples build is JVM-only, so the Scala.js snippets have no companion files. To run one, make a scratch sbt project with addSbtPlugin("org.scala-js" % "sbt-scalajs" % "1.22.0") in project/plugins.sbt and this build.sbt, copy a snippet into src/main/scala, and run it with sbt run, which needs Node on your path:

scalaVersion := "3.9.0"
enablePlugins(ScalaJSPlugin)
scalaJSUseMainModuleInitializer := true
scalacOptions += "-experimental"
libraryDependencies += "dev.zio" %%% "zio-blocks-streams" % "0.0.56"

Going Further​