Skip to main content

How to Use ZIO Blocks in a ZIO or cats-effect Application

In this guide, we use ZIO Blocks inside an application that already runs on ZIO 2 or cats-effect 3. By the end, you will have a small orders service that parses JSON orders, stores them in SQLite, authorizes payments through an asynchronous gateway, writes an archive file and streams order events, with every ZIO Blocks call running where your effect system expects it, and with interruption and cleanup working across the boundary.

We take five cases one at a time: pure modules, blocking calls, resources, Async values and streams.

Every section shows a ZIO tab and a cats-effect tab. Only ZIO has library help, the sql-zio module. Every other bridge on this page is a few lines of your own code, which you copy into your project.

The Problem​

ZIO Blocks says it needs no effect system. Then the question is where its calls go in an application that has one. Wrapping each call in the first constructor that compiles works on a laptop, and fails under load or on interruption.

For example, here is a first attempt at three operations of an orders service, written for ZIO 2.1.27 (not compiled here):

def allOrders(transactor: JdbcTransactor): Task[List[Order]] =
  ZIO.succeed(transactor.connect(OrdersDb.all()))

def authorize(gateway: PaymentGateway, order: Order): Task[String] =
  ZIO.fromFuture(implicit ec => gateway.authorize(order).toFuture)

def events(path: Path): Task[List[OrderEvent]] =
  ZIO.succeed(OrderEvents.fromFile(path).runCollect).absolve.map(_.toList)

Each line compiles, and each one is wrong in a way no test notices. allOrders runs JDBC on ZIO's compute threads, and an SQLException becomes a defect instead of a failure. authorize loses cancellation: when the fiber is interrupted, the Future from Async#toFuture can't cancel the payment authorization, so it runs on. events reads the whole file into memory on a compute thread, so it is neither a stream nor interruptible. The cats-effect versions with IO.delay and IO.fromFuture block a compute thread and lose cancellation in the same way.

In this guide, we put each kind of ZIO Blocks call where it belongs, and write one small bridge per direction for the cases where ZIO Blocks and the effect system both manage something: a resource, a running computation or a stream.

Prerequisites​

Add the ZIO Blocks modules this guide uses, the SQLite JDBC driver, and your effect system. The examples are tested with ZIO 2.1.27, cats-effect 3.7.1 and fs2 3.14.0:

// Scala 2.13 and Scala 3.3 to 3.8
libraryDependencies ++= Seq(
  "dev.zio" %% "zio-blocks-schema"  % "0.0.56",
  "dev.zio" %% "zio-blocks-sql"     % "0.0.56",
  "dev.zio" %% "zio-blocks-streams" % "0.0.56",
  "dev.zio" %% "zio-blocks-async"   % "0.0.56",
  "dev.zio" %% "zio-blocks-scope"   % "0.0.56"
)

// Scala 3.9 and newer: the same modules as zio-blocks-next-*, plus -experimental
libraryDependencies += "dev.zio" %% "zio-blocks-next-schema" % "0.0.56" // and so on
scalacOptions += "-experimental"

// The SQLite JDBC driver
libraryDependencies += "org.xerial" % "sqlite-jdbc" % "3.53.4.0"

// For a ZIO application
libraryDependencies ++= Seq(
  "dev.zio" %% "zio"         % "2.1.27",
  "dev.zio" %% "zio-streams" % "2.1.27"
)

// For a cats-effect application
libraryDependencies ++= Seq(
  "org.typelevel" %% "cats-effect" % "3.7.1",
  "co.fs2"        %% "fs2-core"    % "3.14.0"
)

The sql module is published for Scala 3 only, so the orders model and the programs that use it are Scala 3. The bridges themselves (the resource, Async and stream helpers below) compile on Scala 2.13 and Scala 3. The streams module needs JDK 21 or newer at runtime.

This guide assumes you know your effect system, and the basics of Async, Scope and Resource and Stream.

The Orders Model​

The service keeps orders in an orders table and reads order events from a JSON-lines file. Everything in this file is ZIO Blocks code, with no effect system in sight: Order and OrderEvent derive a Schema for JSON, OrdersDb holds the SQL, and OrderEvents.fromFile is a ZIO Blocks Stream that opens the file when it runs and prints a line when it closes it:

zio-blocks-examples/effect-integration/src/main/scala/effectintegration/Orders.scala
package effectintegration

import java.nio.charset.StandardCharsets.UTF_8
import java.nio.file.{Files, Path}
import scala.jdk.CollectionConverters._
import zio.blocks.schema._
import zio.blocks.sql._
import zio.blocks.streams.Stream

final case class Order(id: Long, customer: String, totalCents: Long)

object Order {
  implicit val schema: Schema[Order] = Schema.derived
}

final case class OrderEvent(orderId: Long, status: String)

object OrderEvent {
  implicit val schema: Schema[OrderEvent] = Schema.derived
}

object OrdersDb {
  def createTable()(using DbCon): Unit =
    sql"""CREATE TABLE orders (
      id INTEGER PRIMARY KEY,
      customer TEXT NOT NULL,
      total_cents INTEGER NOT NULL
    )""".update

  def insert(order: Order)(using DbCon): Int =
    sql"INSERT INTO orders (id, customer, total_cents) VALUES (${order.id}, ${order.customer}, ${order.totalCents})".update

  def all()(using DbCon): List[Order] =
    sql"SELECT id, customer, total_cents FROM orders ORDER BY id".query[Order]

  def totals()(using DbCon): Stream[Throwable, Long] =
    sql"SELECT total_cents FROM orders".queryStream[Long].flatMap(chunk => Stream.fromIterable(chunk))

  // A fresh SQLite file with an empty orders table.
  def open(): JdbcTransactor = {
    val file = Files.createTempFile("orders", ".db")
    file.toFile.deleteOnExit()
    val transactor = JdbcTransactor.fromUrl(s"jdbc:sqlite:$file", SqlDialect.SQLite)
    transactor.connect(createTable())
    transactor
  }
}

object OrderEvents {
  // One JSON object per line. The file opens when the stream runs and closes when the run ends.
  def fromFile(path: Path): Stream[SchemaError, OrderEvent] =
    Stream.fromAcquireRelease(
      Files.newBufferedReader(path, UTF_8),
      reader => { reader.close(); println("order-events file closed") }
    ) { reader =>
      Stream
        .fromIterator(reader.lines().iterator().asScala)
        .flatMap(line => line.fromJson[OrderEvent].fold(Stream.fail(_), Stream.succeed(_)))
    }

