Add Channel#closeWithElement

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]
 
@@ -188,6 +194,28 @@ object Channel {
                 (s, rightFalse.pure[F])
           }
 
+        def sendAndClose(a: A) =
+          F.deferred[Unit].flatMap { producer =>
+            state.flatModifyFull { case (_, state) =>
+              state match {
+                case s @ State(_, _, _, _, closed @ true) =>
+                  (s, Channel.closed[Unit].pure[F])
+
+                case State(values, size, waiting, producers, closed @ false) =>
+                  if (size < capacity)
+                    (
+                      State(a :: values, size + 1, None, producers, false),
+                      notifyStream(waiting).as(rightUnit)
+                    )
+                  else
+                    (
+                      State(values, size, None, (a, producer) :: producers, true),
+                      notifyStream(waiting).as(rightUnit) <* signalClosure
+                    )
+              }
+            }
+          }
+
         def close =
           state.flatModify {
             case s @ State(_, _, _, _, closed @ true) =>

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

@@ -251,4 +251,18 @@ class ChannelSuite extends Fs2Suite {
 
     TestControl.executeEmbed(test)
   }
+
+  test("sendAndClose closes right after sending the last element") {
+    val v = Vector(1, 2, 3, 4)
+    val capacity = 3
+    val p = for {
+      chan <- Channel.bounded[IO, Int](capacity)
+      _ <- v.traverse(chan.sendAndClose)
+      isClosed <- chan.isClosed
+      res <- chan.stream.chunks.take(1).compile.lastOrError
+    } yield (res.toVector, isClosed)
+
+    p._1F.assertEquals(v.take(capacity))
+    p._2F.assertEquals(true)
+  }
 }