How to Stream Files and Sockets Without Leaking Resources

In this guide, we read a Scala bookshop's inventory export with ZIO Blocks Streams so that every file, channel and socket the stream opens is closed again, whether the stream finishes, fails, stops early or never finds its input. By the end, you will have a stream that opens a file only when it runs, a typed error for a missing file, two resources that close in reverse order, a file copy that closes both ends, and a socket read that disconnects when it is done.

We start with one file and one acquire/release pair, then break the stream on purpose in three ways, move to scope Resource values, compare the managed and unmanaged I/O constructors, and finish with sinks, sockets and finalizers:

Reading a File That Always Closes
Stopping Early Still Closes the File
Failing Mid-Stream Still Closes the File
Reporting a Missing File as a Typed Error
Reusing and Composing Scope Resources
Managed and Unmanaged Handles
Copying a File Through a Sink
Reading from a Socket
Cleanup That Is Not a Resource
Pulling Elements by Hand

The Problem

A file handle must be closed exactly once, after the last row is read, and also when reading stops early or fails halfway. A lazy sequence makes that hard: the code that opens the file returns before the code that reads it runs, so there is no single place where try/finally covers both.

Without a streaming library, reading the inventory usually goes one of three ways:

import java.io.BufferedReader
import java.nio.file.{Files, Path}
import scala.jdk.CollectionConverters._

// Lazy, but nobody ever closes the reader.
def inventoryRowsLeaky(path: Path): Iterator[String] = {
  val reader: BufferedReader = Files.newBufferedReader(path)
  reader.lines().iterator().asScala
}

// Closes the reader, but before the caller has read a single row.
def inventoryRowsClosedTooSoon(path: Path): Iterator[String] = {
  val reader: BufferedReader = Files.newBufferedReader(path)
  try reader.lines().iterator().asScala
  finally reader.close()
}

// Correct, but every caller repeats it, and the rows must not escape the block.
def lowStockTitles(path: Path): List[String] = {
  val reader: BufferedReader = Files.newBufferedReader(path)
  try {
    reader
      .lines()
      .iterator()
      .asScala
      .filter(row => row.split(",")(1).toInt < 5)
      .map(row => row.split(",")(0))
      .toList
  } finally reader.close()
}

The first version leaks one file handle per call until the process runs out of them. The second fails on the first read with IOException: Stream closed. The third works, but it reads everything into a List, and each new question about the inventory needs another copy of the try/finally. Sockets have the same problem with a connection instead of a file.

In this guide, we let the stream own the handle: it opens the handle when it starts, and closes it when it stops, for whatever reason it stops.

Prerequisites

Add the streams module to your build.sbt:

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

// Scala 3.7 and newer
libraryDependencies += "dev.zio" %% "zio-blocks-next-streams" % "0.0.56"

The streams module depends on the scope module, so zio.blocks.scope.Resource and Scope are available without another dependency. On the JVM, the streams module needs JDK 21 or newer. NioStreams, NioSinks, Stream#start and the blocking terminals such as Stream#runCollect exist only on the JVM; the other constructors on this page also work on Scala.js, where you run the stream with an asynchronous terminal such as Stream#runCollectAsync.

This guide assumes you have run a Stream before and know that every terminal returns an Either with typed failures on the Left. The beginner tutorial "Your First Streaming Pipeline" covers that ground. For Resource and Scope, see Compile-Time Resource Safety with Scope.

Most snippets below share these imports:

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

import java.io.BufferedReader
import java.nio.charset.StandardCharsets.UTF_8
import java.nio.file.{Files, Path}
import scala.jdk.CollectionConverters._

The bookshop exports its inventory as one title,copies row per book. We write a copy to a bookshop directory under the system's temporary directory so that every snippet runs on any machine:

import java.nio.file.Paths

val shop: Path = Files.createDirectories(Paths.get(System.getProperty("java.io.tmpdir"), "bookshop"))

val inventory: Path = shop.resolve("inventory.csv")

Files.writeString(
  inventory,
  """Zionomicon,12
    |Programming in Scala,7
    |Functional Programming in Scala,3
    |Scala for the Impatient,0
    |Hands-on Scala Programming,4
    |""".stripMargin
)

Reading a File That Always Closes