  def write(lines: String*): Path = {
    val path = Files.createTempFile("order-events", ".jsonl")
    path.toFile.deleteOnExit()
    Files.write(path, lines.asJava, UTF_8)
  }
}

OrdersDb.totals returns a Stream of the order totals from Frag#queryStream. Like every Frag#queryStream result, it must be consumed inside the JdbcTransactor#connect block that opened its connection.

Call the Pure Modules Directly​

schema, its codecs, optics, html and parsing configuration from a string are pure: they start no threads, do no I/O and hold no resources. Call them from your effect like any other library function. A decode returns Either[SchemaError, A], and SchemaError extends Exception, so it fits the error channel of ZIO.fromEither and of IO.fromEither as it is. An encode can't fail, so ZIO.succeed and IO.pure are enough:

zio-blocks-examples/effect-integration/src/main/scala/effectintegration/Step1PureZio.scala
//> using scala 3.9.0
//> using dep dev.zio::zio-blocks-schema:0.0.56
//> using dep dev.zio::zio-blocks-sql:0.0.56
//> using dep dev.zio::zio-blocks-streams:0.0.56
//> using dep dev.zio::zio:2.1.27
//> using options -experimental
//> using jvm 21
//> using file Orders.scala
// Run from this folder:  scala-cli run Step1PureZio.scala
// Or from zio-blocks-examples:  sbt "effectIntegration/runMain effectintegration.Step1PureZio"

package effectintegration

import zio._
import zio.blocks.schema._

object Step1PureZio extends ZIOAppDefault {
  def parseOrder(body: String): IO[SchemaError, Order] = ZIO.fromEither(body.fromJson[Order])

  def render(order: Order): UIO[String] = ZIO.succeed(order.toJsonString)

  val run =
    for {
      order  <- parseOrder("""{"id":1,"customer":"Ada","totalCents":4200}""")
      _      <- Console.printLine(s"parsed: $order")
      json   <- render(order.copy(totalCents = 4700))
      _      <- Console.printLine(s"rendered: $json")
      result <- parseOrder("""{"id":2,"customer":"Grace"}""").either
      _      <- Console.printLine(s"missing field: ${result.left.map(_.getMessage)}")
    } yield ()
}

Both programs print the same lines. The missing field arrives in the error channel as a typed SchemaError:

parsed: Order(1,Ada,4200)
rendered: {"id":1,"customer":"Ada","totalCents":4700}
missing field: Left(missing required field "totalCents" at: .)

Run Blocking Calls on the Blocking Pool​

Some ZIO Blocks calls block the calling thread: every JDBC call, every blocking Stream terminal on the JVM, and Async#block. Run them with ZIO.attemptBlocking or IO.blocking, so they use the effect system's pool for blocking work instead of the few compute threads its fibers share. Building a Stream or composing a Resource blocks nothing, because nothing runs until a terminal or a scope runs it:

