alsoTo - propagate error between sinks

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

@@ -1046,14 +1046,14 @@ trait SourceOps[+T] { outer: Source[T] =>
  other.errorOrClosed(r)
  false
  case t: T @unchecked =>
- if c2.sendOrClosed(t).isValue then
- if other.sendOrClosed(t).isValue then true
- else
-  c2.doneOrClosed().discard
- false
- else
- other.doneOrClosed().discard
- false
+ c2.sendOrClosed(t) match
+ case ChannelClosed.Done => other.doneOrClosed().discard; false
+ case ChannelClosed.Error(r) => other.errorOrClosed(r).discard; false
+ case _ =>
+ other.sendOrClosed(t) match
+  case ChannelClosed.Done => c2.doneOrClosed().discard; false
+  case ChannelClosed.Error(r) => c2.errorOrClosed(r).discard; false
+  case _ => true
  }
  }
  c2

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

@@ -4,6 +4,8 @@ import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 import ox.*
 
+import scala.util.{Failure, Try}
+
 class SourceOpsAlsoToTest extends AnyFlatSpec with Matchers {
 
  behavior of "Source.alsoTo"
@@ -36,6 +38,19 @@ class SourceOpsAlsoToTest extends AnyFlatSpec with Matchers {
  f.join() shouldBe List(1, 2, 3)
  }
 
+ it should "close main channel with error when other errors" in supervised {
+ val c = Channel.withCapacity[Int](1)
+ val f = fork {
+ c.receiveOrClosed()
+ c.receiveOrClosed()
+ c.receiveOrClosed()
+ c.errorOrClosed(new RuntimeException("stop!"))
+ }
+
+ Try(Source.fromIterable(1 to 100).alsoTo(c).toList) shouldBe a[Failure[RuntimeException]]
+ f.join()
+ }
+
  it should "close other channel when main closes" in supervised {
  val other = Channel.rendezvous[Int]
  val forkOther = fork {
@@ -50,4 +65,19 @@ class SourceOpsAlsoToTest extends AnyFlatSpec with Matchers {
  // the channel's buffer is resized internally when it closes, see `com.softwaremill.jox.Channel.closeOrClosed`
  forkOther.join().size should be < 25
  }
+
+ it should "close other channel with error when main errors" in supervised {
+ val other = Channel.rendezvous[Int]
+ val forkOther = fork {
+ Try(other.toList)
+ }
+ val main = Source.fromIterable(1 to 100).alsoTo(other).asInstanceOf[Channel[Int]]
+
+ main.receiveOrClosed()
+ main.receiveOrClosed()
+ main.receiveOrClosed()
+ main.errorOrClosed(new RuntimeException("stop!"))
+
+ forkOther.join() shouldBe a[Failure[RuntimeException]]
+ }
 }