feat(Stream): switchMap

[dilame]

Type

• Refactor
• Feature
• Bug Fix
• Optimization
• Documentation Update

Description

Related

• Related Issue #
• Closes #

[changeset-bot]

🦋 Changeset detected

Latest commit: b667ac0

The changes in this PR will be included in the next version bump.

This PR includes changesets to release 30 packages

Name	Type
effect	Minor
@effect/cli	Major
@effect/cluster-browser	Major
@effect/cluster-node	Major
@effect/cluster-workflow	Major
@effect/cluster	Major
@effect/experimental	Major
@effect/opentelemetry	Major
@effect/platform-browser	Major
@effect/platform-bun	Major
@effect/platform-node-shared	Major
@effect/platform-node	Major
@effect/platform	Major
@effect/printer-ansi	Major
@effect/printer	Major
@effect/rpc-http	Major
@effect/rpc	Major
@effect/schema	Major
@effect/sql-d1	Major
@effect/sql-drizzle	Major
@effect/sql-mssql	Major
@effect/sql-mysql2	Major
@effect/sql-pg	Major
@effect/sql-sqlite-bun	Major
@effect/sql-sqlite-node	Major
@effect/sql-sqlite-react-native	Major
@effect/sql-sqlite-wasm	Major
@effect/sql	Major
@effect/typeclass	Major
@effect/vitest	Major

Not sure what this means? Click here to learn what changesets are.

Click here if you're a maintainer who wants to add another changeset to this PR

[tim-smart]

What is the benefit over flatMap(f, { switch: true })?

[mikearnaldi]

I guess just verbosity, not sure if this is common enough to justify it

[dilame]

RxJS also do have a switchMap operator, and it's one of the most important and widely used operator.

I decided to make it as a separate operator not just because of verbosity, but because of clarity and visualization.

This is a real code from my project

export const BinanceWsUnitSingleLayer = Layer.effect(
  BinanceWsUnitTag,
  Effect.gen(function* () {
    // ....................................................................
    return {
      createStream(
        stream: string,
      ): Stream.Stream<
        unknown,
        | BinanceWsConnectionError
        | BinanceWsResponseError
        | BinanceWsConnectionCloseError
      > {
        return Stream.suspend(() => {
          debug(`📤Requesting stream "${stream}"`);
          let stream$ = streams.get(stream);
          if (stream$) {
            return stream$;
          }
          stream$ = connection.pipe(
            Stream.flatMap(
              (ws) =>
                Effect.request(
                  BinanceWsSubscribeRequest({ stream, ws }),
                  RequestResolver.contextFromEffect(subscribe),
                ).pipe(
                  Stream.unwrap,
                  Stream.ensuring(
                    Effect.gen(function* () {
                      debug(`🔻 Unsubscribing ${stream}`);
                      streams.delete(stream);
                      yield* Effect.request(
                        BinanceWsUnsubscribeRequest({ stream, ws }),
                        RequestResolver.contextFromEffect(unsubscribe),
                      );
                    }),
                  ),
                ),
              { switch: true, concurrency: 1 },
            ),
          );
          streams.set(stream, stream$);
          return stream$;
        });
      },
    } satisfies BinanceWsUnit;
  }),
)

When you fluently read this code, it is not really easy to notice the options parameter for flatMap, you only see Stream.flatMap, but default flatMap behaviour is different. In addition, there are no clear documentation about what does switch option means. On the site it is written

if the order of concatenation is not important for us, we can use the switch option.

And it doesn't describes this concept.

Actually, i thinks that the switch option changes the behaviour dramatically enough to extract it to the separate operator with it's own documentation and example, which i also provided in this PR.