Add ZIO.filterPar (#3915)

core-tests/shared/src/test/scala/zio/ZIOSpec.scala

@@ -457,6 +457,24 @@ object ZIOSpec extends ZIOBaseSpec {
           assert(effects)(equalTo(List(2, 4, 6, 3, 5, 6)))
       }
     ),
+    suite("filterPar")(
+      testM("filters a collection in parallel using an effectual predicate") {
+        val as = Iterable(2, 4, 6, 3, 5, 6, 10, 11, 15, 17, 20, 22, 23, 25, 28)
+        for {
+          results <- ZIO.filterPar(as)(a => UIO(a % 2 == 0))
+        } yield assert(results)(equalTo(List(2, 4, 6, 6, 10, 20, 22, 28)))
+      }
+    ),
+    suite("filterNotPar")(
+      testM(
+        "filters a collection in parallel using an effectual predicate, removing all elements that satisfy the predicate"
+      ) {
+        val as = Iterable(2, 4, 6, 3, 5, 6, 10, 11, 15, 17, 20, 22, 23, 25, 28)
+        for {
+          results <- ZIO.filterNotPar(as)(a => UIO(a % 2 == 0))
+        } yield assert(results)(equalTo(List(3, 5, 11, 15, 17, 23, 25)))
+      }
+    ),
     suite("filterOrElse")(
       testM("returns checked failure from held value") {
         val goodCase =

core/shared/src/main/scala/zio/IO.scala

@@ -292,12 +292,24 @@ object IO {
   def filter[E, A](as: Iterable[A])(f: A => IO[E, Boolean]): IO[E, List[A]] =
     ZIO.filter(as)(f)
 
+  /**
+   * @see [[zio.ZIO.filterPar]]
+   */
+  def filterPar[E, A](as: Iterable[A])(f: A => IO[E, Boolean]): IO[E, List[A]] =
+    ZIO.filterPar(as)(f)
+
   /**
    * @see [[zio.ZIO.filterNot]]
    */
   def filterNot[E, A](as: Iterable[A])(f: A => IO[E, Boolean]): IO[E, List[A]] =
     ZIO.filterNot(as)(f)
 
+  /**
+   * @see [[zio.ZIO.filterNotPar]]
+   */
+  def filterNotPar[E, A](as: Iterable[A])(f: A => IO[E, Boolean]): IO[E, List[A]] =
+    ZIO.filterNotPar(as)(f)
+
   /**
    * @see See [[zio.ZIO.firstSuccessOf]]
    */

core/shared/src/main/scala/zio/RIO.scala

@@ -316,12 +316,24 @@ object RIO {
   def filter[R, A](as: Iterable[A])(f: A => RIO[R, Boolean]): RIO[R, List[A]] =
     ZIO.filter(as)(f)
 
+  /**
+   * @see [[zio.ZIO.filterPar]]
+   */
+  def filterPar[R, A](as: Iterable[A])(f: A => RIO[R, Boolean]): RIO[R, List[A]] =
+    ZIO.filterPar(as)(f)
+
   /**
    * @see [[zio.ZIO.filterNot]]
    */
   def filterNot[R, A](as: Iterable[A])(f: A => RIO[R, Boolean]): RIO[R, List[A]] =
     ZIO.filterNot(as)(f)
 
+  /**
+   * @see [[zio.ZIO.filterNotPar]]
+   */
+  def filterNotPar[R, A](as: Iterable[A])(f: A => RIO[R, Boolean]): RIO[R, List[A]] =
+    ZIO.filterNotPar(as)(f)
+
   /**
    * @see See [[zio.ZIO.first]]
    */

core/shared/src/main/scala/zio/Task.scala

@@ -291,12 +291,24 @@ object Task extends TaskPlatformSpecific {
   def filter[A](as: Iterable[A])(f: A => Task[Boolean]): Task[List[A]] =
     ZIO.filter(as)(f)
 
+  /**
+   * @see [[zio.ZIO.filterPar]]
+   */
+  def filterPar[A](as: Iterable[A])(f: A => Task[Boolean]): Task[List[A]] =
+    ZIO.filterPar(as)(f)
+
   /**
    * @see [[zio.ZIO.filterNot]]
    */
   def filterNot[A](as: Iterable[A])(f: A => Task[Boolean]): Task[List[A]] =
     ZIO.filterNot(as)(f)
 
+  /**
+   * @see [[zio.ZIO.filterNotPar]]
+   */
+  def filterNotPar[A](as: Iterable[A])(f: A => Task[Boolean]): Task[List[A]] =
+    ZIO.filterNotPar(as)(f)
+
   /**
    * @see See [[zio.ZIO.firstSuccessOf]]
    */

core/shared/src/main/scala/zio/UIO.scala

@@ -265,12 +265,24 @@ object UIO {
   def filter[A](as: Iterable[A])(f: A => UIO[Boolean]): UIO[List[A]] =
     ZIO.filter(as)(f)
 
+  /**
+   * @see [[zio.ZIO.filterPar]]
+   */
+  def filterPar[A](as: Iterable[A])(f: A => UIO[Boolean]): UIO[List[A]] =
+    ZIO.filterPar(as)(f)
+
   /**
    * @see [[zio.ZIO.filterNot]]
    */
   def filterNot[A](as: Iterable[A])(f: A => UIO[Boolean]): UIO[List[A]] =
     ZIO.filterNot(as)(f)
 
+  /**
+   * @see [[zio.ZIO.filterNotPar]]
+   */
+  def filterNotPar[A](as: Iterable[A])(f: A => UIO[Boolean]): UIO[List[A]] =
+    ZIO.filterNotPar(as)(f)
+
   /**
    * @see [[zio.ZIO.firstSuccessOf]]
    */

core/shared/src/main/scala/zio/URIO.scala

@@ -288,12 +288,24 @@ object URIO {
   def filter[R, A](as: Iterable[A])(f: A => URIO[R, Boolean]): URIO[R, List[A]] =
     ZIO.filter(as)(f)
 
+  /**
+   * @see [[zio.ZIO.filterPar]]
+   */
+  def filterPar[R, A](as: Iterable[A])(f: A => URIO[R, Boolean]): URIO[R, List[A]] =
+    ZIO.filterPar(as)(f)
+
   /**
    * @see [[zio.ZIO.filterNot]]
    */
   def filterNot[R, A](as: Iterable[A])(f: A => URIO[R, Boolean]): URIO[R, List[A]] =
     ZIO.filterNot(as)(f)
 
+  /**
+   * @see [[zio.ZIO.filterNotPar]]
+   */
+  def filterNotPar[R, A](as: Iterable[A])(f: A => URIO[R, Boolean]): URIO[R, List[A]] =
+    ZIO.filterNotPar(as)(f)
+
   /**
    * @see [[zio.ZIO.first]]
    */

core/shared/src/main/scala/zio/ZIO.scala

@@ -2617,13 +2617,28 @@ object ZIO extends ZIOCompanionPlatformSpecific {
       f(a).zipWith(zio)((p, as) => if (p) a :: as else as)
     }
 
+  /**
+   * Filters the collection in parallel using the specified effectual predicate.
+   * See [[filter]] for a sequential version of it.
+   */
+  def filterPar[R, E, A](as: Iterable[A])(f: A => ZIO[R, E, Boolean]): ZIO[R, E, List[A]] =
+    ZIO.foreachPar(as)(a => f(a).map(if (_) Some(a) else None)).map(_.flatten)
+
   /**
    * Filters the collection using the specified effectual predicate, removing
    * all elements that satisfy the predicate.
    */
   def filterNot[R, E, A](as: Iterable[A])(f: A => ZIO[R, E, Boolean]): ZIO[R, E, List[A]] =
     filter(as)(f(_).map(!_))
 
+  /**
+   * Filters the collection in parallel using the specified effectual predicate,
+   * removing all elements that satisfy the predicate.
+   * See [[filterNot]] for a sequential version of it.
+   */
+  def filterNotPar[R, E, A](as: Iterable[A])(f: A => ZIO[R, E, Boolean]): ZIO[R, E, List[A]] =
+    filterPar(as)(f(_).map(!_))
+
   /**
    * Returns an effectful function that extracts out the first element of a
    * tuple.