feat: implement last and lastOption operators

The `lastOption` operator returns the last element in `Source` wrapped in
`Some` or `None` in case when source is empty or failed. Note that this
is a terminal operation for source e.g.:

  Source.empty[Int].lastOption()  // None
  val s = Source.fromValues(1, 2)
  s.lastOption()                  // Some(2)
  s.receive()                     // ChannelClosed.Done

The `last` operator returns the last element in `Source` or throws
`NoSuchElementException` in case when it is empty. In case when `receive()`
fails then `ChannelClosedException.Error` exception is thrown. It is also
a terminal operation e.g.:

  Source.empty[Int].last()        // throws NoSuchElementException("cannot obtain last from an empty source")
  val s = Source.fromValues(1, 2)
  s.last()                        // 2
  s.receive()                     // ChannelClosed.Done

Note that ChannelClosedException.Error was improved to contain `cause`
exception (if available).

Author: geminicaprograms

core/src/main/scala/ox/channels/ChannelClosed.scala

@@ -8,6 +8,6 @@ object ChannelClosed:
   case class Error(reason: Option[Throwable]) extends ChannelClosed
   case object Done extends ChannelClosed
 
-enum ChannelClosedException(reason: Option[Throwable]) extends Exception:
+enum ChannelClosedException(reason: Option[Throwable]) extends Exception(reason.orNull):
   case Error(reason: Option[Throwable]) extends ChannelClosedException(reason)
   case Done() extends ChannelClosedException(None)

core/src/main/scala/ox/channels/SourceOps.scala

@@ -565,6 +565,61 @@ trait SourceOps[+T] { this: Source[T] =>
         case ChannelClosed.Error(r) => throw r.getOrElse(new NoSuchElementException("getting head failed"))
         case t: T @unchecked        => t
     }
+
+  /** Returns the last element from this source wrapped in `Some` or `None` when the source is empty or fails during the `receive()``
+    * operation. Note that `lastOption` is a terminal operation leaving the source in `ChannelClosed.Done` state.
+    *
+    * @return
+    *   A `Some(last element)` if source is not empty or None` otherwise.
+    * @example
+    *   {{{
+    *   import ox.*
+    *   import ox.channels.Source
+    *
+    *   supervised {
+    *     Source.empty[Int].lastOption()  // None
+    *     val s = Source.fromValues(1, 2)
+    *     s.lastOption()                  // Some(2)
+    *     s.receive()                     // ChannelClosed.Done
+    *   }
+    *   }}}
+    */
+  def lastOption(): Option[T] = Try(last()).toOption
+
+  /** Returns the last element from this source or throws `NoSuchElementException` when the source is empty. In case when the `receive()`
+    * operation fails then `ChannelClosedException.Error` exception is thrown. Note that `last` is a terminal operation leaving the source
+    * in `ChannelClosed.Done` state.
+    *
+    * @return
+    *   A last element if source is not empty or throws otherwise.
+    * @throws NoSuchElementException
+    *   When source is empty or `receive()` failed without error.
+    * @throws ChannelClosedException.Error
+    *   When `receive()` fails then this exception is thrown.
+    * @example
+    *   {{{
+    *   import ox.*
+    *   import ox.channels.Source
+    *
+    *   supervised {
+    *     Source.empty[Int].last()        // throws NoSuchElementException("cannot obtain last from an empty source")
+    *     val s = Source.fromValues(1, 2)
+    *     s.last()                        // 2
+    *     s.receive()                     // ChannelClosed.Done
+    *   }
+    *   }}}
+    */
+  def last(): T =
+    supervised {
+      var value: Option[T] = None
+      repeatUntil {
+        receive() match
+          case ChannelClosed.Done     => true
+          case e: ChannelClosed.Error => throw e.toThrowable
+          case t: T @unchecked        => value = Some(t); false
+      }
+      value.getOrElse(throw new NoSuchElementException("cannot obtain last from an empty source"))
+    }
 }
 
 trait SourceCompanionOps:

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

@@ -0,0 +1,30 @@
+package ox.channels
+
+import org.scalatest.OptionValues
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+import ox.*
+
+class SourceOpsLastOptionTest extends AnyFlatSpec with Matchers with OptionValues {
+  behavior of "SourceOps.lastOption"
+
+  it should "return None for the empty source" in supervised {
+    Source.empty[Int].lastOption() shouldBe None
+  }
+
+  it should "return None for the failed source" in supervised {
+    Source
+      .failed(new RuntimeException("source is broken"))
+      .lastOption() shouldBe None
+  }
+
+  it should "return last element wrapped in Some for the non-empty source" in supervised {
+    Source.fromValues(1, 2).lastOption().value shouldBe 2
+  }
+
+  it should "drain the source" in supervised {
+    val s = Source.fromValues(1)
+    s.lastOption().value shouldBe 1
+    s.receive() shouldBe ChannelClosed.Done
+  }
+}

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

@@ -0,0 +1,39 @@
+package ox.channels
+
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+import ox.*
+
+class SourceOpsLastTest extends AnyFlatSpec with Matchers {
+  behavior of "SourceOps.last"
+
+  it should "throw NoSuchElementException for the empty source" in supervised {
+    the[NoSuchElementException] thrownBy {
+      Source.empty[Int].last()
+    } should have message "cannot obtain last from an empty source"
+  }
+
+  it should "throw ChannelClosedException.Error with exception and message that was thrown during retrieval" in supervised {
+    the[ChannelClosedException.Error] thrownBy {
+      Source
+        .failed(new RuntimeException("source is broken"))
+        .last()
+    } should have message "java.lang.RuntimeException: source is broken"
+  }
+
+  it should "throw ChannelClosedException.Error for source failed without exception" in supervised {
+    the[ChannelClosedException.Error] thrownBy {
+      Source.failedWithoutReason[Int]().last()
+    }
+  }
+
+  it should "return last element for the non-empty source" in supervised {
+    Source.fromValues(1, 2).last() shouldBe 2
+  }
+
+  it should "drain the source" in supervised {
+    val s = Source.fromValues(1)
+    s.last() shouldBe 1
+    s.receive() shouldBe ChannelClosed.Done
+  }
+}