Stream.fromAcquireRelease takes three parts: acquire opens the resource, release closes it, and a function receives the open resource and returns the stream that reads it. We print a line in acquire and in release so that the output shows when each one runs:

def inventoryRows(path: Path): Stream[Nothing, String] =
  Stream.fromAcquireRelease(
    acquire = {
      println(s"opened ${path.getFileName}")
      Files.newBufferedReader(path)
    },
    release = (reader: BufferedReader) => {
      reader.close()
      println(s"closed ${path.getFileName}")
    }
  )(reader => Stream.fromIterator(reader.lines().iterator().asScala))

acquire is a by-name parameter, so calling inventoryRows opens nothing. Each run of the stream evaluates acquire once, builds the inner stream from BufferedReader#lines, and calls release exactly once when that run ends.

Running it with Stream#runForeach prints each row as it arrives:

inventoryRows(inventory).runForeach(row => println(row))
// opened inventory.csv
// Zionomicon,12
// Programming in Scala,7
// Functional Programming in Scala,3
// Scala for the Impatient,0
// Hands-on Scala Programming,4
// closed inventory.csv
// res2: Either[Nothing, Unit] = Right(())

The file opens before the first row and closes after the last one. When the resource is an AutoCloseable, as a BufferedReader is, you can leave release out: its default calls AutoCloseable#close. We keep the explicit release on this page only for the printed line.

Stopping Early Still Closes the File

Stream#take stops pulling after a given number of elements. The rest of the file is never read, and the stream still closes it:

inventoryRows(inventory).take(2).runForeach(row => println(row))
// opened inventory.csv
// Zionomicon,12
// Programming in Scala,7
// closed inventory.csv
// res3: Either[Nothing, Unit] = Right(())

closed inventory.csv appears after two rows, while three rows are still unread. The same holds for Stream#takeWhile, for Stream#head, and for any sink that finishes before the input ends.

Failing Mid-Stream Still Closes the File

A second export has a damaged row, where the copy count is a word instead of a number:

val damaged: Path = shop.resolve("inventory-damaged.csv")

Files.writeString(
  damaged,
  """Zionomicon,12
    |Programming in Scala,seven
    |Functional Programming in Scala,3
    |""".stripMargin
)

final case class Stock(title: String, copies: Int)

Parsing with String#toInt throws on the damaged row. A thrown exception is a defect: the terminal rethrows it instead of returning a Left, and the stream closes the file on the way out:

def parseOrThrow(row: String): Stock = {
  val fields = row.split(",")
  Stock(fields(0), fields(1).toInt)
}

try println(inventoryRows(damaged).map(parseOrThrow).runCollect)
catch { case e: NumberFormatException => println(s"thrown: $e") }
// opened inventory-damaged.csv
// closed inventory-damaged.csv
// thrown: java.lang.NumberFormatException: For input string: "seven"

To make a bad row an expected outcome instead of a crash, we fail the stream with a typed error. Stream.fail emits no elements and fails with its argument, and Stream#flatMap lets each row choose between a one-element stream and a failed one:

final case class BadRow(row: String)

def parse(row: String): Stream[BadRow, Stock] = {
  val fields = row.split(",")
  fields(1).toIntOption match {
    case Some(copies) => Stream.succeed(Stock(fields(0), copies))
    case None         => Stream.fail(BadRow(row))
  }
}

println(inventoryRows(damaged).flatMap(parse).runCollect)
// opened inventory-damaged.csv
// closed inventory-damaged.csv
// Left(BadRow(Programming in Scala,seven))

The run ends with Left(BadRow(...)), and the file is closed before the terminal returns, exactly as on success.

Recover on the outer stream

If you add recovery with Stream#catchAll, attach it to the outer stream, as in inventoryRows(damaged).flatMap(parse).catchAll(...), not to the stream that parse returns. In 0.0.56, a blocking terminal such as Stream#runCollect throws the internal Stream$AsyncBoundaryRequired$ when Stream#catchAll sits inside Stream#flatMap and a stateful operator such as Stream#grouped follows it.

Reporting a Missing File as a Typed Error

acquire can fail too. When the file does not exist, Files.newBufferedReader throws NoSuchFileException, and like any exception thrown by acquire, it is a defect that the terminal rethrows:

