Improve Arrow Fx Coroutines docs for release

arrow-fx-coroutines/README.MD

@@ -208,9 +208,11 @@ tailrec suspend fun sleeper(): Unit {
 This also means that our new sleep can back-pressure `timeOutOrNull`.
 
 ```kotlin:ank
-suspend main(): Unit {
+import arrow.fx.coroutines.*
+
+suspend fun main(): Unit {
  val r = timeOutOrNull(1.seconds) {
- uncancelable { sleep(2.seconds) }
+ uncancellable { sleep(2.seconds) }
  } // r is null, but took 2 seconds.
 }
 ```
@@ -306,7 +308,7 @@ val resources: List<Resource<File>> =
 val resource: Resource<List<File>> =
  resources.sequence(Resource.applicative())
 
-suspend main(): Unit {
+suspend fun main(): Unit {
  resource.use { files ->
  files.parTraverse(IODispatchers.IOPool) { file ->
  file.toString()
@@ -328,7 +330,7 @@ Simple constructs like `suspend fun Either.catch(f: () -> A): Either<Throwable,
 A simple example might be to repeat an action `n` times, similar to the `repeat` function in the standard library.
 
 ```kotlin:ank
-suspend main(): Unit {
+suspend fun main(): Unit {
  repeat(Schedule.recurs<A>(n)) {
  println("Hello")
  }
@@ -338,7 +340,7 @@ suspend main(): Unit {
 Alternatively we can re-use this `Schedule` to `retry` a `suspend fun` `n` times when it fails.
 
 ```kotlin:ank
-suspend main(): Unit {
+suspend fun main(): Unit {
  retry(Schedule.recurs<A>(n)) {
  println("I am going to do nothing but throw a tantrum!")
  throw RuntimeException("Boom!")
@@ -353,7 +355,7 @@ fun <A> schedule(): Schedule<A, List<A>> = Schedule {
  (recurs<A>(10) and spaced(10.seconds)) zipRight collect()
 }
 
-suspend main(): Unit {
+suspend fun main(): Unit {
  var count = Atomic(0)
 
  val history: List<Int> = repeat(schedule<Int>()) {

arrow-fx-coroutines/src/main/kotlin/arrow/fx/coroutines/Bracket.kt

@@ -56,6 +56,17 @@ suspend fun <A> uncancellable(f: suspend () -> A): A =
  }
  }
 
+/**
+ * Registers an [onCancel] handler after [fa].
+ * [onCancel] is guaranteed to be called in case of cancellation, otherwise it's ignored.
+ *
+ * Useful for wiring cancellation tokens between fibers, building inter-op with other effect systems or testing.
+ *
+ * @param fa program that you want to register handler on
+ * @param onCancel handler to run when [fa] gets cancelled.
+ * @see guarantee for registering a handler that is guaranteed to always run.
+ * @see guaranteeCase for registering a handler that executes for any [ExitCase].
+ */
 suspend fun <A> onCancel(
  fa: suspend () -> A,
  onCancel: suspend () -> Unit
@@ -66,22 +77,149 @@ suspend fun <A> onCancel(
  }
 }
 
+/**
+ * Guarantees execution of a given [finalizer] after [fa] regardless of success, error or cancellation.
+ *
+ * As best practice, it's not a good idea to release resources via [guarantee].
+ * since [guarantee] doesn't properly model acquiring, using and releasing resources.
+ * It only models scheduling of a finalizer after a given suspending program,
+ * so you should prefer [Resource] or [bracket] which captures acquiring,
+ * using and releasing into 3 separate steps to ensure resource safety.
+ *
+ * @param fa program that you want to register handler on
+ * @param finalizer handler to run after [fa].
+ * @see guaranteeCase for registering a handler that tracks the [ExitCase] of [fa].
+ */
 suspend fun <A> guarantee(
  fa: suspend () -> A,
- release: suspend () -> Unit
-): A = guaranteeCase(fa) { release.invoke() }
+ finalizer: suspend () -> Unit
+): A = guaranteeCase(fa) { finalizer.invoke() }
 
+/**
+ * Guarantees execution of a given [finalizer] after [fa] regardless of success, error or cancellation., allowing
+ * for differentiating between exit conditions with to the [ExitCase] argument of the finalizer.
+ *
+ * As best practice, it's not a good idea to release resources via [guaranteeCase].
+ * since [guaranteeCase] doesn't properly model acquiring, using and releasing resources.
+ * It only models scheduling of a finalizer after a given suspending program,
+ * so you should prefer [Resource] or [bracketCase] which captures acquiring,
+ * using and releasing into 3 separate steps to ensure resource safety.
+ *
+ * @param fa program that you want to register handler on
+ * @param finalizer handler to run after [fa].
+ * @see guarantee for registering a handler that ignores the [ExitCase] of [fa].
+ */
 suspend fun <A> guaranteeCase(
  fa: suspend () -> A,
- release: suspend (ExitCase) -> Unit
-): A = bracketCase({ Unit }, { fa.invoke() }, { _, ex -> release(ex) })
+ finalizer: suspend (ExitCase) -> Unit
+): A = bracketCase({ Unit }, { fa.invoke() }, { _, ex -> finalizer(ex) })
 
+/**
+ * Meant for specifying tasks with safe resource acquisition and release in the face of errors and interruption.
+ * It would be the equivalent of an async capable `try/catch/finally` statements in mainstream imperative languages for resource
+ * acquisition and release.
+ *
+ * @param acquire the action to acquire the resource
+ *
+ * @param use is the action to consume the resource and produce a result.
+ * Once the resulting suspend program terminates, either successfully, error or disposed,
+ * the [release] function will run to clean up the resources.
+ *
+ * @param release is the action that's supposed to release the allocated resource after `use` is done, irregardless
+ * of its exit condition.
+ *
+ * ```kotlin:ank:playground
+ * import arrow.fx.IO
+ *
+ * class File(url: String) {
+ * fun open(): File = this
+ * fun close(): Unit {}
+ * override fun toString(): String = "This file contains some interesting content!"
+ * }
+ *
+ * suspend fun openFile(uri: String): File = File(uri).open()
+ * suspend fun closeFile(file: File): Unit = file.close()
+ * suspend fun fileToString(file: File): String = file.toString()
+ *
+ * suspend fun main(): Unit {
+ * //sampleStart
+ * val res = bracketCase(
+ * acquire = { openFile("data.json") },
+ * use = { file -> fileToString(file) }),
+ * release = { file: File -> closeFile(file) }
+ * )
+ * //sampleEnd
+ * println(res)
+ * }
+ * ```
+ */
 suspend fun <A, B> bracket(
  acquire: suspend () -> A,
  use: suspend (A) -> B,
  release: suspend (A) -> Unit
 ): B = bracketCase(acquire, use, { a, _ -> release(a) })
 
+/**
+ * A way to safely acquire a resource and release in the face of errors and cancellation.
+ * It uses [ExitCase] to distinguish between different exit cases when releasing the acquired resource.
+ *
+ * [bracketCase] exists out of a three stages:
+ * 1. acquisition
+ * 2. consumption
+ * 3. releasing
+ *
+ * 1. Resource acquisition is **NON CANCELLABLE**.
+ * If resource acquisition fails, meaning no resource was actually successfully acquired then we short-circuit the effect.
+ * Reason being, we cannot [release] what we did not `acquire` first. Same reason we cannot call [use].
+ * If it is successful we pass the result to stage 2 [use].
+ *
+ * 2. Resource consumption is like any other `suspend` effect. The key difference here is that it's wired in such a way that
+ * [release] **will always** be called either on [ExitCase.Cancelled], [ExitCase.Failure] or [ExitCase.Completed].
+ * If it failed than the resulting [suspend] from [bracketCase] will be the error, otherwise the result of [use].
+ *
+ * 3. Resource releasing is **NON CANCELLABLE**, otherwise it could result in leaks.
+ * In the case it throws the resulting [suspend] will be either the error or a composed error if one occurred in the [use] stage.
+ *
+ * @param acquire the action to acquire the resource
+ *
+ * @param use is the action to consume the resource and produce a result.
+ * Once the resulting suspend program terminates, either successfully, error or disposed,
+ * the [release] function will run to clean up the resources.
+ *
+ * @param release the allocated resource after [use] terminates.
+ *
+ * ```kotlin:ank:playground
+ * import arrow.fx.coroutines.*
+ *
+ * class File(url: String) {
+ * fun open(): File = this
+ * fun close(): Unit {}
+ * }
+ *
+ * suspend fun File.content(): String =
+ * "This file contains some interesting content!"
+ * suspend fun openFile(uri: String): File = File(uri).open()
+ * suspend fun closeFile(file: File): Unit = file.close()
+ *
+ * suspend fun main(): Unit {
+ * //sampleStart
+ * val res = bracketCase(
+ * acquire = { openFile("data.json") },
+ * use = { file -> file.content() },
+ * release = { file, exitCase ->
+ * when (exitCase) {
+ * is ExitCase.Completed -> println("File closed with $exitCase")
+ * is ExitCase.Cancelled -> println("Program cancelled with $exitCase")
+ * is ExitCase.Failure -> println("Program failed with $exitCase")
+ * }
+ * closeFile(file)
+ * }
+ * )
+ * //sampleEnd
+ * println(res)
+ * }
+ * ```
+ */
 suspend fun <A, B> bracketCase(
  acquire: suspend () -> A,
  use: suspend (A) -> B,

arrow-fx-coroutines/src/main/kotlin/arrow/fx/coroutines/ConcurrentVar.kt

@@ -6,7 +6,7 @@ import kotlinx.atomicfu.atomic
 
 /**
  * [ConcurrentVar] is a mutable concurrent safe variable which is either `empty` or contains a `single value` of type [A].
- *
+ * It behaves the same as a single element [arrow.fx.coroutines.stream.concurrent.Queue].
  * When trying to [put] or [take], it'll suspend when it's respectively [isEmpty] or [isNotEmpty].
  *
  * There are also operators that return immediately, [tryTake] & [tryPut],
@@ -18,19 +18,12 @@ import kotlinx.atomicfu.atomic
  * ```kotlin:ank:playground
  * import arrow.fx.coroutines.*
  *
- * suspend fun fibonacci(n: Int, prev: Int = 0, next: Int = 1): Int =
- * when (n) {
- * 0 -> prev
- * 1 -> next
- * else -> fibonacci(n - 1, next, prev + next)
- * }
- *
  * suspend fun main(): Unit {
  * val mvar = ConcurrentVar.empty<Int>()
  *
  * ForkConnected {
- * val asyncFib = fibonacci(50)
- * mvar.put(asyncFib)
+ * sleep(3.seconds)
+ * mvar.put(5)
  * }
  *
  * val r = mvar.take() // suspend until Fork puts result in MVar

arrow-fx-coroutines/src/main/kotlin/arrow/fx/coroutines/Fiber.kt

@@ -35,8 +35,24 @@ internal fun <A> Fiber(promise: UnsafePromise<A>, conn: SuspendConnection): Fibe
 
 /**
  * Launches a new suspendable cancellable coroutine within a [Fiber].
- * It does so by connecting the created [Fiber]'s cancellation to the parent.
- * If the parent gets cancelled, then this [Fiber] will also get cancelled.
+ * It does so by connecting the created [Fiber]'s cancellation to the callers `suspend` scope.
+ * If the caller of `ForkConnected` gets cancelled, then this [Fiber] will also get cancelled.
+ *
+ * ```kotlin:ank:playground
+ * import arrow.fx.coroutines.*
+ *
+ * suspend fun main(): Unit {
+ * val parent = ForkConnected {
+ * ForkConnected { // cancellation connected to parent
+ * onCancel({ never<Unit>() }) {
+ * println("I got cancelled by my parent")
+ * }
+ * }
+ * }
+ * sleep(1.seconds)
+ * parent.cancel()
+ * }
+ * ```
  *
  * You can [Fiber.join] or [Fiber.cancel] the computation.
  * Cancelling this [Fiber] **will not** cancel its parent.
@@ -53,6 +69,7 @@ suspend fun <A> ForkConnected(ctx: CoroutineContext = ComputationPool, f: suspen
  Fiber(promise, conn2)
  }
 
+/** @see ForkConnected **/
 suspend fun <A> (suspend () -> A).forkConnected(ctx: CoroutineContext = ComputationPool): Fiber<A> =
  ForkConnected(ctx, this)
 
@@ -107,6 +124,7 @@ suspend fun <A> ForkScoped(
  Fiber(promise, conn2)
 }
 
+/** @see ForkScoped */
 suspend fun <A> (suspend () -> A).forkScoped(
  ctx: CoroutineContext = ComputationPool,
  interruptWhen: suspend () -> Unit
@@ -117,17 +135,18 @@ suspend fun <A> (suspend () -> A).forkScoped(
  * You can [Fiber.join] or [Fiber.cancel] the computation.
  *
  * **BEWARE** you immediately leak the [Fiber] when launching without connection control.
- * Use [ForkConnected] or safely launch the fiber as a [Resource] or using [Fiber].
+ * Use [ForkConnected] or safely launch the fiber as a [Resource] or using [bracketCase].
  *
  * @see ForkConnected for a fork operation that wires cancellation to its parent in a safe way.
  */
+suspend fun <A> ForkAndForget(ctx: CoroutineContext = ComputationPool, f: suspend () -> A): Fiber<A> =
+ f.forkAndForget(ctx)
+
+/** @see ForkAndForget */
 suspend fun <A> (suspend () -> A).forkAndForget(ctx: CoroutineContext = ComputationPool): Fiber<A> {
  val promise = UnsafePromise<A>()
  // A new SuspendConnection, because its cancellation is now decoupled from our current one.
  val conn = SuspendConnection()
  startCoroutineCancellable(CancellableContinuation(ctx, conn, promise::complete))
  return Fiber(promise, conn)
 }
-
-suspend fun <A> ForkAndForget(ctx: CoroutineContext = ComputationPool, f: suspend () -> A): Fiber<A> =
- f.forkAndForget(ctx)

arrow-fx-coroutines/src/main/kotlin/arrow/fx/coroutines/ParTraverse.kt

@@ -84,6 +84,23 @@ suspend fun <A, B> Iterable<A>.parTraverse(f: suspend (A) -> B): List<B> =
  * Traverse this [Iterable] and and run all mappers [f] on [CoroutineContext].
  * Cancelling this operation cancels all running tasks.
  *
+ * ```kotlin:ank:playground
+ * import arrow.fx.coroutines.*
+ *
+ * data class User(val id: Int)
+ *
+ * suspend fun main(): Unit {
+ * //sampleStart
+ * suspend fun getUserById(id: Int): User =
+ * User(id)
+ *
+ * val res = listOf(1, 2, 3)
+ * .parTraverse(ComputationPool, ::getUserById)
+ * //sampleEnd
+ * println(res)
+ * }
+ * ```
+ *
  * **WARNING** it runs in parallel depending on the capabilities of the provided [CoroutineContext].
  * We ensure they start in sequence so it's guaranteed to finish on a single threaded context.
  *

arrow-fx-coroutines/src/main/kotlin/arrow/fx/coroutines/ParTupledN.kt

@@ -13,6 +13,27 @@ import kotlin.coroutines.intrinsics.COROUTINE_SUSPENDED
  * Parallel maps [fa], [fb] in parallel on [ComputationPool].
  * Cancelling this operation cancels both operations running in parallel.
  *
+ * ```kotlin:ank:playground
+ * import arrow.fx.coroutines.*
+ *
+ * suspend fun main(): Unit {
+ * //sampleStart
+ * val result = parMapN(
+ * { "First one is on ${Thread.currentThread().name}" },
+ * { "Second one is on ${Thread.currentThread().name}" }
+ * ) { (a, b) ->
+ * "$a\n$b"
+ * }
+ * //sampleEnd
+ * println(result)
+ * }
+ * ```
+ *
+ * @param fa value to parallel map
+ * @param fb value to parallel map
+ * @param f function to map/combine value [A] and [B]
+ * ```
+ *
  * @see parMapN for the same function that can race on any [CoroutineContext].
  */
 suspend fun <A, B, C> parMapN(fa: suspend () -> A, fb: suspend () -> B, f: (Pair<A, B>) -> C): C =