Implement Either.orThrow

build.sbt

@@ -46,7 +46,7 @@ lazy val core: Project = (project in file("core"))
   .settings(
     name := "core",
     libraryDependencies ++= Seq(
-      "com.softwaremill.jox" % "channels" % "0.3.0",
+      "com.softwaremill.jox" % "channels" % "0.3.1",
       scalaTest
     ),
     // Check IO usage in core

core/src/main/scala/ox/either.scala

@@ -122,9 +122,18 @@ object either:
     transparent inline def ok(): A = f.join().ok()
 
   extension [E](e: E)
+    /** Fail the computation, short-circuiting to the enclosing [[either]] block. */
     transparent inline def fail(): Nothing =
       summonFrom {
         case given boundary.Label[Either[E, Nothing]] => break(Left(e))
         case given boundary.Label[Either[Nothing, Nothing]] =>
           error("The enclosing `either` call uses a different error type.\nIf it's explicitly typed, is the error type correct?")
       }
+
+  extension [E <: Throwable, T](e: Either[E, T])
+    /** Unwrap the right-value of the `Either`, throwing the contained exception if this is a lef-value. For a variant which allows
+      * unwrapping `Either`s, propagates errors and doesn't throw exceptions, see [[apply]].
+      */
+    def orThrow: T = e match
+      case Right(value)    => value
+      case Left(throwable) => throw throwable

core/src/test/scala/ox/EitherTest.scala

@@ -3,7 +3,7 @@ package ox
 import org.scalatest.exceptions.TestFailedException
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
-import ox.either.{fail, ok}
+import ox.either.{fail, ok, orThrow}
 
 import scala.util.boundary.Label
 
@@ -164,6 +164,16 @@ class EitherTest extends AnyFlatSpec with Matchers:
       """)
   }
 
+  "orThrow" should "unwrap the value for a Right-value" in {
+    val v: Either[Exception, Int] = Right(10)
+    v.orThrow shouldBe 10
+  }
+
+  it should "throw exceptions for a Left-value" in {
+    val v: Either[Exception, Int] = Left(new RuntimeException("boom!"))
+    intercept[RuntimeException](v.orThrow).getMessage shouldBe "boom!"
+  }
+
   private transparent inline def receivesNoEitherNestingError(inline code: String): Unit =
     val errs = scala.compiletime.testing.typeCheckErrors(code)
     if !errs

core/src/test/scala/ox/channels/SourceOpsFlattenTest.scala

@@ -58,14 +58,14 @@ class SourceOpsFlattenTest extends AnyFlatSpec with Matchers with OptionValues {
       source.send {
         val subSource = Channel.bufferedDefault[Int]
         subSource.send(20)
-        forkUnsupervised {
+        fork {
           lockA.await() // 30 won't be added until, lockA is released after 20 consumption
           subSource.send(30)
           subSource.done()
         }
         subSource
       }
-      forkUnsupervised {
+      fork {
         lockB.await() // 40 won't be added until, lockB is released after 30 consumption
         source.send(Source.fromValues(40))
         source.done()
@@ -86,25 +86,25 @@ class SourceOpsFlattenTest extends AnyFlatSpec with Matchers with OptionValues {
       val child1 = Channel.rendezvous[Int]
       val lock = CountDownLatch(1)
       fork {
-        child1.send(10)
-        lock.await() // wait for child2 to emit an error
-        child1.send(30) // `flatten` will not receive this, as it will be short-circuited by the error
+        lock.await() // wait for the error to be discovered
+        child1.send(10) // `flatten` will not receive this, as it will be short-circuited by the error
         child1.doneOrClosed()
       }
       val child2 = Channel.rendezvous[Int]
       fork {
-        child2.send(20)
         child2.error(new Exception("intentional failure"))
-        lock.countDown()
       }
       val source = Source.fromValues(child1, child2)
       val flattenSource = {
         implicit val capacity: StageCapacity = StageCapacity(0)
         source.flatten
       }
-      Set(flattenSource.receive(), flattenSource.receive()) shouldBe Set(10, 20)
+
       flattenSource.receiveOrClosed() should be(a[ChannelClosed.Error])
-      child1.receive() shouldBe 30
+
+      // no values should be piped by the flattening process after the error
+      lock.countDown()
+      child1.receive() shouldBe 10
       child1.receiveOrClosed() shouldBe ChannelClosed.Done
     }
   }
@@ -143,26 +143,26 @@ class SourceOpsFlattenTest extends AnyFlatSpec with Matchers with OptionValues {
   it should "stop pulling from the sources when the receiver is closed" in {
     val child1 = Channel.rendezvous[Int]
 
-    Thread.startVirtualThread(() => {
-      child1.send(10)
-      // at this point `flatten` channel is closed
-      // so although `flatten` thread receives "20" element
-      // it can not push it to its output channel and it will be lost
-      child1.send(20)
-      child1.send(30)
-      child1.done()
-    })
-
     supervised {
+      fork {
+        child1.send(10)
+        // at this point `flatten` channel is closed
+        // so although `flatten` thread receives "20" element
+        // it can not push it to its output channel and it will be lost
+        child1.send(20)
+        child1.send(30)
+        child1.done()
+      }
+
       val source = Source.fromValues(child1)
       val flattenSource = {
         implicit val capacity: StageCapacity = StageCapacity(0)
         source.flatten
       }
       flattenSource.receive() shouldBe 10
-    }
 
-    child1.receiveOrClosed() shouldBe 30
-    child1.receiveOrClosed() shouldBe ChannelClosed.Done
+      child1.receiveOrClosed() shouldBe 30
+      child1.receiveOrClosed() shouldBe ChannelClosed.Done
+    }
   }
 }

doc/basics/error-handling.md

@@ -189,3 +189,18 @@ val outerResult: Either[Exception, Unit] = either:
 ```
 
 After this change refactoring `returnsEither` to return `Either[Exception, Int]` would yield a compile error on `returnsEither.ok()`.
+
+## Other `Either` utilities
+
+For `Either` instances where the left-side is an exception, the right-value of an `Either` can be unwrapped using `.orThrow`.
+The exception on the left side is thrown if it is present:
+
+```scala mdoc:compile-only
+import ox.either.orThrow
+
+val v1: Either[Exception, Int] = Right(10)
+assert(v1.orThrow == 10)
+
+val v2: Either[Exception, Int] = Left(new RuntimeException("boom!"))
+v2.orThrow // throws RuntimeException("boom!")
+```

doc/channels/io.md

@@ -53,7 +53,7 @@ A `Source[Chunk[Byte]]` can be directed to write to an `OutputStream`:
 
 ```scala mdoc:compile-only
 import ox.channels.Source
-import ox.{Chunk, IO, supervised}
+import ox.{IO, supervised}
 import java.io.ByteArrayOutputStream
 
 val outputStream = new ByteArrayOutputStream()
@@ -97,7 +97,7 @@ A `Source[Chunk[Byte]]` can be written to a file under a given path:
 ```scala mdoc:compile-only
 
 import ox.channels.Source
-import ox.{Chunk, IO, supervised}
+import ox.{IO, supervised}
 import java.nio.file.Paths
 
 supervised {

doc/repeat.md

@@ -50,7 +50,7 @@ See [scheduled](scheduled.md) for details on how to create custom schedules.
 
 ```scala mdoc:compile-only
 import ox.UnionMode
-import ox.scheduling.{Jitter, Schedule, repeat, repeatEither, repeatWithErrorMode, RepeatConfig}
+import ox.scheduling.{Schedule, repeat, repeatEither, repeatWithErrorMode, RepeatConfig}
 import ox.resilience.{retry, RetryConfig}
 import scala.concurrent.duration.*