val missing: Path = shop.resolve("inventory-2025.csv")
// missing: Path = /tmp/bookshop/inventory-2025.csv

try println(inventoryRows(missing).runCollect)
catch { case e: java.nio.file.NoSuchFileException => println(s"thrown: $e") }
// opened inventory-2025.csv
// thrown: java.nio.file.NoSuchFileException: /tmp/bookshop/inventory-2025.csv

opened inventory-2025.csv is printed because our println runs before Files.newBufferedReader throws. No closed line follows: nothing was acquired, so there is nothing to release.

A missing input file is often a normal condition, such as an export that has not arrived yet. Stream.attempt runs a block and turns a non-fatal exception into a typed Throwable error. Opening the file with Stream.attempt and handing the open reader to Stream.fromAcquireRelease makes the failure a Left and keeps the close guarantee once the file is open:

def inventoryRowsOrError(path: Path): Stream[Throwable, String] =
  Stream.attempt(Files.newBufferedReader(path)).flatMap { reader =>
    Stream.fromAcquireRelease(reader)(r => Stream.fromIterator(r.lines().iterator().asScala))
  }

println(inventoryRowsOrError(missing).runCollect)
// Left(java.nio.file.NoSuchFileException: /tmp/bookshop/inventory-2025.csv)
println(inventoryRowsOrError(inventory).take(1).runCollect)
// Right(Chunk(Zionomicon,12))

The missing file comes back as Left(java.nio.file.NoSuchFileException: ...), and the existing file reads as before. Here release is left out, so the default closes the reader.

Reusing and Composing Scope Resources

A Resource from the scope module describes how to acquire and release a value, independently of any stream. Stream.fromResource runs one in a fresh Scope for each run of the stream, passes the acquired value to the function, and closes the scope when the run ends. Resource.acquireRelease builds a Resource from an acquire block and a release function:

import zio.blocks.scope.Resource
import java.io.BufferedWriter

def openReader(path: Path): Resource[BufferedReader] =
  Resource.acquireRelease {
    println(s"opened ${path.getFileName}")
    Files.newBufferedReader(path)
  } { reader =>
    reader.close()
    println(s"closed ${path.getFileName}")
  }

def openWriter(path: Path): Resource[BufferedWriter] =
  Resource.acquireRelease {
    println(s"opened ${path.getFileName}")
    Files.newBufferedWriter(path)
  } { writer =>
    writer.close()
    println(s"closed ${path.getFileName}")
  }

Reading the inventory through openReader behaves like inventoryRows:

println(
  Stream
    .fromResource(openReader(inventory))(reader => Stream.fromIterator(reader.lines().iterator().asScala))
    .take(1)
    .runCollect
)
// opened inventory.csv
// closed inventory.csv
// Right(Chunk(Zionomicon,12))

The difference shows when one stream needs several resources. Resource#flatMap acquires the second resource after the first, so a for comprehension describes both. Here the stream reads the inventory and writes every title with fewer than five copies to a restock list:

val restock: Path = shop.resolve("restock.csv")

val inventoryAndRestock: Resource[(BufferedReader, BufferedWriter)] =
  for {
    reader <- openReader(inventory)
    writer <- openWriter(restock)
  } yield (reader, writer)

Running the stream with Stream#runDrain writes the list, and the output shows the order in which the two files open and close:

println(
  Stream
    .fromResource(inventoryAndRestock) { case (reader, writer) =>
      Stream
        .fromIterator(reader.lines().iterator().asScala)
        .filter(row => row.split(",")(1).toInt < 5)
        .tapEach { row =>
          writer.write(row)
          writer.newLine()
        }
    }
    .runDrain
)
// opened inventory.csv
// opened restock.csv
// closed restock.csv
// closed inventory.csv
// Right(())

print(Files.readString(restock))
// Functional Programming in Scala,3
// Scala for the Impatient,0
// Hands-on Scala Programming,4

The writer opens last and closes first: the scope releases resources in the reverse order of acquisition, so the restock file is flushed and closed before the inventory it was read from. Resource#zip combines two resources with the same order.

The two constructors differ in who describes the resource:

Situation	Use
One resource, opened and closed right where the stream is built	`Stream.fromAcquireRelease`
A `Resource` value that already exists, for example one shared with code that uses `Scope`	`Stream.fromResource`
Several resources that must open in order and close in reverse order	`Stream.fromResource`

