Inline try* helpers; fix CircuitBreakerTest flakiness; use in: style

- Remove fromJoxTryReceiveOrClosed/fromJoxTrySendOrClosed helper methods
  and inline their logic in tryReceiveOrClosed/trySendOrClosed, delegating
  to the existing fromJoxOrT and fromJox methods. Expose fromJox as
  private[ox] to enable this.
- Update ChannelTryTest to use Scala 3 in: syntax (no braces)
- Fix CircuitBreakerTest flakiness by wrapping transition assertions in
  eventually{}, giving a 1s window for actor-dispatched state changes to
  propagate under JVM load

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: adamw

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

@@ -76,7 +76,12 @@ trait Source[+T] extends SourceOps[T] with SourceDrainOps[T]:
     *   [[Some]] with the received value, [[None]] if no value is immediately available, or [[ChannelClosed]] when the channel is closed.
     */
   def tryReceiveOrClosed(): Option[T] | ChannelClosed =
-    ChannelClosed.fromJoxTryReceiveOrClosed(delegate.tryReceiveOrClosed())
+    val r = delegate.tryReceiveOrClosed()
+    if r == null then None
+    else
+      ChannelClosed.fromJoxOrT[T](r.asInstanceOf[AnyRef]) match
+        case c: ChannelClosed => c
+        case v: T @unchecked  => Some(v)
 
   /** Receive a value from the channel. For a variant which throws exceptions when the channel is closed, use [[receive]].
     *
@@ -176,7 +181,12 @@ trait Sink[-T]:
     *   `true` if the value was sent, `false` if there's no space/waiting receiver, or [[ChannelClosed]] when the channel is closed.
     */
   def trySendOrClosed(t: T): Boolean | ChannelClosed =
-    ChannelClosed.fromJoxTrySendOrClosed(delegate.asInstanceOf[JSink[T]].trySendOrClosed(t))
+    val r = delegate.asInstanceOf[JSink[T]].trySendOrClosed(t)
+    if r == null then true // null = sent
+    else
+      ChannelClosed.fromJox(r.asInstanceOf[AnyRef]) match
+        case c: ChannelClosed => c
+        case _                => false // sentinel = not sent
 
   /** Send a value to the channel. For a variant which throws exceptions when the channel is closed, use [[send]].
     *

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

@@ -16,27 +16,7 @@ object ChannelClosed:
   private[ox] def fromJoxOrUnit(joxResult: AnyRef): Unit | ChannelClosed =
     if joxResult == null then () else fromJox(joxResult).asInstanceOf[ChannelClosed]
 
-  /** Converts the result of jox's `tryReceiveOrClosed()` (which returns `T | JChannelClosed | null`) to `Option[T] | ChannelClosed`. `null`
-    * means no value was immediately available.
-    */
-  private[ox] def fromJoxTryReceiveOrClosed[T](joxResult: AnyRef): Option[T] | ChannelClosed =
-    joxResult match
-      case null             => None
-      case _: JChannelDone  => Done
-      case e: JChannelError => Error(e.cause())
-      case v                => Some(v.asInstanceOf[T])
-
-  /** Converts the result of jox's `trySendOrClosed()` (which returns `null | JChannelClosed | sentinel`) to `Boolean | ChannelClosed`.
-    * `null` means the value was sent, any other non-`ChannelClosed` value (sentinel) means it was not sent.
-    */
-  private[ox] def fromJoxTrySendOrClosed(joxResult: AnyRef): Boolean | ChannelClosed =
-    joxResult match
-      case null             => true
-      case _: JChannelDone  => Done
-      case e: JChannelError => Error(e.cause())
-      case _                => false
-
-  private def fromJox(joxResult: AnyRef): AnyRef | ChannelClosed =
+  private[ox] def fromJox(joxResult: AnyRef): AnyRef | ChannelClosed =
     joxResult match
       case _: JChannelDone  => Done
       case e: JChannelError => Error(e.cause())

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

@@ -6,129 +6,107 @@ import org.scalatest.matchers.should.Matchers
 class ChannelTryTest extends AnyFlatSpec with Matchers:
   // trySend
 
-  "trySend" should "return true when there is space in the buffer" in {
+  "trySend" should "return true when there is space in the buffer" in:
     val c = Channel.buffered[Int](2)
     c.trySend(1) shouldBe true
     c.trySend(2) shouldBe true
-  }
 
-  it should "return false when the buffer is full" in {
+  it should "return false when the buffer is full" in:
     val c = Channel.buffered[Int](1)
     c.trySend(1) shouldBe true
     c.trySend(2) shouldBe false
-  }
 
