arrow-fx-coroutines / arrow.fx.coroutines.stream / ResourceOps / <init>

<init>

ResourceOps(s: Stream<O>)

DSL boundary to access terminal operators as a Resource Allows for consume a Stream as a Resource, meaning the root scope of the Stream remains open until Resource.use returns.

This allows for safe consumption of streaming resources in terminal operators, and inside the Resource.use combinator.

import arrow.fx.coroutines.stream.*
import arrow.fx.coroutines.Atomic

class Logger {
  private val state = Atomic.unsafe(emptyList<String>())

  suspend fun log(msg: String): Unit =
    state.update { it + msg }

  suspend fun dumpLog(): Unit =
    println(state.get())
}

fun openFileWithName(name: String): String =
  "File($name)"

//sampleStart
suspend fun main(): Unit {
  val logger = Logger()

  Stream.bracket({ openFileWithName("a") }, { name -> logger.log("finalizing: $name") })
    .append { Stream.bracket({ openFileWithName("b") }, { name -> logger.log("finalizing: $name") }) }
    .asResource()
    .lastOrError()
    .use { last -> logger.log("Using $last") }

  logger.dumpLog() // [finalizing: File(a), Using File(b), finalizing: File(b)]
}
//sampleEnd

As you can see here, we can use the last streamed File before it gets closed by the Stream. Since we consumed the Stream as asResource().lastOrError(), this extends the last scope to the returned Resource, so we can safely use it and the Stream still properly closes all resources opened with bracket.