Managed and Unmanaged Handles

The streams module also builds streams from handles you have already opened: Stream.fromInputStream and NioStreams.fromChannel are managed, so the stream closes the handle when the run ends. Stream.fromInputStreamUnmanaged and NioStreams.fromChannelUnmanaged read the same way but never close it; the caller owns the handle. All four emit Byte elements and report read failures as a typed IOException.

Unlike acquire in Stream.fromAcquireRelease, the handle is an ordinary argument, so it is open from the moment the stream is built. A managed stream closes it after the first run, and a second run reads a closed handle. The helper text turns the collected bytes back into a String:

def text(bytes: Chunk[Byte]): String = new String(bytes.toArray, UTF_8)

val managed: Stream[java.io.IOException, Byte] =
  Stream.fromInputStream(Files.newInputStream(inventory))
// managed: Stream[IOException, Byte] = Stream.fromAcquireRelease(...)

println(managed.take(10).runCollect.map(text))
// Right(Zionomicon)
println(managed.take(10).runCollect.map(text))
// Left(java.nio.channels.ClosedChannelException)

The second run fails with ClosedChannelException, because the first run closes the handle from Files.newInputStream. A managed stream built from a handle is good for one run, and if it never runs, the handle is never closed. To get a stream that opens a fresh handle on every run, open the handle in acquire and read it with the unmanaged constructor, so that only Stream.fromAcquireRelease closes it:

val rerunnable: Stream[java.io.IOException, Byte] =
  Stream.fromAcquireRelease(Files.newInputStream(inventory))(in => Stream.fromInputStreamUnmanaged(in))
// rerunnable: Stream[IOException, Byte] = Stream.fromAcquireRelease(...)

println(rerunnable.take(10).runCollect.map(text))
// Right(Zionomicon)
println(rerunnable.take(10).runCollect.map(text))
// Right(Zionomicon)

The unmanaged constructors are right when the stream borrows a handle that something else owns: System.in, a socket that the program keeps using after the stream ends, or a channel managed by a surrounding Using or Scope. Here the caller opens a FileChannel and keeps it:

import java.nio.channels.FileChannel

val channel: FileChannel = FileChannel.open(inventory)

Reading ten bytes through the unmanaged constructor leaves the channel open:

println(NioStreams.fromChannelUnmanaged(channel).take(10).runCollect.map(text))
// Right(Zionomicon)
println(channel.isOpen)
// true

channel.close()

NioStreams.fromChannel is the managed counterpart: given the same channel, it closes the channel at the end of the run.

Constructor	Who closes the handle	Running the stream twice
`Stream.fromInputStream(is)`	The stream, after each run	The second run reads a closed handle
`Stream.fromInputStreamUnmanaged(is)`	The caller	Both runs read the same open handle
`NioStreams.fromChannel(ch)`	The stream, after each run	The second run reads a closed handle
`NioStreams.fromChannelUnmanaged(ch)`	The caller	Both runs read the same open handle
`Stream.fromAcquireRelease(open)(unmanaged)`	The stream, after each run	Each run opens a new handle

Stream.fromJavaReader and Stream.fromJavaReaderUnmanaged follow the same rule for a java.io.Reader of characters.

Copying a File Through a Sink

On the writing side, NioSinks.fromChannel writes every byte of a stream to a WritableByteChannel and does not close it; Sink.fromOutputStream does the same for an OutputStream. The stream owns the input, so the code that runs the stream owns the output. Here scala.util.Using.resource closes the backup file even if the copy throws:

import java.nio.file.StandardOpenOption.{CREATE, TRUNCATE_EXISTING, WRITE}
import scala.util.Using

val backup: Path = shop.resolve("inventory-backup.csv")
// backup: Path = /tmp/bookshop/inventory-backup.csv

val copied: Either[java.io.IOException, Unit] =
  Using.resource(FileChannel.open(backup, CREATE, TRUNCATE_EXISTING, WRITE)) { out =>
    NioStreams.fromChannel(FileChannel.open(inventory)).run(NioSinks.fromChannel(out))
  }
// copied: Either[IOException, Unit] = Right(())