-  it should "return true for an unlimited channel" in {
+  it should "return true for an unlimited channel" in:
     val c = Channel.unlimited[Int]
     c.trySend(1) shouldBe true
     c.trySend(2) shouldBe true
-  }
 
-  it should "throw ChannelClosedException.Done when the channel is done" in {
+  it should "throw ChannelClosedException.Done when the channel is done" in:
     val c = Channel.buffered[Int](2)
     c.done()
-    assertThrows[ChannelClosedException.Done] {
+    assertThrows[ChannelClosedException.Done]:
       c.trySend(1)
-    }
-  }
 
-  it should "throw ChannelClosedException.Error when the channel is in error" in {
+  it should "throw ChannelClosedException.Error when the channel is in error" in:
     val c = Channel.buffered[Int](2)
     c.error(new RuntimeException("test"))
-    assertThrows[ChannelClosedException.Error] {
+    assertThrows[ChannelClosedException.Error]:
       c.trySend(1)
-    }
-  }
 
   // trySendOrClosed
 
-  "trySendOrClosed" should "return true when there is space in the buffer" in {
+  "trySendOrClosed" should "return true when there is space in the buffer" in:
     val c = Channel.buffered[Int](2)
     c.trySendOrClosed(1) shouldBe true
     c.trySendOrClosed(2) shouldBe true
-  }
 
-  it should "return false when the buffer is full" in {
+  it should "return false when the buffer is full" in:
     val c = Channel.buffered[Int](1)
     c.trySendOrClosed(1) shouldBe true
     c.trySendOrClosed(2) shouldBe false
-  }
 
-  it should "return ChannelClosed.Done when the channel is done" in {
+  it should "return ChannelClosed.Done when the channel is done" in:
     val c = Channel.buffered[Int](2)
     c.done()
     c.trySendOrClosed(1) shouldBe ChannelClosed.Done
-  }
 
-  it should "return ChannelClosed.Error when the channel is in error" in {
+  it should "return ChannelClosed.Error when the channel is in error" in:
     val c = Channel.buffered[Int](2)
     val ex = new RuntimeException("test")
     c.error(ex)
     c.trySendOrClosed(1) should matchPattern { case ChannelClosed.Error(_) => }
-  }
 
   // tryReceive
 
-  "tryReceive" should "return Some with the value when one is available" in {
+  "tryReceive" should "return Some with the value when one is available" in:
     val c = Channel.buffered[Int](2)
     c.send(1)
     c.send(2)
     c.tryReceive() shouldBe Some(1)
     c.tryReceive() shouldBe Some(2)
-  }
 
-  it should "return None when the buffer is empty" in {
+  it should "return None when the buffer is empty" in:
     val c = Channel.buffered[Int](2)
     c.tryReceive() shouldBe None
-  }
 
-  it should "throw ChannelClosedException.Done when the channel is done and empty" in {
+  it should "throw ChannelClosedException.Done when the channel is done and empty" in:
     val c = Channel.buffered[Int](2)
     c.done()
-    assertThrows[ChannelClosedException.Done] {
+    assertThrows[ChannelClosedException.Done]:
       c.tryReceive()
-    }
-  }
 
-  it should "throw ChannelClosedException.Error when the channel is in error" in {
+  it should "throw ChannelClosedException.Error when the channel is in error" in:
     val c = Channel.buffered[Int](2)
     c.send(1)
     c.error(new RuntimeException("test"))
-    assertThrows[ChannelClosedException.Error] {
+    assertThrows[ChannelClosedException.Error]:
       c.tryReceive()
-    }
-  }
 
   // tryReceiveOrClosed
 
-  "tryReceiveOrClosed" should "return Some with the value when one is available" in {
+  "tryReceiveOrClosed" should "return Some with the value when one is available" in:
     val c = Channel.buffered[Int](2)
     c.send(1)
     c.tryReceiveOrClosed() shouldBe Some(1)
-  }
 
-  it should "return None when the buffer is empty" in {
+  it should "return None when the buffer is empty" in:
     val c = Channel.buffered[Int](2)
     c.tryReceiveOrClosed() shouldBe None
-  }
 
-  it should "return ChannelClosed.Done when the channel is done and empty" in {
+  it should "return ChannelClosed.Done when the channel is done and empty" in:
     val c = Channel.buffered[Int](2)
     c.done()
     c.tryReceiveOrClosed() shouldBe ChannelClosed.Done
-  }
 
