Remove TerminalOps from Stream (#260)

Author: nomisRev

arrow-libs/fx/arrow-fx-coroutines/src/main/kotlin/arrow/fx/coroutines/stream/ParJoin.kt

@@ -74,7 +74,6 @@ internal suspend fun <O> runInner(
                 outputQ.enqueue1(Some(s))
               }
               .interruptWhen(done.map { it.isDefined() }) // must be AFTER enqueue to the sync queue, otherwise the process may hang to enq last item while being interrupted
-              .compile()
               .drain()
           }.swap().orNull()
           val e2 = lease.cancel().swap().orNull()
@@ -104,7 +103,6 @@ internal suspend fun <O> Stream<Stream<O>>.runOuter(
         runInner(ctx, inner, done, outputQ, running, available, outerScope)
       }
     }.interruptWhen(done.map { it.isDefined() })
-      .compile()
       .drain()
   }
 
@@ -165,7 +163,6 @@ internal suspend fun signalResult(done: SignallingAtomic<Option<Option<Throwable
  *       }
  *     }
  *     .parJoin(10, IOPool)
- *     .compile()
  *     .drain()
  * //sampleEnd
  * ```
@@ -196,7 +193,6 @@ fun <O> Stream<Stream<O>>.parJoin(
       running.discrete() // Await everyone stop running
         .dropWhile { it > 0 }
         .take(1)
-        .compile()
         .drain()
 
       signalResult(done)

arrow-libs/fx/arrow-fx-coroutines/src/main/kotlin/arrow/fx/coroutines/stream/Pull.kt

@@ -590,19 +590,6 @@ fun <O> Pull<O, Unit>.firstOrNull(f: (O) -> Boolean): Pull<Nothing, PullUncons1<
     }
   }
 
-/** Returns the last element of the input, if non-empty. */
-fun <O> Pull<O, Unit>.lastOrNull(): Pull<Nothing, O?> {
-  fun go(prev: O?, s: Pull<O, Unit>): Pull<Nothing, O?> =
-    s.unconsOrNull().flatMap { uncons ->
-      when (uncons) {
-        null -> Pull.just(prev)
-        else -> go(uncons.head.lastOrNull() ?: prev, uncons.tail)
-      }
-    }
-
-  return go(null, this)
-}
-
 /** Writes a single `true` value if all input matches the predicate, `false` otherwise. */
 fun <O> Pull<O, Unit>.forall(p: (O) -> Boolean): Pull<Nothing, Boolean> =
   unconsOrNull().flatMap { uncons ->

arrow-libs/fx/arrow-fx-coroutines/src/main/kotlin/arrow/fx/coroutines/stream/ResourceTerminalOps.kt arrow-libs/fx/arrow-fx-coroutines/src/main/kotlin/arrow/fx/coroutines/stream/ResourceOps.kt

@@ -2,12 +2,55 @@ package arrow.fx.coroutines.stream
 
 import arrow.fx.coroutines.Resource
 
+/** Opens DSL to consume [Stream] as a [Resource]. */
+fun <O> Stream<O>.asResource(): ResourceOps<O> =
+  ResourceOps(this)
+
 /**
- * DSL boundary to access terminal operators as resources
+ * 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.
+ *
+ * ```kotlin:ank:playground
+ * 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
+ * ```
  *
- * Terminal operators consume the stream
- */ // TODO report inline results in Exception in thread "main" java.lang.VerifyError: Bad type on operand stack
-/* inline */ class ResourceTerminalOps<O>(private val s: Stream<O>) {
+ * 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`.
+ */
+class ResourceOps<O>(private val s: Stream<O>) {
 
   suspend fun toList(): Resource<List<O>> =
     compiler(mutableListOf()) { acc, ch -> acc.apply { addAll(ch.toList()) } }