println(Files.readString(backup) == Files.readString(inventory))
// true

Stream#run returns the sink's result in a Right, here Right(()), or the first IOException in a Left. The input channel is opened inside the Using block and run at once, so the managed NioStreams.fromChannel closes it after this single run.

Reading from a Socket

A SocketChannel is a ReadableByteChannel, so the NIO constructors read sockets the same way they read files. To keep the example self-contained, a small server on the loopback address sends the inventory to every client and hangs up:

import java.net.{InetAddress, InetSocketAddress, ServerSocket}
import java.nio.channels.SocketChannel

val server = new ServerSocket(0, 50, InetAddress.getLoopbackAddress)
val serverThread = new Thread(() =>
  for (_ <- 1 to 2) {
    val client = server.accept()
    try client.getOutputStream.write(Files.readAllBytes(inventory))
    catch { case _: java.io.IOException => () }
    finally client.close()
  }
)
serverThread.start()

val address = new InetSocketAddress(InetAddress.getLoopbackAddress, server.getLocalPort)

NioStreams.fromChannel(SocketChannel.open(address)) would connect as soon as the stream is built. Connecting in acquire instead connects on each run and disconnects when the run ends:

val inventoryFeed: Stream[java.io.IOException, Byte] =
  Stream.fromAcquireRelease(
    acquire = {
      val socket = SocketChannel.open(address)
      println("connected")
      socket
    },
    release = (socket: SocketChannel) => {
      socket.close()
      println("disconnected")
    }
  )(socket => NioStreams.fromChannelUnmanaged(socket))

The first run reads until the server hangs up, and the second stops after ten bytes:

print(inventoryFeed.runCollect.map(text).getOrElse(""))
// connected
// disconnected
// Zionomicon,12
// Programming in Scala,7
// Functional Programming in Scala,3
// Scala for the Impatient,0
// Hands-on Scala Programming,4
println(inventoryFeed.take(10).runCollect.map(text))
// connected
// disconnected
// Right(Zionomicon)

Both runs disconnect, including the one that stops long before the server is done. We stop the server once both clients are served:

serverThread.join()
server.close()

Cleanup That Is Not a Resource

Some cleanup has no handle to close: deleting a temporary file, releasing a lock, or logging that a job is done. Stream#ensuring runs a block exactly once when a run of the stream ends, after completion, failure or early termination. Here the stream reads the backup made earlier and deletes it when it is done:

val fromBackup: Stream[Nothing, String] =
  inventoryRows(backup).ensuring {
    Files.deleteIfExists(backup)
    println("deleted inventory-backup.csv")
  }
// fromBackup: Stream[Nothing, String] = Stream.fromAcquireRelease(...).ensuring(...)

println(fromBackup.take(1).runCollect)
// opened inventory-backup.csv
// closed inventory-backup.csv
// deleted inventory-backup.csv
// Right(Chunk(Zionomicon,12))
println(Files.exists(backup))
// false

The reader closes first, because it belongs to the inner stream, and the finalizer runs after it. An exception thrown by the finalizer is a defect. Stream#ensuringAsync takes an Async[Unit] finalizer for asynchronous runs.

Pulling Elements by Hand

Sometimes code needs to read a stream piece by piece instead of handing it to a terminal, for example to read a header first and decide what to do with the rest. On the JVM, Stream#start opens the stream as a Reader that belongs to a Scope, and closing the scope closes the reader and every resource behind it:

import zio.blocks.scope.Scope

println(
  Scope.global.scoped { implicit scope =>
    import scope._
    val reader = inventoryRows(inventory).start
    $(reader)(r => r.readN[String](2).mkString(" | "))
  }
)
// opened inventory.csv
// closed inventory.csv
// Zionomicon,12 | Programming in Scala,7

closed inventory.csv is printed when the scoped block ends, not when Reader#readN returns. Scope#$ is how a scope lets you use a value it owns; the Scope tutorial explains why the reader cannot leave the block.

Putting It Together

The complete example is a nightly restock job. It downloads the inventory export from a loopback server into a staging file, reads the staging file and writes a restock list through two composed Resource values, fails with a typed error on a damaged row, and deletes the staging file with Stream#ensuring:

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

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