-  it should "return ChannelClosed.Error when the channel is in error" in {
+  it should "return ChannelClosed.Error when the channel is in error" in:
     val c = Channel.buffered[Int](2)
     c.send(1)
     c.error(new RuntimeException("test"))
     c.tryReceiveOrClosed() should matchPattern { case ChannelClosed.Error(_) => }
-  }
 
-  it should "return Some(value) from a done channel that still has buffered values" in {
+  it should "return Some(value) from a done channel that still has buffered values" in:
     val c = Channel.buffered[Int](2)
     c.send(42)
     c.done()
     c.tryReceiveOrClosed() shouldBe Some(42)
-  }
 end ChannelTryTest

core/src/test/scala/ox/resilience/CircuitBreakerTest.scala

@@ -3,11 +3,15 @@ package ox.resilience
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 import org.scalatest.OptionValues
+import org.scalatest.concurrent.Eventually
+import org.scalatest.time.{Millis, Seconds, Span}
 import scala.concurrent.duration.*
 import ox.*
 import org.scalatest.EitherValues
 
-class CircuitBreakerTest extends AnyFlatSpec with Matchers with OptionValues with EitherValues:
+class CircuitBreakerTest extends AnyFlatSpec with Matchers with OptionValues with EitherValues with Eventually:
+  implicit override val patienceConfig: PatienceConfig =
+    PatienceConfig(timeout = Span(1, Seconds), interval = Span(50, Millis))
   behavior of "Circuit Breaker run operations"
 
   it should "run operation when metrics are not exceeded" in supervised {
@@ -190,23 +194,23 @@ class CircuitBreakerTest extends AnyFlatSpec with Matchers with OptionValues wit
 
     // 750ms: the first operation failed, should be open
     sleep(500.millis)
-    circuitBreaker.stateMachine.state shouldBe a[CircuitBreakerState.Open]
+    eventually { circuitBreaker.stateMachine.state shouldBe a[CircuitBreakerState.Open] }
 
     // 1750ms: first operation failed more than 1s ago, second operation failed less than 1s ago and was ignored
     sleep(1.second)
-    circuitBreaker.stateMachine.state shouldBe a[CircuitBreakerState.HalfOpen]
+    eventually { circuitBreaker.stateMachine.state shouldBe a[CircuitBreakerState.HalfOpen] }
 
     // 2250ms: more than 1s after the last failing operation, should be now half-open
     sleep(500.millis)
     circuitBreaker.stateMachine.state shouldBe a[CircuitBreakerState.HalfOpen]
 
     // 3250ms: at 2500ms 1 sec timeout on halfOpen state passes, we go back to open
     sleep(1.second)
-    circuitBreaker.stateMachine.state shouldBe a[CircuitBreakerState.Open]
+    eventually { circuitBreaker.stateMachine.state shouldBe a[CircuitBreakerState.Open] }
 
     // 3750ms: at 3500ms we go to halfOpen again
     sleep(1000.millis)
-    circuitBreaker.stateMachine.state shouldBe a[CircuitBreakerState.HalfOpen]
+    eventually { circuitBreaker.stateMachine.state shouldBe a[CircuitBreakerState.HalfOpen] }
   }
 
   it should "correctly calculate metrics when results come in after state change" in supervised {
@@ -246,18 +250,17 @@ class CircuitBreakerTest extends AnyFlatSpec with Matchers with OptionValues wit
 
     // 750ms: the first operation failed, should be open
     sleep(500.millis)
-    circuitBreaker.stateMachine.state shouldBe a[CircuitBreakerState.Open]
+    eventually { circuitBreaker.stateMachine.state shouldBe a[CircuitBreakerState.Open] }
 
     // 1750ms: first operation failed more than 1s ago, second operation failed less than 1s ago and was ignored
     sleep(1.second)
-    circuitBreaker.stateMachine.state shouldBe a[CircuitBreakerState.HalfOpen]
+    eventually { circuitBreaker.stateMachine.state shouldBe a[CircuitBreakerState.HalfOpen] }
 
     // 2250ms: complete enough operations for halfOpen state - since success should switch back to Closed
     sleep(500.millis)
     circuitBreaker.runOrDropEither(Right("c")).discard
-    sleep(100.millis) // wait for state to register
     // Should go back to closed, we have one successful operation
-    circuitBreaker.stateMachine.state shouldBe a[CircuitBreakerState.Closed]
+    eventually { circuitBreaker.stateMachine.state shouldBe a[CircuitBreakerState.Closed] }
   }
 
 end CircuitBreakerTest