feat: use a better approach to implement it

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

@@ -393,6 +393,38 @@ trait SourceOps[+T] { outer: Source[T] =>
  }
  c
 
+ /** Pipes the elements of child sources into the output source. If the parent source or any of the child sources emit an error, the
+ * pulling stops and the output source emits the error.
+ */
+ def flatten[U](using Ox, StageCapacity, T <:< Source[U]): Source[U] = {
+ val c2 = StageCapacity.newChannel[U]
+
+ forkPropagate(c2) {
+ var pool = List[Source[T] | Source[U]](this)
+ repeatWhile {
+ selectOrClosed(pool) match {
+ case ChannelClosed.Done =>
+ // TODO: best to remove the specific channel that signalled to be Done
+ pool = pool.filterNot(_.isClosedForReceiveDetail.contains(ChannelClosed.Done))
+ if pool.isEmpty then
+ c2.doneOrClosed()
+ false
+ else true
+ case ChannelClosed.Error(e) =>
+ c2.errorOrClosed(e)
+ false
+ case t: Source[T] @unchecked =>
+ pool = t :: pool
+ true
+ case r: U @unchecked =>
+ c2.sendOrClosed(r).isValue
+ }
+ }
+ }
+
+ c2
+ }
+
  /** Concatenates this source with the `other` source. The resulting source will emit elements from this source first, and then from the
  * `other` source.
  *

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

@@ -8,7 +8,7 @@ import ox.*
 import java.util.concurrent.CountDownLatch
 import scala.collection.mutable.ListBuffer
 
-class SourceOfSourceOpsTest extends AnyFlatSpec with Matchers with OptionValues {
+class SourceOpsFlattenTest extends AnyFlatSpec with Matchers with OptionValues {
 
  "flatten" should "pipe all elements of the child sources into the output source" in {
  supervised {