ZIO Blocks callBlocks the thread?Where to run it
Schema codecs, optics, html renderingNoAnywhere
Building a Stream (Stream.fromIterable, Stream#map, ...)No, it is a descriptionAnywhere
Transactor#connect, Transactor#transact, Frag#queryYes, JDBCZIO.attemptBlocking, IO.blocking
Stream#runCollect, Stream#runFold, Stream#runForeachYes, on the JVMZIO.attemptBlocking, IO.blocking
Async#blockYes, until the Async completesZIO.attemptBlocking, IO.blocking
Scope#scoped, Scope#allocateRuns acquire and release on the callerBlocking pool if they do I/O
Building an Async (Async.attempt, Async#map)Runs its synchronous work at onceLike the work it does

Errors map in three shapes. A typed Either from a decode or from a stream terminal joins the error channel with ZIO.fromEither or ZIO#absolve, and with IO.fromEither or IO#rethrow. An exception that a ZIO Blocks call throws, such as an SQLException, becomes a failure inside ZIO.attemptBlocking and IO.blocking. A defect stays a defect.

The orders service stores two orders, lists them, sums their totals with a stream that it consumes inside JdbcTransactor#connect, and then inserts a duplicate id. Each blocking call prints the thread it runs on:

zio-blocks-examples/effect-integration/src/main/scala/effectintegration/Step2BlockingZio.scala
//> using scala 3.9.0
//> using dep dev.zio::zio-blocks-schema:0.0.56
//> using dep dev.zio::zio-blocks-sql:0.0.56
//> using dep dev.zio::zio-blocks-streams:0.0.56
//> using dep org.xerial:sqlite-jdbc:3.53.4.0
//> using dep dev.zio::zio:2.1.27
//> using options -experimental
//> using jvm 21
//> using file Orders.scala
// Run from this folder:  scala-cli run Step2BlockingZio.scala
// Or from zio-blocks-examples:  sbt "effectIntegration/runMain effectintegration.Step2BlockingZio"

package effectintegration

import zio._
import zio.blocks.sql._

object Step2BlockingZio extends ZIOAppDefault {
  def thread: String = Thread.currentThread.getName

  def saveOrder(transactor: JdbcTransactor, order: Order): Task[Unit] =
    ZIO.attemptBlocking {
      println(s"insert order ${order.id} on $thread")
      transactor.transact(OrdersDb.insert(order))
    }.unit

  // The stream must be consumed inside `connect`; its Either joins the error channel.
  def revenue(transactor: JdbcTransactor): Task[Long] =
    ZIO.attemptBlocking(transactor.connect(OrdersDb.totals().runFold(0L)(_ + _))).absolve

  val run =
    for {
      _          <- ZIO.succeed(println(s"compute work on $thread"))
      transactor <- ZIO.attemptBlocking(OrdersDb.open())
      _          <- saveOrder(transactor, Order(1, "Ada", 4200))
      _          <- saveOrder(transactor, Order(2, "Grace", 1999))
      orders     <- ZIO.attemptBlocking(transactor.connect(OrdersDb.all()))
      _          <- Console.printLine(s"orders: $orders")
      total      <- revenue(transactor)
      _          <- Console.printLine(s"revenue: $total cents")
      duplicate  <- saveOrder(transactor, Order(1, "Ada", 4200)).either
      _          <- Console.printLine(s"duplicate id: ${duplicate.left.map(_.getClass.getName)}")
    } yield ()
}

The ZIO program prints its compute thread first, then runs every insert on ZIO's blocking pool. The duplicate id fails with the driver's exception in the error channel:

compute work on ZScheduler-Worker-4
insert order 1 on zio-default-blocking-1
insert order 2 on zio-default-blocking-1
orders: List(Order(1,Ada,4200), Order(2,Grace,1999))
revenue: 6199 cents
insert order 1 on zio-default-blocking-1
duplicate id: Left(org.sqlite.SQLiteException)

The cats-effect program prints the same results. Its blocking calls run on threads named io-compute-blocker-<n>, which is how cats-effect 3 names a compute thread it has handed over to blocking work while a replacement takes its place:

compute work on io-compute-13
insert order 1 on io-compute-blocker-13
insert order 2 on io-compute-blocker-13
orders: List(Order(1,Ada,4200), Order(2,Grace,1999))
revenue: 6199 cents
insert order 1 on io-compute-blocker-13
duplicate id: Left(org.sqlite.SQLiteException)
On ZIO, sql-zio writes the wrappers for you

The sql-zio module's TransactorZIO returns ZIO effects that already run on the blocking pool, comes as a ZLayer, and runs transactions whose body is a ZIO effect. How to Use the sql Module from a ZIO Application covers it. There is no cats-effect counterpart, so on cats-effect you write IO.blocking as above.

Frag#queryStream can't leave its JdbcTransactor#connect block: the connection closes when the block returns, and ZIO Blocks has no public way to keep it open for longer. To stream query results through your effect system, consume them inside the block, as revenue does, and turn file and socket streams into effect streams with the bridge from the streams section.

Lift a ZIO Blocks Resource Into Your Effect System​

A ZIO Blocks Resource describes how to acquire a value and how to release it, and a Scope runs the release when it closes. Your effect system has the same idea, so the bridge hands the lifetime to one of the two. The archive file of placed orders is our resource. It prints a line when it opens and when it closes, so we can count the closes:

zio-blocks-examples/effect-integration/src/main/scala/effectintegration/OrderArchive.scala
package effectintegration

import java.io.BufferedWriter
import java.nio.charset.StandardCharsets.UTF_8
import java.nio.file.{Files, Path}
import zio.blocks.schema._
import zio.blocks.scope.Resource

// An append-only JSON-lines file of placed orders: a resource that must be closed exactly once.
final class OrderArchive(path: Path) extends AutoCloseable {
  private val writer: BufferedWriter = Files.newBufferedWriter(path, UTF_8)
  println("archive opened")

  def append(order: Order): Unit = {
    writer.write(order.toJsonString)
    writer.newLine()
  }

  def close(): Unit = {
    writer.close()
    println("archive closed")
  }
}

object OrderArchive {
  def resource(path: Path): Resource[OrderArchive] = Resource.fromAutoCloseable(new OrderArchive(path))
}

ZIO Blocks code normally enters a scope with Scope#scoped, whose block keeps every resource inside it. An effect system acquires and releases in two separate steps instead, so the bridge opens a child of Scope.global with Scope#open, allocates the resource in it, and takes the value out with Scope#leak. Scope#leak is the escape hatch ZIO Blocks provides for interop; it warns at compile time, and @nowarn silences that warning here because the caller closes the scope. Closing the child scope runs the resource's finalizers:

zio-blocks-examples/effect-integration/src/main/scala/effectintegration/Opened.scala
package effectintegration

import scala.annotation.nowarn
import zio.blocks.scope.{Resource, Scope}

// Opens a ZIO Blocks resource outside a `scoped` block, for an effect system that manages the
// lifetime itself. The result pairs the value with a function that releases it.
object Opened {

  @nowarn // `leak` warns on purpose: the value leaves the scope, and the caller closes it
  def resource[A](resource: Resource[A]): (A, () => Unit) = {
    val open = Scope.global.open()
    try (open.scope.leak(open.scope.allocate(resource)), () => open.close().orThrow())
    catch { case t: Throwable => open.close(); throw t }
  }
}

The bridges build on Opened.resource for one direction. For the other, they run the effect system's own acquire when ZIO Blocks allocates, and its release when the ZIO Blocks scope closes:

zio-blocks-examples/effect-integration/src/main/scala/effectintegration/ZioResources.scala
package effectintegration

import zio._
import zio.blocks.scope.Resource

object ZioResources {

  // ZIO Blocks Resource -> ZIO: acquired on the blocking pool, released when the ZIO Scope closes.
  def fromResource[A](resource: Resource[A]): ZIO[Scope, Throwable, A] =
    ZIO
      .acquireRelease(ZIO.attemptBlocking(Opened.resource(resource))) { case (_, release) =>
        ZIO.attemptBlocking(release()).orDie
      }
      .map(_._1)

  // ZIO -> ZIO Blocks Resource: runs `acquire` in a fresh ZIO Scope, closed with the ZIO Blocks scope.
  def toResource[A](acquire: ZIO[Scope, Throwable, A])(implicit runtime: Runtime[Any]): Resource[A] =
    Resource.acquireRelease {
      Unsafe.unsafe { implicit unsafe =>
        val scope = runtime.unsafe.run(Scope.make).getOrThrow()
        runtime.unsafe.run(scope.extend[Any](acquire)) match {
          case Exit.Success(a)     => (a, scope)
          case Exit.Failure(cause) =>
            runtime.unsafe.run(scope.close(Exit.failCause(cause)))
            throw cause.squash
        }
      }
    } { case (_, scope) =>
      Unsafe.unsafe(implicit unsafe => runtime.unsafe.run(scope.close(Exit.unit)).getOrThrow())
    }
      .map(_._1)
}

ZioResources.fromResource returns a ZIO[Scope, Throwable, A], so ZLayer.scoped turns it into a layer and ZIO.scoped runs it. CatsResources.fromResource returns a cats-effect Resource, which Resource#use runs. In the other direction, ZioResources.toResource and CatsResources.toResource run your effect system from inside ZIO Blocks code with Runtime#unsafe or Dispatcher#unsafeRunSync, which blocks the thread, so call ZIO Blocks code that allocates it from ZIO.attemptBlocking or IO.blocking. A Dispatcher is cats-effect's tool for running effects from code that is not itself an effect; it comes from Dispatcher.parallel[IO].

The programs below use the archive four times: once successfully, once with a failing body, once interrupted while in use, and once as a resource your effect system manages and ZIO Blocks code uses. Each use opens and closes the archive exactly once:

-- success, through a ZLayer
archive opened
archive closed
-- failure
archive opened
archive closed
result: Left(payment declined)
-- interruption
archive opened
archive closed
-- a ZIO-managed archive used from ZIO Blocks code
archive opened
archive closed

The cats-effect program prints the same lines, with -- success in place of -- success, through a ZLayer, -- cancellation in place of -- interruption, and "cats-effect-managed" in the last label.

Runnable version, the ZIO tab: Step3ResourcesZio.scala
zio-blocks-examples/effect-integration/src/main/scala/effectintegration/Step3ResourcesZio.scala
//> using scala 3.9.0
//> using dep dev.zio::zio-blocks-schema:0.0.56
//> using dep dev.zio::zio-blocks-sql:0.0.56
//> using dep dev.zio::zio-blocks-streams:0.0.56
//> using dep dev.zio::zio-blocks-scope:0.0.56
//> using dep dev.zio::zio:2.1.27
//> using options -experimental
//> using jvm 21
//> using file Orders.scala
//> using file OrderArchive.scala
//> using file Opened.scala
//> using file ZioResources.scala
// Run from this folder:  scala-cli run Step3ResourcesZio.scala
// Or from zio-blocks-examples:  sbt "effectIntegration/runMain effectintegration.Step3ResourcesZio"

package effectintegration

import java.nio.file.Files
import zio._
import zio.blocks.scope.{Scope => BlocksScope}

object Step3ResourcesZio extends ZIOAppDefault {
  val order = Order(1, "Ada", 4200)

  val run =
    ZIO.runtime[Any].flatMap { implicit runtime =>
      for {
        path   <- ZIO.attemptBlocking(Files.createTempFile("archive", ".jsonl"))
        archive = ZioResources.fromResource(OrderArchive.resource(path))
        _      <- Console.printLine("-- success, through a ZLayer")
        _      <- ZIO
               .serviceWithZIO[OrderArchive](a => ZIO.attemptBlocking(a.append(order)))
               .provide(ZLayer.scoped(archive))
        _         <- Console.printLine("-- failure")
        failed    <- ZIO.scoped(archive *> ZIO.fail(new Exception("payment declined"))).either
        _         <- Console.printLine(s"result: ${failed.left.map(_.getMessage)}")
        _         <- Console.printLine("-- interruption")
        fiber     <- ZIO.scoped(archive *> ZIO.never).fork
        _         <- ZIO.sleep(100.millis)
        _         <- fiber.interrupt
        _         <- Console.printLine("-- a ZIO-managed archive used from ZIO Blocks code")
        zioManaged = ZioResources.toResource(ZIO.fromAutoCloseable(ZIO.attempt(new OrderArchive(path))))
        _         <- ZIO.attemptBlocking {
               BlocksScope.global.scoped { scope =>
                 import scope._
                 val a = allocate(zioManaged)
                 $(a)(_.append(order))
               }
             }
        _ <- ZIO.attemptBlocking(Files.delete(path))
      } yield ()
    }
}
Runnable version, the cats-effect tab: Step3ResourcesCats.scala
zio-blocks-examples/effect-integration/src/main/scala/effectintegration/Step3ResourcesCats.scala
//> using scala 3.9.0
//> using dep dev.zio::zio-blocks-schema:0.0.56
//> using dep dev.zio::zio-blocks-sql:0.0.56
//> using dep dev.zio::zio-blocks-streams:0.0.56
//> using dep dev.zio::zio-blocks-scope:0.0.56
//> using dep org.typelevel::cats-effect:3.7.1
//> using options -experimental
//> using jvm 21
//> using file Orders.scala
//> using file OrderArchive.scala
//> using file Opened.scala
//> using file CatsResources.scala
// Run from this folder:  scala-cli run Step3ResourcesCats.scala
// Or from zio-blocks-examples:  sbt "effectIntegration/runMain effectintegration.Step3ResourcesCats"

package effectintegration

import java.nio.file.Files
import scala.concurrent.duration._
import cats.effect.{IO, IOApp, Resource}
import cats.effect.std.Dispatcher
import zio.blocks.scope.Scope

object Step3ResourcesCats extends IOApp.Simple {
  val order = Order(1, "Ada", 4200)

  val run =
    Dispatcher.parallel[IO].use { implicit dispatcher =>
      for {
        path       <- IO.blocking(Files.createTempFile("archive", ".jsonl"))
        archive     = CatsResources.fromResource(OrderArchive.resource(path))
        _          <- IO.println("-- success")
        _          <- archive.use(a => IO.blocking(a.append(order)))
        _          <- IO.println("-- failure")
        failed     <- archive.use(_ => IO.raiseError[Unit](new Exception("payment declined"))).attempt
        _          <- IO.println(s"result: ${failed.left.map(_.getMessage)}")
        _          <- IO.println("-- cancellation")
        fiber      <- archive.use(_ => IO.never[Unit]).start
        _          <- IO.sleep(100.millis)
        _          <- fiber.cancel
        _          <- IO.println("-- a cats-effect-managed archive used from ZIO Blocks code")
        catsManaged = CatsResources.toResource(Resource.fromAutoCloseable(IO.blocking(new OrderArchive(path))))
        _          <- IO.blocking {
               Scope.global.scoped { scope =>
                 import scope._
                 val a = allocate(catsManaged)
                 $(a)(_.append(order))
               }
             }
        _ <- IO.blocking(Files.delete(path))
      } yield ()
    }
}

Bridge Async and Your Effect, With Cancellation​

Async is eager: building one runs its synchronous work at once, like a Future, while ZIO and IO values are descriptions that run each time the runtime executes them. So the bridge into your effect takes the Async by name and builds it each time the effect runs. Cancellation must also cross in both directions: interrupting the fiber must cancel the Async, and cancelling the Async must interrupt the effect.

Async#toFuture followed by ZIO.fromFuture or IO.fromFuture loses cancellation in both directions, because a Future can't be cancelled. Async.fromFuture and Async.fromCompletionStage don't forward cancellation to their source either (a known bug in 0.0.56). The bridges therefore use the cancellable helper from How to Wrap Callback and Future-Based APIs, which builds an Async whose cancellation calls a function you supply:

zio-blocks-examples/effect-integration/src/main/scala/effectintegration/Cancellable.scala
package effectintegration

import zio.blocks.async.{Async, Completer, Pollable}

// The `cancellable` helper from "How to Wrap Callback and Future APIs": an `Async` whose
// cancellation calls the function that `register` returns.
object Cancellable {
  def cancellable[A](register: Completer[A] => (() => Unit)): Async[A] =
    new Pollable[A] {
      private val result     = new Completer[A]
      private val cancelCall = register(result)

      def poll(onComplete: Runnable): Async[A] = {
        val polled = result.poll(onComplete)
        // While pending, answer with this Pollable rather than the Completer, so
        // the driver keeps waiting on us and our cancel() is the one it calls.
        if (polled.asInstanceOf[AnyRef] eq result) this else polled
      }

      override def cancel(): Unit = cancelCall()
    }
}

The payment gateway is ZIO Blocks code with a callback API. It authorizes a payment after a delay, and cancelling the Async cancels the pending authorization and prints a line:

zio-blocks-examples/effect-integration/src/main/scala/effectintegration/PaymentGateway.scala
package effectintegration

import java.util.concurrent.{Executors, ScheduledExecutorService, TimeUnit}
import zio.blocks.async.Async
import Cancellable.cancellable

// ZIO Blocks code with an asynchronous API: authorizing a payment takes `latencyMillis`,
// and cancelling the `Async` cancels the pending authorization.
final class PaymentGateway(latencyMillis: Long) {
  private val timer: ScheduledExecutorService =
    Executors.newSingleThreadScheduledExecutor { (r: Runnable) =>
      val thread = new Thread(r, "payment-gateway")
      thread.setDaemon(true)
      thread
    }

  def authorize(order: Order): Async[String] =
    cancellable[String] { completer =>
      val pending = timer.schedule(
        new Runnable { def run(): Unit = completer.succeed(s"auth-${order.id}") },
        latencyMillis,
        TimeUnit.MILLISECONDS
      )
      () => {
        pending.cancel(false)
        println(s"payment gateway: authorization of order ${order.id} cancelled")
      }
    }
}

Into your effect, the bridge starts the Async with Async#start, reports its outcome through Async#foldCause to the effect system's callback, and returns Async.Running#cancel as the effect's canceller. Out of your effect, it runs the effect with a cancellable future and hands cancellable the function that cancels it:

zio-blocks-examples/effect-integration/src/main/scala/effectintegration/ZioAsync.scala
package effectintegration

import scala.concurrent.ExecutionContext
import zio._
import zio.blocks.async._
import Cancellable.cancellable

object ZioAsync {

  // ZIO Blocks Async -> ZIO: each run builds the Async; interrupting the fiber cancels it.
  def fromAsync[A](async: => Async[A]): Task[A] =
    ZIO.asyncInterrupt[Any, Throwable, A] { callback =>
      val running = async.foldCause(e => callback(ZIO.fail(e)))(a => callback(ZIO.succeed(a))).start
      Left(ZIO.succeed(running.cancel()))
    }

  // ZIO -> ZIO Blocks Async: starts the effect now; cancelling the Async interrupts its fiber.
  def toAsync[A](task: Task[A])(implicit runtime: Runtime[Any]): Async[A] =
    cancellable[A] { completer =>
      val future                        = Unsafe.unsafe(implicit unsafe => runtime.unsafe.runToFuture(task))
      implicit val ec: ExecutionContext = ExecutionContext.parasitic
      future.onComplete(_.fold(completer.fail, completer.succeed))
      () => { future.cancel(); () }
    }
}

The programs below authorize one payment, then authorize again under a 100 ms timeout that is shorter than the gateway's 300 ms delay. In the other direction, ZIO Blocks code blocks on a fraud check written as a ZIO or cats-effect effect, then starts a second check and cancels it after 100 ms. The ZIO program prints:

authorized: auth-1
payment gateway: authorization of order 1 cancelled
with a 100 ms timeout: None
fraud check passed: true
fraud check of order 1 interrupted
after cancelling the Async

The timeout interrupts the fiber, and the interruption reaches the gateway, which cancels the authorization. Cancelling the Async interrupts the fraud check's fiber, whose ZIO#onInterrupt prints its line before the marker printed afterwards. The cats-effect program prints timed out from IO#timeoutTo instead of None, and cancelled from IO#onCancel.

Runnable version, the ZIO tab: Step4AsyncZio.scala
zio-blocks-examples/effect-integration/src/main/scala/effectintegration/Step4AsyncZio.scala
//> using scala 3.9.0
//> using dep dev.zio::zio-blocks-schema:0.0.56
//> using dep dev.zio::zio-blocks-sql:0.0.56
//> using dep dev.zio::zio-blocks-streams:0.0.56
//> using dep dev.zio::zio-blocks-async:0.0.56
//> using dep dev.zio::zio:2.1.27
//> using options -experimental
//> using jvm 21
//> using file Orders.scala
//> using file PaymentGateway.scala
//> using file Cancellable.scala
//> using file ZioAsync.scala
// Run from this folder:  scala-cli run Step4AsyncZio.scala
// Or from zio-blocks-examples:  sbt "effectIntegration/runMain effectintegration.Step4AsyncZio"

package effectintegration

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

object Step4AsyncZio extends ZIOAppDefault {
  val gateway = new PaymentGateway(latencyMillis = 300)
  val order   = Order(1, "Ada", 4200)

  // A ZIO service that ZIO Blocks code wants to call.
  def fraudCheck(order: Order): Task[Boolean] =
    ZIO
      .sleep(1.second)
      .as(order.totalCents < 100000)
      .onInterrupt(Console.printLine(s"fraud check of order ${order.id} interrupted").orDie)

  val run =
    ZIO.runtime[Any].flatMap { implicit runtime =>
      for {
        auth     <- ZioAsync.fromAsync(gateway.authorize(order))
        _        <- Console.printLine(s"authorized: $auth")
        timedOut <- ZioAsync.fromAsync(gateway.authorize(order)).timeout(100.millis)
        _        <- Console.printLine(s"with a 100 ms timeout: $timedOut")
        passed   <- ZIO.attemptBlocking(ZioAsync.toAsync(fraudCheck(order)).block)
        _        <- Console.printLine(s"fraud check passed: $passed")
        running  <- ZIO.succeed(ZioAsync.toAsync(fraudCheck(order)).start)
        _        <- ZIO.sleep(100.millis)
        _        <- ZIO.succeed(running.cancel())
        _        <- ZIO.sleep(100.millis)
        _        <- Console.printLine("after cancelling the Async")
      } yield ()
    }
}
Runnable version, the cats-effect tab: Step4AsyncCats.scala
zio-blocks-examples/effect-integration/src/main/scala/effectintegration/Step4AsyncCats.scala
//> using scala 3.9.0
//> using dep dev.zio::zio-blocks-schema:0.0.56
//> using dep dev.zio::zio-blocks-sql:0.0.56
//> using dep dev.zio::zio-blocks-streams:0.0.56
//> using dep dev.zio::zio-blocks-async:0.0.56
//> using dep org.typelevel::cats-effect:3.7.1
//> using options -experimental
//> using jvm 21
//> using file Orders.scala
//> using file PaymentGateway.scala
//> using file Cancellable.scala
//> using file CatsAsync.scala
// Run from this folder:  scala-cli run Step4AsyncCats.scala
// Or from zio-blocks-examples:  sbt "effectIntegration/runMain effectintegration.Step4AsyncCats"

package effectintegration

import scala.concurrent.duration._
import cats.effect.{IO, IOApp}
import cats.effect.std.Dispatcher
import zio.blocks.async._

object Step4AsyncCats extends IOApp.Simple {
  val gateway = new PaymentGateway(latencyMillis = 300)
  val order   = Order(1, "Ada", 4200)

  // A cats-effect service that ZIO Blocks code wants to call.
  def fraudCheck(order: Order): IO[Boolean] =
    IO.sleep(1.second)
      .as(order.totalCents < 100000)
      .onCancel(IO.println(s"fraud check of order ${order.id} cancelled"))

  val run =
    Dispatcher.parallel[IO].use { implicit dispatcher =>
      for {
        auth     <- CatsAsync.fromAsync(gateway.authorize(order))
        _        <- IO.println(s"authorized: $auth")
        timedOut <- CatsAsync.fromAsync(gateway.authorize(order)).timeoutTo(100.millis, IO.pure("timed out"))
        _        <- IO.println(s"with a 100 ms timeout: $timedOut")
        passed   <- IO.blocking(CatsAsync.toAsync(fraudCheck(order)).block)
        _        <- IO.println(s"fraud check passed: $passed")
        running  <- IO(CatsAsync.toAsync(fraudCheck(order)).start)
        _        <- IO.sleep(100.millis)
        _        <- IO(running.cancel())
        _        <- IO.sleep(100.millis)
        _        <- IO.println("after cancelling the Async")
      } yield ()
    }
}

Turn a ZIO Blocks Stream Into a ZStream or fs2 Stream​

A ZIO Blocks Stream runs on the thread that pulls it, and a pull can block. To read one from a ZStream or an fs2 stream, we open it once with Stream#start, pull chunks of elements on the blocking pool, and close it when the effect stream ends, fails or is interrupted. BlockingPull wraps that reader:

zio-blocks-examples/effect-integration/src/main/scala/effectintegration/BlockingPull.scala
package effectintegration

import scala.annotation.nowarn
import zio.blocks.chunk.Chunk
import zio.blocks.scope.Scope
import zio.blocks.streams.Stream
import zio.blocks.streams.internal.StreamError
import zio.blocks.streams.io.Reader

// A ZIO Blocks stream opened for pulling from outside, for an effect system's stream.
// Every call blocks, so call it on a blocking pool.
final class BlockingPull[E, A] private (open: Scope.OpenScope, reader: Reader.SyncReader[A]) {
  @volatile private var failed = false

  // Up to `max` elements, or an empty chunk at the end. A pull throws the stream's typed error
  // wrapped in `StreamError`, an internal class; this unwraps it into a `Left`.
  def next(max: Int): Either[E, Chunk[A]] =
    try Right(reader.readUpToN[A](max))
    catch {
      case error: StreamError => failed = true; Left(error.value.asInstanceOf[E])
      case defect: Throwable  => failed = true; throw defect
    }

  // After a failed pull the stream's finalizers have already run, and closing the reader
  // throws that failure again (ZIO Blocks 0.0.56), so close errors are ignored then.
  def close(): Unit = {
    val finalization = open.close()
    if (!failed) finalization.orThrow()
  }
}

object BlockingPull {
  @nowarn // `leak` warns on purpose: the reader leaves the scope, and `close` closes it
  def apply[E, A](stream: Stream[E, A]): BlockingPull[E, A] = {
    val open = Scope.global.open()
    try new BlockingPull[E, A](open, open.scope.leak(stream.start(open.scope)))
    catch { case t: Throwable => open.close(); throw t }
  }
}

Reader.SyncReader#readUpToN returns at most max elements and an empty chunk at the end, so the effect stream asks for one chunk at a time and the ZIO Blocks stream never runs further ahead than that: this is the back-pressure. Two details are workarounds for 0.0.56. A pull throws the stream's typed error wrapped in StreamError, a class in the internal package, so BlockingPull#next unwraps it into a Left. And after a failed pull, closing the reader throws the same failure again, although the stream's finalizers have already run, so BlockingPull#close ignores close errors then (a known bug in 0.0.56).

The stream bridges use BlockingPull in one direction. In the other, each run of the ZIO Blocks stream starts the effect stream behind a bounded buffer, and closing the run interrupts or cancels it. ZIO has ZStream#toIterator for this; fs2 has no equivalent, so CatsStreams.fromFs2 feeds a bounded Queue from a background fiber:

zio-blocks-examples/effect-integration/src/main/scala/effectintegration/ZioStreams.scala
package effectintegration

import zio._
import zio.stream.ZStream
import zio.blocks.streams.Stream

object ZioStreams {

  // ZIO Blocks Stream -> ZStream: pulls up to `chunkSize` elements at a time on the blocking pool,
  // and closes the stream when the ZStream ends, fails or is interrupted.
  def toZStream[E, A](stream: => Stream[E, A], chunkSize: Int = 256): ZStream[Any, E, A] =
    ZStream
      .acquireReleaseWith(ZIO.attemptBlocking(BlockingPull(stream)).orDie)(pull =>
        ZIO.attemptBlocking(pull.close()).orDie
      )
      .flatMap { pull =>
        ZStream.repeatZIOChunkOption {
          ZIO.attemptBlocking(pull.next(chunkSize)).orDie.flatMap {
            case Left(error)                   => ZIO.fail(Some(error))
            case Right(chunk) if chunk.isEmpty => ZIO.fail(None)
            case Right(chunk)                  => ZIO.succeed(Chunk.fromIterable(chunk))
          }
        }
      }

  // ZStream -> ZIO Blocks Stream: each run starts the ZStream; closing the run interrupts it.
  def fromZStream[E, A](zstream: ZStream[Any, E, A])(implicit runtime: Runtime[Any]): Stream[E, A] =
    Stream.fromResource(ZioResources.toResource(zstream.toIterator)) { iterator =>
      Stream.fromIterator(iterator).flatMap(_.fold(Stream.fail(_), Stream.succeed(_)))
    }
}

The programs below read the order-events file whole, read only its first event, read a file with a malformed line, and interrupt a stream that never ends. Then they show back-pressure both ways, by printing each element the producer emits while the consumer takes three. The ZIO program prints:

-- whole file
order-events file closed
events: Chunk(placed,paid,shipped)
-- first event only
order-events file closed
events: Chunk(placed)
-- a malformed line
order-events file closed
result: Left(expected '{' at: .)
-- interruption
endless stream closed
-- back-pressure, ZIO Blocks -> ZStream
zio blocks produced 1
zio blocks produced 2
zio blocks produced 3
zio blocks produced 4
zstream took: Chunk(1,2,3)
-- back-pressure, ZStream -> ZIO Blocks
zstream produced 1
zstream produced 2
zstream produced 3
zstream finalized
zio blocks took: Right(Chunk(1,2,3))

The file is closed exactly once in each run: after the last event, after ZStream#take has what it needs, and after the malformed line, whose SchemaError arrives in the ZStream's typed error channel. With a chunk size of 2, the ZIO Blocks stream produces 4 elements for a consumer that takes 3, and no more.

The cats-effect program prints the same lines with List in place of Chunk, fs2 in place of zstream, and "cancellation" in its label. Its last block shows the bounded queue, of size 2 here, letting the fs2 stream run a few elements ahead before it is cancelled:

-- back-pressure, fs2 -> ZIO Blocks
fs2 produced 1
fs2 produced 2
fs2 produced 3
fs2 produced 4
fs2 produced 5
fs2 produced 6
fs2 finalized
zio blocks took: Right(Chunk(1,2,3))
Runnable version, the ZIO tab: Step5StreamsZio.scala
zio-blocks-examples/effect-integration/src/main/scala/effectintegration/Step5StreamsZio.scala
//> using scala 3.9.0
//> using dep dev.zio::zio-blocks-schema:0.0.56
//> using dep dev.zio::zio-blocks-sql:0.0.56
//> using dep dev.zio::zio-blocks-streams:0.0.56
//> using dep dev.zio::zio-blocks-scope:0.0.56
//> using dep dev.zio::zio:2.1.27
//> using dep dev.zio::zio-streams:2.1.27
//> using options -experimental
//> using jvm 21
//> using file Orders.scala
//> using file BlockingPull.scala
//> using file ZioStreams.scala
//> using file ZioResources.scala
//> using file Opened.scala
// Run from this folder:  scala-cli run Step5StreamsZio.scala
// Or from zio-blocks-examples:  sbt "effectIntegration/runMain effectintegration.Step5StreamsZio"

package effectintegration

import zio._
import zio.stream.ZStream
import zio.blocks.streams.Stream

object Step5StreamsZio extends ZIOAppDefault {
  val placed  = """{"orderId":1,"status":"placed"}"""
  val paid    = """{"orderId":1,"status":"paid"}"""
  val shipped = """{"orderId":1,"status":"shipped"}"""

  val run =
    ZIO.runtime[Any].flatMap { implicit runtime =>
      for {
        events <- ZIO.attemptBlocking(OrderEvents.write(placed, paid, shipped))
        _      <- Console.printLine("-- whole file")
        all    <- ZioStreams.toZStream(OrderEvents.fromFile(events)).runCollect
        _      <- Console.printLine(s"events: ${all.map(_.status)}")
        _      <- Console.printLine("-- first event only")
        first  <- ZioStreams.toZStream(OrderEvents.fromFile(events)).take(1).runCollect
        _      <- Console.printLine(s"events: ${first.map(_.status)}")
        _      <- Console.printLine("-- a malformed line")
        broken <- ZIO.attemptBlocking(OrderEvents.write(placed, "not json", shipped))
        failed <- ZioStreams.toZStream(OrderEvents.fromFile(broken)).runCollect.either
        _      <- Console.printLine(s"result: ${failed.left.map(_.getMessage)}")
        _      <- Console.printLine("-- interruption")
        endless = Stream.fromIterator(Iterator.from(1)).ensuring(println("endless stream closed"))
        fiber  <- ZioStreams.toZStream(endless).runDrain.fork
        _      <- ZIO.sleep(100.millis)
        _      <- fiber.interrupt
        _      <- Console.printLine("-- back-pressure, ZIO Blocks -> ZStream")
        counted = Stream.fromIterable(1 to 1000).map { n => println(s"zio blocks produced $n"); n }
        taken  <- ZioStreams.toZStream(counted, chunkSize = 2).take(3).runCollect
        _      <- Console.printLine(s"zstream took: $taken")
        _      <- Console.printLine("-- back-pressure, ZStream -> ZIO Blocks")
        ticks   = ZStream
                  .iterate(1)(_ + 1)
                  .tap(n => Console.printLine(s"zstream produced $n"))
                  .ensuring(Console.printLine("zstream finalized").orDie)
        back <- ZIO.attemptBlocking(ZioStreams.fromZStream(ticks).take(3).runCollect)
        _    <- Console.printLine(s"zio blocks took: $back")
      } yield ()
    }
}
Runnable version, the cats-effect tab: Step5StreamsCats.scala
zio-blocks-examples/effect-integration/src/main/scala/effectintegration/Step5StreamsCats.scala
//> using scala 3.9.0
//> using dep dev.zio::zio-blocks-schema:0.0.56
//> using dep dev.zio::zio-blocks-sql:0.0.56
//> using dep dev.zio::zio-blocks-streams:0.0.56
//> using dep dev.zio::zio-blocks-scope:0.0.56
//> using dep org.typelevel::cats-effect:3.7.1
//> using dep co.fs2::fs2-core:3.14.0
//> using options -experimental
//> using jvm 21
//> using file Orders.scala
//> using file BlockingPull.scala
//> using file CatsStreams.scala
//> using file CatsResources.scala
//> using file Opened.scala
// Run from this folder:  scala-cli run Step5StreamsCats.scala
// Or from zio-blocks-examples:  sbt "effectIntegration/runMain effectintegration.Step5StreamsCats"

package effectintegration

import scala.concurrent.duration._
import cats.effect.{IO, IOApp}
import cats.effect.std.Dispatcher
import zio.blocks.streams.Stream

object Step5StreamsCats extends IOApp.Simple {
  val placed  = """{"orderId":1,"status":"placed"}"""
  val paid    = """{"orderId":1,"status":"paid"}"""
  val shipped = """{"orderId":1,"status":"shipped"}"""

  val run =
    Dispatcher.parallel[IO].use { implicit dispatcher =>
      for {
        events <- IO.blocking(OrderEvents.write(placed, paid, shipped))
        _      <- IO.println("-- whole file")
        all    <- CatsStreams.toFs2(OrderEvents.fromFile(events)).compile.toList
        _      <- IO.println(s"events: ${all.map(_.status)}")
        _      <- IO.println("-- first event only")
        first  <- CatsStreams.toFs2(OrderEvents.fromFile(events)).take(1).compile.toList
        _      <- IO.println(s"events: ${first.map(_.status)}")
        _      <- IO.println("-- a malformed line")
        broken <- IO.blocking(OrderEvents.write(placed, "not json", shipped))
        failed <- CatsStreams.toFs2(OrderEvents.fromFile(broken)).compile.toList.attempt
        _      <- IO.println(s"result: ${failed.left.map(_.getMessage)}")
        _      <- IO.println("-- cancellation")
        endless = Stream.fromIterator(Iterator.from(1)).ensuring(println("endless stream closed"))
        fiber  <- CatsStreams.toFs2(endless).compile.drain.start
        _      <- IO.sleep(100.millis)
        _      <- fiber.cancel
        _      <- IO.println("-- back-pressure, ZIO Blocks -> fs2")
        counted = Stream.fromIterable(1 to 1000).map { n => println(s"zio blocks produced $n"); n }
        taken  <- CatsStreams.toFs2(counted, chunkSize = 2).take(3).compile.toList
        _      <- IO.println(s"fs2 took: $taken")
        _      <- IO.println("-- back-pressure, fs2 -> ZIO Blocks")
        ticks   = fs2.Stream
                  .iterate(1)(_ + 1)
                  .covary[IO]
                  .evalTap(n => IO.println(s"fs2 produced $n"))
                  .onFinalize(IO.println("fs2 finalized"))
        back <- IO.blocking(CatsStreams.fromFs2(ticks, buffer = 2).take(3).runCollect)
        _    <- IO.println(s"zio blocks took: $back")
      } yield ()
    }
}

Putting It Together​

The complete service uses each case once. It parses three JSON requests (the third has no total), authorizes each payment through the gateway with a one-second timeout that fails the order instead of returning None, stores the order on the blocking pool, and appends it to the archive, which the program opens as a lifted ZIO Blocks resource for its whole run. A request that fails to parse or to authorize is printed as rejected: … and does not stop the run. Then it sums the revenue with a stream inside JdbcTransactor#connect, and streams a shipment events file through ZStream or fs2, keeping only the shipped events. Unlike the sections, the gateway answers after 50 ms:

zio-blocks-examples/effect-integration/src/main/scala/effectintegration/CompleteZio.scala
//> using scala 3.9.0
//> using dep dev.zio::zio-blocks-schema:0.0.56
//> using dep dev.zio::zio-blocks-sql:0.0.56
//> using dep dev.zio::zio-blocks-streams:0.0.56
//> using dep dev.zio::zio-blocks-scope:0.0.56
//> using dep dev.zio::zio-blocks-async:0.0.56
//> using dep org.xerial:sqlite-jdbc:3.53.4.0
//> using dep dev.zio::zio:2.1.27
//> using dep dev.zio::zio-streams:2.1.27
//> using options -experimental
//> using jvm 21
//> using file Orders.scala
//> using file OrderArchive.scala
//> using file PaymentGateway.scala
//> using file Cancellable.scala
//> using file Opened.scala
//> using file ZioResources.scala
//> using file ZioAsync.scala
//> using file BlockingPull.scala
//> using file ZioStreams.scala
// Run from this folder:  scala-cli run CompleteZio.scala
// Or from zio-blocks-examples:  sbt "effectIntegration/runMain effectintegration.CompleteZio"

package effectintegration

import java.nio.file.Files
import zio._
import zio.blocks.schema._
import zio.blocks.sql._

object CompleteZio extends ZIOAppDefault {
  val gateway = new PaymentGateway(latencyMillis = 50)

  val requests = List(
    """{"id":1,"customer":"Ada","totalCents":4200}""",
    """{"id":2,"customer":"Grace","totalCents":1999}""",
    """{"id":3,"customer":"Linus"}"""
  )

  def placeOrder(transactor: JdbcTransactor, archive: OrderArchive)(body: String): Task[String] =
    for {
      order <- ZIO.fromEither(body.fromJson[Order])
      auth  <- ZioAsync.fromAsync(gateway.authorize(order)).timeoutFail(new Exception("payment timed out"))(1.second)
      _     <- ZIO.attemptBlocking(transactor.transact(OrdersDb.insert(order)))
      _     <- ZIO.attemptBlocking(archive.append(order))
    } yield s"order ${order.id} placed ($auth)"

  val run =
    ZIO.scoped {
      for {
        transactor <- ZIO.attemptBlocking(OrdersDb.open())
        path       <- ZIO.attemptBlocking(Files.createTempFile("archive", ".jsonl"))
        archive    <- ZioResources.fromResource(OrderArchive.resource(path))
        _          <- ZIO.foreachDiscard(requests) { body =>
               placeOrder(transactor, archive)(body).foldZIO(
                 error => Console.printLine(s"rejected: ${error.getMessage}"),
                 placed => Console.printLine(placed)
               )
             }
        revenue <- ZIO.attemptBlocking(transactor.connect(OrdersDb.totals().runFold(0L)(_ + _))).absolve
        _       <- Console.printLine(s"revenue: $revenue cents")
        events  <- ZIO.attemptBlocking(
                    OrderEvents.write(
                      """{"orderId":1,"status":"shipped"}""",
                      """{"orderId":2,"status":"paid"}""",
                      """{"orderId":2,"status":"shipped"}"""
                    )
                  )
        _ <- ZioStreams
               .toZStream(OrderEvents.fromFile(events))
               .filter(_.status == "shipped")
               .foreach(event => Console.printLine(s"shipped: order ${event.orderId}"))
      } yield ()
    }
}

Both programs print the same lines. The archive closes last, when the program's scope or Resource ends:

archive opened
order 1 placed (auth-1)
order 2 placed (auth-2)
rejected: missing required field "totalCents" at: .
revenue: 6199 cents
shipped: order 1
shipped: order 2
order-events file closed
archive closed

Running the Examples​

All runnable code from this guide is in zio-blocks-examples/effect-integration/src/main/scala/effectintegration/. Run any file from that folder with scala-cli run <File>.scala, or from zio-blocks-examples with sbt "effectIntegration/runMain effectintegration.<ObjectName>".

Going Further​