Migrate to Scala 3.9.0 LTS

.scala-steward.conf

@@ -1,3 +1,3 @@
 updates.pin = [
-  { groupId = "org.scala-lang", artifactId = "scala3-library", version = "3.3." }
+  { groupId = "org.scala-lang", artifactId = "scala3-library", version = "3.9." }
 ]

build.sbt

@@ -7,8 +7,8 @@ commonSmlBuildSettings
 ossPublishSettings
 
 organization := "com.softwaremill.ox"
-scalaVersion := "3.3.8"
-scalacOptions ++= Seq("-Yfuture-lazy-vals", "-java-output-version", "21")
+scalaVersion := "3.9.0"
+scalacOptions ++= Seq("-java-output-version", "21")
 Test / scalacOptions += "-Wconf:msg=unused value of type org.scalatest.Assertion:s"
 Test / scalacOptions += "-Wconf:msg=unused value of type org.scalatest.compatible.Assertion:s"
 mimaPreviousArtifacts := Set.empty // we only use MiMa for `core` for now, using enableMimaSettings

core/src/main/scala/ox/Chunk.scala

@@ -2,6 +2,7 @@ package ox
 
 import java.nio.charset.Charset
 import java.nio.charset.StandardCharsets
+import scala.annotation.unused
 import scala.reflect.ClassTag
 
 /** An immutable finite indexed sequence of elements, backed by multiple IArrays. Optimized for efficient concatenation by maintaining a
@@ -184,10 +185,11 @@ case object Empty extends Chunk[Nothing]:
 object Chunk:
   def empty[A]: Chunk[A] = Empty
 
-  def fromArray[A: ClassTag](array: Array[A]): Chunk[A] =
+  // keeping unused parameter for binary compatibility - deleting it triggered MiMa
+  def fromArray[A](array: Array[A])(using @unused ct: ClassTag[A]): Chunk[A] =
     if array.isEmpty then Empty else NonEmptyChunk(List(IArray.unsafeFromArray(array)))
 
-  def fromIArray[A: ClassTag](array: IArray[A]): Chunk[A] =
+  def fromIArray[A](array: IArray[A])(using @unused ct: ClassTag[A]): Chunk[A] =
     if array.isEmpty then Empty else NonEmptyChunk(List(array))
 
   /** Creates a chunk from elements */

core/src/main/scala/ox/ErrorMode.scala

@@ -1,5 +1,6 @@
 package ox
 
+import scala.annotation.unused
 import scala.reflect.ClassTag
 
 /** Describes the representation of application errors. Such errors have type `E` and are reported in context `F`.
@@ -32,12 +33,12 @@ trait ErrorMode[E, F[_]]:
   /** Adds a suppressed exception to the value being represented by `error`. This is only called if `isError(error)` returns `true`. By
     * default, the suppressed exception is discarded and the original value is returned.
     */
-  def addSuppressedException[T](error: F[T], e: Throwable): F[T] = error
+  def addSuppressedException[T](error: F[T], @unused e: Throwable): F[T] = error
 
   /** Adds a suppressed application error to the value being represented by `error`. This is only called if `isError(error)` returns `true`.
     * By default, the suppressed application error is discarded and the original value is returned.
     */
