Add Stream.filterMapEffectOption combinator

.changeset/odd-crabs-fry.md

@@ -0,0 +1,5 @@
+---
+"effect": minor
+---
+
+Added Stream.filterMapEffectOption combinator for effectful filtering and mapping in a single step

packages/effect/src/Stream.ts

@@ -1670,6 +1670,22 @@ export const filterMapEffect: {
   ): Stream<A2, E | E2, R | R2>
 } = internal.filterMapEffect
 
+/**
+ * Performs an effectful filter and map in a single step, where the provided
+ * function returns an `Option` wrapped in an `Effect`.
+ *
+ * @category utils
+ */
+export const filterMapEffectOption: {
+  <A, A2, E2, R2>(
+    f: (a: A) => Effect.Effect<Option.Option<A2>, E2, R2>
+  ): <E, R>(self: Stream<A, E, R>) => Stream<A2, E2 | E, R2 | R>
+  <A, E, R, A2, E2, R2>(
+    self: Stream<A, E, R>,
+    f: (a: A) => Effect.Effect<Option.Option<A2>, E2, R2>
+  ): Stream<A2, E | E2, R | R2>
+} = internal.filterMapEffectOption
+
 /**
  * Transforms all elements of the stream for as long as the specified partial
  * function is defined.

packages/effect/src/internal/stream.ts

@@ -2599,6 +2599,47 @@ export const filterMapEffect = dual<
     })
 )
 
+/** @internal */
+export const filterMapEffectOption = dual<
+  <A, A2, E2, R2>(
+    f: (a: A) => Effect.Effect<Option.Option<A2>, E2, R2>
+  ) => <E, R>(self: Stream.Stream<A, E, R>) => Stream.Stream<A2, E2 | E, R2 | R>,
+  <A, E, R, A2, E2, R2>(
+    self: Stream.Stream<A, E, R>,
+    f: (a: A) => Effect.Effect<Option.Option<A2>, E2, R2>
+  ) => Stream.Stream<A2, E2 | E, R2 | R>
+>(
+  2,
+  <A, E, R, A2, E2, R2>(
+    self: Stream.Stream<A, E, R>,
+    f: (a: A) => Effect.Effect<Option.Option<A2>, E2, R2>
+  ): Stream.Stream<A2, E | E2, R | R2> =>
+    suspend(() => {
+      const loop = (
+        iterator: Iterator<A>
+      ): Channel.Channel<Chunk.Chunk<A2>, Chunk.Chunk<A>, E | E2, E, unknown, unknown, R | R2> => {
+        const next = iterator.next()
+        if (next.done) {
+          return core.readWithCause({
+            onInput: (input) => loop(input[Symbol.iterator]()),
+            onFailure: core.failCause,
+            onDone: core.succeed
+          })
+        } else {
+          return pipe(
+            f(next.value),
+            Effect.map(Option.match({
+              onNone: () => loop(iterator),
+              onSome: (a2) => core.flatMap(core.write(Chunk.of(a2)), () => loop(iterator))
+            })),
+            channel.unwrap
+          )
+        }
+      }
+      return new StreamImpl(pipe(toChannel(self), core.pipeTo(loop(Chunk.empty<A>()[Symbol.iterator]()))))
+    })
+)
+
 /** @internal */
 export const filterMapWhile = dual<
   <A, A2>(

packages/effect/test/Stream/filtering.test.ts

@@ -2,6 +2,7 @@ import * as Chunk from "effect/Chunk"
 import * as Effect from "effect/Effect"
 import * as Either from "effect/Either"
 import { pipe } from "effect/Function"
+import * as Option from "effect/Option"
 import * as Stream from "effect/Stream"
 import * as it from "effect/test/utils/extend"
 import { assert, describe } from "vitest"
@@ -46,4 +47,33 @@ describe("Stream", () => {
         [Either.right(1), Either.right(2), Either.left("boom")]
       )
     }))
+
+  it.effect("filterMapEffectOption", () =>
+    Effect.gen(function*($) {
+      const result = yield* $(
+        Stream.make(1, 2, 3, 4),
+        Stream.filterMapEffectOption((n) =>
+          Effect.succeed(
+            n % 2 === 0 ? Option.some(n * 2) : Option.none()
+          )
+        ),
+        Stream.runCollect
+      )
+      assert.deepStrictEqual(Array.from(result), [4, 8])
+    }))
+
+  it.effect("filterMapEffectOption - with failure", () =>
+    Effect.gen(function*($) {
+      const result = yield* $(
+        Stream.make(1, 2, 3),
+        Stream.filterMapEffectOption((n) =>
+          n === 2
+            ? Effect.fail("boom")
+            : Effect.succeed(Option.some(n))
+        ),
+        Stream.runCollect,
+        Effect.either
+      )
+      assert.deepStrictEqual(result, Either.left("boom"))
+    }))
 })