import java.io.{BufferedReader, BufferedWriter, IOException}
import java.net.{InetAddress, InetSocketAddress, ServerSocket}
import java.nio.channels.{FileChannel, SocketChannel}
import java.nio.charset.StandardCharsets.UTF_8
import java.nio.file.StandardOpenOption.{CREATE, TRUNCATE_EXISTING, WRITE}
import java.nio.file.{Files, Path}
import scala.jdk.CollectionConverters._
import scala.util.Using

/**
 * How to Stream Files and Sockets Without Leaking Resources
 *
 * Nightly restock job: downloads the inventory export from a loopback server
 * into a staging file, reads it and writes a restock list through two composed
 * `Resource` values, fails with a typed error on a damaged row, and deletes the
 * staging file with `Stream#ensuring`.
 *
 * Run with: sbt "runMain streamfilesandsockets.CompleteExample"
 */
object CompleteExample {

  final case class Stock(title: String, copies: Int)

  final case class BadRow(row: String)

  sealed trait RestockError
  final case class DownloadFailed(cause: IOException) extends RestockError
  final case class DamagedExport(row: BadRow)         extends RestockError

  val cleanExport: String =
    """Zionomicon,12
      |Programming in Scala,7
      |Functional Programming in Scala,3
      |Scala for the Impatient,0
      |Hands-on Scala Programming,4
      |""".stripMargin

  val damagedExport: String =
    """Zionomicon,12
      |Programming in Scala,seven
      |Functional Programming in Scala,3
      |""".stripMargin

  def parse(row: String): Stream[BadRow, Stock] = {
    val fields = row.split(",")
    fields(1).toIntOption match {
      case Some(copies) => Stream.succeed(Stock(fields(0), copies))
      case None         => Stream.fail(BadRow(row))
    }
  }

  def openReader(path: Path): Resource[BufferedReader] =
    Resource.acquireRelease {
      println(s"opened ${path.getFileName}")
      Files.newBufferedReader(path)
    } { reader =>
      reader.close()
      println(s"closed ${path.getFileName}")
    }

  def openWriter(path: Path): Resource[BufferedWriter] =
    Resource.acquireRelease {
      println(s"opened ${path.getFileName}")
      Files.newBufferedWriter(path)
    } { writer =>
      writer.close()
      println(s"closed ${path.getFileName}")
    }

  def connect(address: InetSocketAddress): Stream[IOException, Byte] =
    Stream.fromAcquireRelease(
      acquire = {
        val socket = SocketChannel.open(address)
        println("connected")
        socket
      },
      release = (socket: SocketChannel) => {
        socket.close()
        println("disconnected")
      }
    )(socket => NioStreams.fromChannelUnmanaged(socket))

  def download(address: InetSocketAddress, staging: Path): Either[RestockError, Unit] =
    Using
      .resource(FileChannel.open(staging, CREATE, TRUNCATE_EXISTING, WRITE)) { out =>
        connect(address).run(NioSinks.fromChannel(out))
      }
      .left
      .map(DownloadFailed(_))

  def restock(staging: Path, restockList: Path): Either[RestockError, Unit] = {
    val staged: Resource[(BufferedReader, BufferedWriter)] =
      for {
        reader <- openReader(staging)
        writer <- openWriter(restockList)
      } yield (reader, writer)

    Stream
      .fromResource(staged) { case (reader, writer) =>
        Stream
          .fromIterator(reader.lines().iterator().asScala)
          .flatMap(parse)
          .filter(_.copies < 5)
          .tapEach { stock =>
            writer.write(stock.title)
            writer.newLine()
          }
      }
      .ensuring {
        Files.deleteIfExists(staging)
        println(s"deleted ${staging.getFileName}")
      }
      .mapError(DamagedExport(_))
      .runDrain
  }

  def restockJob(address: InetSocketAddress, dir: Path): Either[RestockError, Unit] = {
    val staging     = dir.resolve("inventory-staging.csv")
    val restockList = dir.resolve("restock.csv")

    val result = download(address, staging).flatMap(_ => restock(staging, restockList))

    println(result)
    print(Files.readString(restockList))
    println(s"staging exists: ${Files.exists(staging)}")
    result
  }

