effect: add Stream.filterMapEffectOption function (Effect-TS#3023)

kristiannotari · Jul 10, 2024 · e9c4a2e · e9c4a2e
1 parent 6bb9507
commit e9c4a2e
Show file tree

Hide file tree

Showing 3 changed files with 54 additions and 1 deletion.
diff --git a/packages/effect/src/Stream.ts b/packages/effect/src/Stream.ts
@@ -2702,6 +2702,36 @@ export const mapEffect: {
   ): Stream<A2, E | E2, R | R2>
 } = _groupBy.mapEffectOptions
 
+/**
+ * Maps over elements of the stream with the specified effectful function
+ * and filters out `None` values
+ *
+ * @example
+ * import { Effect, Stream, Console } from 'effect';
+ *
+ * // returns `Some(number)` if the number is even, otherwise `None`
+ * const even = (a: number) => a % 2 === 0 ? Effect.succeedSome(a) : Effect.succeedNone
+ *
+ * const result = Stream.make(0, 1, 2, 4).pipe(
+ *   Stream.filterMapEffectOption(even),
+ *   Stream.runCollect,
+ *   Effect.runPromise,
+ * )
+ * // => [0, 2, 4]
+ *
+ * @since 3.3.5
+ * @category utils
+ */
+export const filterMapEffectOption: {
+  <A, A2, E2, R2>(
+    fn: (a: A) => Effect.Effect<Option.Option<A2>, E2, R2>
+  ): <E, R>(stream: Stream<A, E, R>) => Stream<A2, E2 | E, R2 | R>
+  <A, E, R, A2, E2, R2>(
+    stream: Stream<A, E, R>,
+    fn: (a: A) => Effect.Effect<Option.Option<A2>, E2, R2>
+  ): Stream<A2, E2 | E, R2 | R>
+} = _groupBy.filterMapEffectOption
+
 /**
  * Transforms the errors emitted by this stream using `f`.
  *

diff --git a/packages/effect/src/internal/groupBy.ts b/packages/effect/src/internal/groupBy.ts
@@ -4,7 +4,7 @@ import * as Chunk from "../Chunk.js"
 import * as Deferred from "../Deferred.js"
 import * as Effect from "../Effect.js"
 import * as Exit from "../Exit.js"
-import { dual, pipe } from "../Function.js"
+import { dual, identity, pipe } from "../Function.js"
 import type * as GroupBy from "../GroupBy.js"
 import * as Option from "../Option.js"
 import { pipeArguments } from "../Pipeable.js"
@@ -206,6 +206,19 @@ export const groupBy = dual<
     )
 )
 
+/** @internal */
+export const filterMapEffectOption = dual<
+  <A, A2, E2, R2>(
+    fn: (a: A) => Effect.Effect<Option.Option<A2>, E2, R2>
+  ) => <E, R>(
+    stream: Stream.Stream<A, E, R>
+  ) => Stream.Stream<A2, E2 | E, R2 | R>,
+  <A, E, R, A2, E2, R2>(
+    stream: Stream.Stream<A, E, R>,
+    fn: (a: A) => Effect.Effect<Option.Option<A2>, E2, R2>
+  ) => Stream.Stream<A2, E2 | E, R2 | R>
+>(2, (s, fn) => stream.filterMap(mapEffectOptions(s, fn), identity))
+
 /** @internal */
 export const mapEffectOptions = dual<
   {

diff --git a/packages/effect/test/Stream/filtering.test.ts b/packages/effect/test/Stream/filtering.test.ts
@@ -46,4 +46,14 @@ describe("Stream", () => {
         [Either.right(1), Either.right(2), Either.left("boom")]
       )
     }))
+
+  it.effect("filterMapEffectOption", () =>
+    Effect.gen(function*() {
+      const even = (a: number) => a % 2 === 0 ? Effect.succeedSome(a) : Effect.succeedNone
+      const result = yield* Stream.make(0, 1, 2, 4).pipe(
+        Stream.filterMapEffectOption(even),
+        Stream.runCollect
+      )
+      assert.deepStrictEqual(Array.from(result), [0, 2, 4])
+    }))
 })