Add evalMapFilter combinator

core/shared/src/main/scala/fs2/Stream.scala

@@ -1015,7 +1015,7 @@ final class Stream[+F[_], +O] private[fs2] (private val free: FreeC[F, O, Unit])
     * res0: Unit = ()
     * }}}
     *
-    * But with `evalMaps`, it will print 4 times:
+    * But with `evalMapChunk`, it will print 4 times:
     * @example {{{
     * scala> Stream(1,2,3,4).evalMapChunk(i => IO(println(i))).take(2).compile.drain.unsafeRunSync
     * res0: Unit = ()
@@ -1049,6 +1049,20 @@ final class Stream[+F[_], +O] private[fs2] (private val free: FreeC[F, O, Unit])
     go(s, this).stream
   }
 
+  /**
+    * Effectfully maps and filters the elements of the stream depending on the optionality of the result of the
+    * application of the effectful function `f`.
+    *
+    * @example {{{
+    * scala> import cats.effect.IO
+    * scala> import cats.implicits._
+    * scala> Stream(1, 2, 3, 4, 5).evalMapFilter(n => IO((n * 2).some.filter(_ % 4 == 0))).compile.toList.unsafeRunSync
+    * res0: List[Int] = List(4, 8)
+    * }}}
+    */
+  def evalMapFilter[F2[x] >: F[x], O2](f: O => F2[Option[O2]]): Stream[F2, O2] =
+    evalMap(f).collect { case Some(v) => v }
+
   /**
     * Like `[[Stream#scan]]`, but accepts a function returning an `F[_]`.
     *

core/shared/src/test/scala/fs2/StreamSpec.scala

@@ -1099,6 +1099,30 @@ class StreamSpec extends Fs2Spec {
       r.map(_._2).compile.toVector.asserting(it => assert(it == sVector.map(f)))
     }
 
+    "evalMapFilter" - {
+      "with effectful optional identity function" in forAll { s: Stream[Pure, Int] =>
+        val s1 = s.toList
+        s.evalMapFilter(n => IO.pure(n.some)).compile.toList.asserting(it => assert(it == s1))
+      }
+
+      "with effectful constant function that returns None for any element" in forAll {
+        s: Stream[Pure, Int] =>
+          s.evalMapFilter(_ => IO.pure(none[Int]))
+            .compile
+            .toList
+            .asserting(it => assert(it.isEmpty))
+      }
+
+      "with effectful function that filters out odd elements" in {
+        Stream
+          .range(1, 10)
+          .evalMapFilter(e => IO.pure(e.some.filter(_ % 2 == 0)))
+          .compile
+          .toList
+          .asserting(it => assert(it == List(2, 4, 6, 8)))
+      }
+    }
+
     "evalScan" in forAll { (s: Stream[Pure, Int], n: String) =>
       val sVector = s.toVector
       val f: (String, Int) => IO[String] = (a: String, b: Int) => IO.pure(a + b)