Function buildStream

packages/effect-app/src/client/apiClientFactory.ts:320–398

(input: any)

              )

            const buildStream = (input: any) => {
              const stream = Stream.unwrap(
                mr.contextEffect.pipe(
                  Effect.flatMap((svcs) =>
                    TheClient
                      .useSync((client) => {
                        const rpcStream = (client as any)[requestAttr]!(
                          Request.make(input)
                        ) as Stream.Stream<any, any, any>
                        return rpcStream.pipe(
                          // Collect server invalidation keys from the "done" chunk, then discard it.
                          Stream.tap((item: any) =>
                            item._tag === "done" || item._tag === "metadata"
                              ? Effect.gen(function*() {
                                const metadata = item.metadata as Invalidation.CommandMetaData
                                const invalidationKeys = yield* InvalidationKeysFromServer
                                yield* Effect.forEach(metadata.invalidateQueries, invalidationKeys.add, {
                                  discard: true
                                })
                                const dependencyRecorder = yield* DataDependencies.getDataDependencyRecorder
                                yield* Effect.forEach(
                                  metadata.dataDependencies.reads,
                                  dependencyRecorder.read,
                                  { discard: true }
                                )
                                yield* Effect.forEach(
                                  metadata.dataDependencies.writes,
                                  dependencyRecorder.write,
                                  { discard: true }
                                )
                              })
                              : Effect.void
                          ),
                          Stream.filter((item: any) => item._tag === "value"),
                          Stream.map((item: any) => item.value),
                          // V2: unwrap StreamFailureChunk — forward keys from failures too,
                          // then re-fail with the original error so callers see the unmodified
                          // error type.
                          Stream.catch((err: any) =>
                            err?._tag === "error" && err?.metadata
                              ? Stream.fromEffect(
                                Effect.gen(function*() {
                                  const metadata = err.metadata as Invalidation.CommandMetaData
                                  const invalidationKeys = yield* InvalidationKeysFromServer
                                  yield* Effect.forEach(metadata.invalidateQueries, invalidationKeys.add, {
                                    discard: true
                                  })
                                  const dependencyRecorder = yield* DataDependencies.getDataDependencyRecorder
                                  yield* Effect.forEach(
                                    metadata.dataDependencies.writes,
                                    dependencyRecorder.write,
                                    { discard: true }
                                  )
                                  return yield* Effect.fail(err.error)
                                })
                              )
                              : Stream.fail(err)
                          ),