-  def addSuppressedError[T](error: F[T], e: E): F[T] = error
+  def addSuppressedError[T](error: F[T], @unused e: E): F[T] = error
 end ErrorMode
 
 /** An error mode which doesn't allow reporting application errors.

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

@@ -2,8 +2,6 @@ package ox.channels
 
 import ox.*
 
-import java.util
-
 trait SourceOps[+T]:
   outer: Source[T] =>
 

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

@@ -344,7 +344,7 @@ def selectOrClosedWithin[TV, T1](
   selectOrClosedWithin(timeout, timeoutValue)(source1.receiveClause) match
     case source1.Received(v) => v
     case c: ChannelClosed    => c
-    case tv                  => timeoutValue
+    case _                   => timeoutValue
 
 /** @see [[selectOrClosedWithin(FiniteDuration, TimeoutValue)(Seq[Source])]]. */
 def selectOrClosedWithin[TV, T1, T2](
@@ -355,7 +355,7 @@ def selectOrClosedWithin[TV, T1, T2](
     case source1.Received(v) => v
     case source2.Received(v) => v
     case c: ChannelClosed    => c
-    case tv                  => timeoutValue
+    case _                   => timeoutValue
 
 /** @see [[selectOrClosedWithin(FiniteDuration, TimeoutValue)(Seq[Source])]]. */
 def selectOrClosedWithin[TimeoutValue, T1, T2, T3](
@@ -367,7 +367,7 @@ def selectOrClosedWithin[TimeoutValue, T1, T2, T3](
     case source2.Received(v) => v
     case source3.Received(v) => v
     case c: ChannelClosed    => c
-    case tv                  => timeoutValue
+    case _                   => timeoutValue
 
 /** @see [[selectOrClosedWithin(FiniteDuration, TimeoutValue)(Seq[Source])]]. */
 def selectOrClosedWithin[TV, T1, T2, T3, T4](
@@ -385,7 +385,7 @@ def selectOrClosedWithin[TV, T1, T2, T3, T4](
     case source3.Received(v) => v
     case source4.Received(v) => v
     case c: ChannelClosed    => c
-    case tv                  => timeoutValue
+    case _                   => timeoutValue
 
 /** @see [[selectOrClosedWithin(FiniteDuration, TimeoutValue)(Seq[Source])]]. */
 def selectOrClosedWithin[TV, T1, T2, T3, T4, T5](
@@ -411,7 +411,7 @@ def selectOrClosedWithin[TV, T1, T2, T3, T4, T5](
     case source4.Received(v) => v
     case source5.Received(v) => v
     case c: ChannelClosed    => c
-    case tv                  => timeoutValue
+    case _                   => timeoutValue
 
 /** Select exactly one source from which to receive a value, within the given timeout. Sources should not repeat.
   *

core/src/main/scala/ox/computeIntensive.scala

@@ -1,5 +1,6 @@
 package ox
 
+import scala.compiletime.uninitialized
 import java.util.concurrent.CompletableFuture
 import java.util.concurrent.ExecutionException
 import java.util.concurrent.ExecutorService
@@ -11,7 +12,7 @@ import java.util.concurrent.atomic.AtomicInteger
 // setting & computing the executor
 //
 
-private var customComputeExecutor: ExecutorService = _
+private var customComputeExecutor: ExecutorService = uninitialized
 
 /** Sets the executor used to run computations passed to [[computeIntensive]]. Should be called once, at the start of the application,
   * before any [[computeIntensive]] calls; the executor's lifecycle (shutdown) is then the responsibility of the caller.

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

@@ -58,7 +58,7 @@ object either:
           t match
             case Left(e)  => break(Left(e))
             case Right(a) => a
-        case given boundary.Label[Either[Nothing, Nothing]] =>
+        case _: 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?")
         case _ => error("`.ok()` can only be used within an `either` call.\nIs it present?")
       }
@@ -77,7 +77,7 @@ object either:
       summonFrom {
         case given boundary.Label[Either[E, Nothing]] =>
           break(t.asInstanceOf[Either[E, Nothing]])
-        case given boundary.Label[Either[Nothing, Nothing]] =>
+        case _: 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?")
         case _ => error("`.ok()` can only be used within an `either` call.\nIs it present?")
       }
@@ -91,7 +91,7 @@ object either:
           t match
             case None    => break(Left(()))
             case Some(a) => a
-        case given boundary.Label[Either[Nothing, Nothing]] =>
+        case _: 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?\nNote that for options, the error type must contain a `Unit`."
           )
@@ -110,8 +110,8 @@ object either:
     /** Unwrap the value of the `Option`, short-circuiting the computation to the enclosing [[either]] on guaranteed `None`. */
     transparent inline def ok(): A =
       summonFrom {
-        case given boundary.Label[Either[Unit, Nothing]]    => break(Left(()))
-        case given boundary.Label[Either[Nothing, Nothing]] =>
+        case given boundary.Label[Either[Unit, Nothing]] => break(Left(()))
+        case _: 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?\nNote that for options, the error type must contain a `Unit`."
           )
@@ -133,8 +133,8 @@ object either:
     /** 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]] =>
+        case given boundary.Label[Either[E, Nothing]]    => break(Left(e))
+        case _: 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?")
       }
   end extension

core/src/main/scala/ox/flow/FlowCompanionOps.scala

@@ -161,7 +161,7 @@ trait FlowCompanionOps:
         case None        => false
 
   /** Create a flow which sleeps for the given `timeout` and then completes as done. */
-  def timeout[T](timeout: FiniteDuration): Flow[T] = usingEmitInline: emit =>
+  def timeout[T](timeout: FiniteDuration): Flow[T] = usingEmitInline: _ =>
     sleep(timeout)
 
   /** Creates a flow which concatenates the given `flows` in order. First elements from the first flow are emitted, then from the second

core/src/main/scala/ox/flow/FlowOps.scala

@@ -2,7 +2,6 @@ package ox.flow
 
 import ox.CancellableFork
 import ox.Fork
-import ox.Ox
 import ox.OxUnsupervised
 import ox.channels.BufferCapacity
 import ox.channels.Channel
@@ -623,30 +622,28 @@ class FlowOps[+T]:
     *   An upper bound on the number of child flows that run in parallel.
     */
   def flattenPar[U](parallelism: Int)(using T <:< Flow[U])(using BufferCapacity): Flow[U] = Flow.usingEmitInline: emit =>
-    case class Nested(child: Flow[U])
-    case object ChildDone
-
     unsupervised:
       val childOutputChannel = BufferCapacity.newChannel[U]
-      val childDoneChannel = Channel.unlimited[ChildDone.type]
+      val childDoneChannel = Channel.unlimited[Unit]
 
       // When an error occurs in the parent, propagating it also to `childOutputChannel`, from which we always
       // `select` in the main loop. That way, even if max parallelism is reached, errors in the parent will
       // be discovered without delay.
-      val parentChannel = outer.map(Nested(_)).onError(childOutputChannel.error(_).discard).runToChannel()
+      val parentChannel = outer.map(t => t: Flow[U]).onError(childOutputChannel.error(_).discard).runToChannel()
 
       var runningChannelCount = 1 // parent is running
       var parentDone = false
 
       while runningChannelCount > 0 do
         assert(runningChannelCount <= parallelism + 1)
 
-        val pool: List[Source[Nested] | Source[U] | Source[ChildDone.type]] =
+        val received =
           // +1, because of the parent channel.
-          if runningChannelCount == parallelism + 1 || parentDone then List(childOutputChannel, childDoneChannel)
-          else List(childOutputChannel, childDoneChannel, parentChannel)
+          if runningChannelCount == parallelism + 1 || parentDone
+          then selectOrClosed(childOutputChannel.receiveClause, childDoneChannel.receiveClause)
+          else selectOrClosed(childOutputChannel.receiveClause, childDoneChannel.receiveClause, parentChannel.receiveClause)
 
-        selectOrClosed(pool) match
+        received match
           // Only `parentChannel` might be done, child completion is signalled via `childDoneChannel`.
           case ChannelClosed.Done =>
             parentDone = isSourceDone(parentChannel)
@@ -656,16 +653,16 @@ class FlowOps[+T]:
 
           case e: ChannelClosed.Error => throw e.toThrowable
 
-          case ChildDone => runningChannelCount -= 1
+          case childDoneChannel.Received(_) => runningChannelCount -= 1
 
-          case Nested(t) =>
+          case parentChannel.Received(child) =>
             forkUnsupervised:
-              t.onDone(childDoneChannel.send(ChildDone)).runPipeToSink(childOutputChannel, propagateDone = false)
+              child.onDone(childDoneChannel.send(())).runPipeToSink(childOutputChannel, propagateDone = false)
             .discard
 
             runningChannelCount += 1
 
-          case u: U @unchecked => emit(u)
+          case childOutputChannel.Received(u) => emit(u)
         end match
       end while
   end flattenPar
@@ -918,7 +915,7 @@ class FlowOps[+T]:
     require(elements > 0, "elements must be > 0")
     require(per.toMillis > 0, "per time must be >= 1 ms")
     val emitEveryMillis = (per.toMillis / elements).millis
-    tap(t => sleep(emitEveryMillis))
+    tap(_ => sleep(emitEveryMillis))
   end throttle
 
   /** If this flow has no elements then elements from an `alternative` flow are emitted by the returned flow. If this flow is failed then
@@ -1249,7 +1246,7 @@ class FlowOps[+T]:
 
   /** Discard all elements emitted by this flow. The returned flow completes only when this flow completes (successfully or with an error).
     */
-  def drain(): Flow[Nothing] = Flow.usingEmitInline: emit =>
+  def drain(): Flow[Nothing] = Flow.usingEmitInline: _ =>
     last.run(FlowEmit.fromInline(_ => ()))
 
   /** Always runs `f` after the flow completes, whether it's because all elements are emitted, or when there's an error. */

core/src/main/scala/ox/flow/FlowReactiveOps.scala

@@ -8,7 +8,6 @@ import ox.channels.Sink
 import ox.channels.forkPropagate
 import ox.channels.selectOrClosed
 import ox.discard
-import ox.fork
 import ox.forkDiscard
 import ox.inScopeRunner
 import ox.tapException

core/src/main/scala/ox/flow/FlowTextOps.scala

@@ -147,7 +147,7 @@ trait FlowTextOps[+T]:
       if str != null then output.apply(str)
       (newBuf, State.Pull)
 
-    def processByteOrderMark(bytes: T, buffer: Chunk[Byte], output: FlowEmit[String]): (Chunk[Byte], State) =
+    def processByteOrderMark(bytes: T, buffer: Chunk[Byte]): (Chunk[Byte], State) =
       // A common case, worth checking in advance
       if buffer == null && bytes.length >= bomSize && !bytes.startsWith(bomUtf8) then (bytes, State.Pull)
       else
@@ -168,7 +168,7 @@ trait FlowTextOps[+T]:
       last.run(
         FlowEmit.fromInline: t =>
           val (newBuffer, newState) = state match
-            case State.ProcessBOM => processByteOrderMark(t, buffer, emit)
+            case State.ProcessBOM => processByteOrderMark(t, buffer)
             case State.Pull       => doPull(t, buffer, emit)
 
           buffer = newBuffer