Optionally close destination channel when calling pipeTo

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

@@ -347,7 +347,7 @@ trait SourceCompanionOps:
       transportChannel.receiveOrClosed() match
         case ChannelClosed.Done           => c.doneOrClosed()
         case ChannelClosed.Error(r)       => c.errorOrClosed(r)
-        case source: Source[T] @unchecked => source.pipeTo(c)
+        case source: Source[T] @unchecked => source.pipeTo(c, propagateDone = true)
     }
     c
 

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

@@ -1,7 +1,7 @@
 package ox.channels
 
 import ox.channels.ChannelClosedUnion.{mapUnlessError, orThrow}
-import ox.{repeatWhile, supervised}
+import ox.{discard, repeatWhile, supervised}
 
 import scala.collection.mutable
 
@@ -37,13 +37,14 @@ trait SourceDrainOps[+T] { outer: Source[T] =>
     val b = List.newBuilder[T]
     foreachOrError(b += _).mapUnlessError(_ => b.result())
 
-  /** Passes each received element from this channel to the given sink. Blocks until the channel is done. When the channel is done or
-    * there's an error, propagates the closed status downstream and returns.
+  /** Passes each received element from this channel to the given sink. Blocks until the channel is done.
+    *
+    * Errors are always propagated. Successful channel completion is propagated when `propagateDone` is set to `true`.
     */
-  def pipeTo(sink: Sink[T]): Unit =
+  def pipeTo(sink: Sink[T], propagateDone: Boolean): Unit =
     repeatWhile {
       receiveOrClosed() match
-        case ChannelClosed.Done     => sink.doneOrClosed(); false
+        case ChannelClosed.Done     => if propagateDone then sink.doneOrClosed().discard; false
         case e: ChannelClosed.Error => sink.errorOrClosed(e.reason); false
         case t: T @unchecked        => sink.send(t); true
     }

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

@@ -815,9 +815,9 @@ trait SourceOps[+T] { outer: Source[T] =>
     val c = StageCapacity.newChannel[U]
     fork {
       receiveOrClosed() match
-        case ChannelClosed.Done     => alternative.pipeTo(c)
+        case ChannelClosed.Done     => alternative.pipeTo(c, propagateDone = true)
         case ChannelClosed.Error(r) => c.errorOrClosed(r)
-        case t: T @unchecked        => c.sendOrClosed(t); pipeTo(c)
+        case t: T @unchecked        => c.sendOrClosed(t); pipeTo(c, propagateDone = true)
     }
     c
 

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

@@ -50,7 +50,23 @@ class SourceOpsTest extends AnyFlatSpec with Matchers with Eventually {
       val c1 = Source.fromValues(1, 2, 3)
       val c2 = Channel.rendezvous[Int]
 
-      fork { c1.pipeTo(c2) }
+      fork {
+        c1.pipeTo(c2, propagateDone = false)
+        c2.done()
+      }
+
+      c2.toList shouldBe List(1, 2, 3)
+    }
+  }
+
+  it should "pipe one source to another (with done propagation)" in {
+    supervised {
+      val c1 = Source.fromValues(1, 2, 3)
+      val c2 = Channel.rendezvous[Int]
+
+      fork {
+        c1.pipeTo(c2, propagateDone = true)
+      }
 
       c2.toList shouldBe List(1, 2, 3)
     }

kafka/src/test/scala/ox/kafka/KafkaTest.scala

@@ -103,7 +103,7 @@ class KafkaTest extends AnyFlatSpec with Matchers with EmbeddedKafka with Before
           .map(in => (in.value.toLong * 2, in))
           .map((value, original) => SendPacket(ProducerRecord[String, String](destTopic, value.toString), original))
           .mapPublishAndCommit(producerSettings)
-          .pipeTo(metadatas)
+          .pipeTo(metadatas, propagateDone = false)
       }
 
       val inDest = KafkaSource.subscribe(consumerSettings, destTopic)