  def main(args: Array[String]): Unit = {
    val dir = Files.createTempDirectory("restock")

    val server       = new ServerSocket(0, 50, InetAddress.getLoopbackAddress)
    val serverThread = new Thread(() =>
      for (payload <- List(cleanExport, damagedExport)) {
        val client = server.accept()
        try client.getOutputStream.write(payload.getBytes(UTF_8))
        catch { case _: IOException => () }
        finally client.close()
      }
    )
    serverThread.start()

    val address = new InetSocketAddress(InetAddress.getLoopbackAddress, server.getLocalPort)

    println("--- clean export ---")
    restockJob(address, dir)

    println("--- damaged export ---")
    restockJob(address, dir)

    serverThread.join()
    server.close()
  }
}

Running the Examples

All examples in this guide have runnable Scala files in the streams-examples project. Clone the repository and change into the examples project:

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

The examples share the sample inventory files, which live in one file:

streams-examples/src/main/scala/streamfilesandsockets/SampleFiles.scala

streams-examples/src/main/scala/streamfilesandsockets/SampleFiles.scala
package streamfilesandsockets

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

import java.io.BufferedReader
import java.nio.charset.StandardCharsets.UTF_8
import java.nio.file.{Files, Path}
import scala.jdk.CollectionConverters._

final case class Stock(title: String, copies: Int)

final case class BadRow(row: String)

/**
 * Shared sample data and helpers for the "How to Stream Files and Sockets
 * Without Leaking Resources" guide.
 */
object SampleFiles {

  /** Creates a temporary directory for one example run. */
  def newShop(): Path = Files.createTempDirectory("bookshop")

  /** Writes the clean inventory export into `shop` and returns its path. */
  def writeInventory(shop: Path): Path = {
    val inventory: Path = shop.resolve("inventory.csv")
    Files.writeString(
      inventory,
      """Zionomicon,12
        |Programming in Scala,7
        |Functional Programming in Scala,3
        |Scala for the Impatient,0
        |Hands-on Scala Programming,4
        |""".stripMargin
    )
    inventory
  }

  /** Writes the damaged inventory export into `shop` and returns its path. */
  def writeDamaged(shop: Path): Path = {
    val damaged: Path = shop.resolve("inventory-damaged.csv")
    Files.writeString(
      damaged,
      """Zionomicon,12
        |Programming in Scala,seven
        |Functional Programming in Scala,3
        |""".stripMargin
    )
    damaged
  }

  def inventoryRows(path: Path): Stream[Nothing, String] =
    Stream.fromAcquireRelease(
      acquire = {
        println(s"opened ${path.getFileName}")
        Files.newBufferedReader(path)
      },
      release = (reader: BufferedReader) => {
        reader.close()
        println(s"closed ${path.getFileName}")
      }
    )(reader => Stream.fromIterator(reader.lines().iterator().asScala))

  def parse(row: String): Stream[BadRow, Stock] = {
    val fields = row.split(",")
    fields(1).toIntOption match {
      case Some(copies) => Stream.succeed(Stock(fields(0), copies))
      case None         => Stream.fail(BadRow(row))
    }
  }

  def text(bytes: Chunk[Byte]): String = new String(bytes.toArray, UTF_8)
}

Each example is an object with a def main method rather than extends App: on the JVM, a blocking terminal that runs inside an object initializer can deadlock once Stream#catchAll is in the pipeline.

Read a File and Stop Early

This example reads the inventory with Stream.fromAcquireRelease, then reads only the first two rows, and prints when the file opens and closes.

streams-examples/src/main/scala/streamfilesandsockets/Step1ReadInventory.scala

streams-examples/src/main/scala/streamfilesandsockets/Step1ReadInventory.scala
package streamfilesandsockets

import java.nio.file.Path

import SampleFiles._

/**
 * How to Stream Files and Sockets Without Leaking Resources
 *
 * Read a File and Stop Early: `Stream.fromAcquireRelease` closes the file after
 * the last row and after an early `take`.
 *
 * Run with: sbt "runMain streamfilesandsockets.Step1ReadInventory"
 */
object Step1ReadInventory {
  def main(args: Array[String]): Unit = {
    val inventory: Path = writeInventory(newShop())

    println(inventoryRows(inventory).runForeach(row => println(row)))

    println(inventoryRows(inventory).take(2).runForeach(row => println(row)))
  }
}