Add Channel#closeWithElement

build.sbt

@@ -219,6 +219,9 @@ ThisBuild / mimaBinaryIssueFilters ++= Seq(
   // Private internal method: #3274
   ProblemFilters.exclude[DirectMissingMethodProblem](
     "fs2.Chunk.platformIterable"
+  ),
+  ProblemFilters.exclude[ReversedMissingMethodProblem](
+    "fs2.concurrent.Channel.sendAndClose"
   )
 )
 

core/shared/src/main/scala/fs2/concurrent/Channel.scala

@@ -110,6 +110,12 @@ sealed trait Channel[F[_], A] {
     */
   def close: F[Either[Channel.Closed, Unit]]
 
+  /** Sends an element through this channel, and closes it right after.
+    *
+    * No-op if the channel is closed, see [[close]] for further info.
+    */
+  def sendAndClose(a: A): F[Either[Channel.Closed, Unit]]
+
   /** Returns true if this channel is closed */
   def isClosed: F[Boolean]
 
@@ -151,7 +157,7 @@ object Channel {
             .drain
         }
 
-        def send(a: A) =
+        def sendImpl(a: A, close: Boolean) =
           F.deferred[Unit].flatMap { producer =>
             state.flatModifyFull { case (poll, state) =>
               state match {
@@ -161,18 +167,24 @@ object Channel {
                 case State(values, size, waiting, producers, closed @ false) =>
                   if (size < capacity)
                     (
-                      State(a :: values, size + 1, None, producers, false),
-                      notifyStream(waiting).as(rightUnit)
+                      State(a :: values, size + 1, None, producers, close),
+                      signalClosure.whenA(close) *> notifyStream(waiting).as(rightUnit)
                     )
                   else
                     (
-                      State(values, size, None, (a, producer) :: producers, false),
-                      notifyStream(waiting).as(rightUnit) <* waitOnBound(producer, poll)
+                      State(values, size, None, (a, producer) :: producers, close),
+                      signalClosure.whenA(close) *>
+                        notifyStream(waiting).as(rightUnit) <*
+                        waitOnBound(producer, poll)
                     )
               }
             }
           }
 
+        def send(a: A) = sendImpl(a, false)
+
+        def sendAndClose(a: A) = sendImpl(a, true)
+
         def trySend(a: A) =
           state.flatModify {
             case s @ State(_, _, _, _, closed @ true) =>

core/shared/src/test/scala/fs2/concurrent/ChannelSuite.scala

@@ -251,4 +251,89 @@ class ChannelSuite extends Fs2Suite {
 
     TestControl.executeEmbed(test)
   }
+
+  test("sendAndClose closes right after sending the last element (bounded case)") {
+    val result = Channel.bounded[IO, Int](1).flatMap { ch =>
+      ch.sendAndClose(0) *> ch.isClosed
+    }
+
+    result.assertEquals(true)
+  }
+
+  test("sendAndClose closes right after sending the last element (unbounded case)") {
+    val result = Channel.unbounded[IO, Int].flatMap { ch =>
+      ch.sendAndClose(0) *> ch.isClosed
+    }
+
+    result.assertEquals(true)
+  }
+
+  test("racing send and sendAndClose should work in bounded(1) case") {
+    val result = Channel.bounded[IO, Int](1).flatMap { ch =>
+      ch.send(0).both(ch.sendAndClose(1)).parProduct(ch.stream.compile.toList)
+    }
+
+    val expectedFirstCase = ((Right(()), Right(())), List(0, 1))
+    val expectedSecondCase = ((Right(()), Left(Channel.Closed)), List(1))
+
+    result.map {
+      case obtained @ ((Right(()), Right(())), List(0, 1)) =>
+        assertEquals(obtained, expectedFirstCase)
+      case obtained @ ((Right(()), Left(Channel.Closed)), List(1)) =>
+        assertEquals(obtained, expectedSecondCase)
+      case _ => fail("An unknown test result was received.")
+    }
+  }
+
+  test("racing send and sendAndClose should work in bounded(2) case") {
+    val result = Channel.bounded[IO, Int](2).flatMap { ch =>
+      ch.send(0).both(ch.sendAndClose(1)).parProduct(ch.stream.compile.toList)
+    }
+
+    val expectedFirstCase = ((Right(()), Right(())), List(0, 1))
+    val expectedSecondCase = ((Right(()), Left(Channel.Closed)), List(1))
+
+    result.map {
+      case obtained @ ((Right(()), Right(())), List(0, 1)) =>
+        assertEquals(obtained, expectedFirstCase)
+      case obtained @ ((Right(()), Left(Channel.Closed)), List(1)) =>
+        assertEquals(obtained, expectedSecondCase)
+      case _ => fail("An unknown test result was received.")
+    }
+  }
+
+  test("racing send and sendAndClose should work in unbounded case") {
+    val result = Channel.unbounded[IO, Int].flatMap { ch =>
+      ch.send(0).both(ch.sendAndClose(1)).parProduct(ch.stream.compile.toList)
+    }
+
+    val expectedFirstCase = ((Right(()), Right(())), List(0, 1))
+    val expectedSecondCase = ((Right(()), Left(Channel.Closed)), List(1))
+
+    result.map {
+      case obtained @ ((Right(()), Right(())), List(0, 1)) =>
+        assertEquals(obtained, expectedFirstCase)
+      case obtained @ ((Right(()), Left(Channel.Closed)), List(1)) =>
+        assertEquals(obtained, expectedSecondCase)
+      case _ => fail("An unknown test result was received.")
+    }
+  }
+
+  test("racing send and sendAndClose should work in synchronous case") {
+    val result = Channel.synchronous[IO, Int].flatMap { ch =>
+      ch.send(0).both(ch.sendAndClose(1)).parProduct(ch.stream.compile.toList)
+    }
+
+    val expectedFirstCase = ((Right(()), Right(())), List(0, 1))
+    val expectedSecondCase = ((Right(()), Left(Channel.Closed)), List(1))
+
+    result.map {
+      case obtained @ ((Right(()), Right(())), List(0, 1)) =>
+        assertEquals(obtained, expectedFirstCase)
+      case obtained @ ((Right(()), Left(Channel.Closed)), List(1)) =>
+        assertEquals(obtained, expectedSecondCase)
+      case _ => fail("An unknown test result was received.")
+    }
+  }
+
 }