Function buildStreamQueryFamily

packages/vue/src/atomQuery.ts:495–522

(
  rt: AtomClientRuntime,
  self: {
    readonly id: string
    readonly handler: (i: I) => Stream.Stream<A, E, any>
    readonly options?: ClientForOptions
    readonly queryKeyProjectionHash?: string
  }
)

        const recordReads = Effect
          .gen(function*() {
            const readsRef = yield* Ref.make(DataDependencies.empty())
            const writesRef = yield* Ref.make(DataDependencies.empty())
            const recorder = DataDependencies.makeDataDependencyRecorder(readsRef, writesRef)
            const result = yield* self
              .handler(input)
              .pipe(Effect.provideService(DataDependencies.DataDependencyRecorder, recorder))
            lastReads = yield* Ref.get(readsRef)
            setQueryReadDependencies(fullKey, lastReads)
            return result
          })
        let liveFetch = false
        const effect = Effect
          .gen(function*() {
            liveFetch = yield* Effect.promise(() => beginLiveQueryFetch(fullKey))
            return yield* recordReads.pipe(Effect.retry({ times: 5, while: isRetryable }))
          })
          .pipe(
            Effect.ensuring(Effect.sync(() => endLiveQueryFetch(liveFetch))),
            Effect.tapCauseIf(Cause.hasDies, (cause) => reportRuntimeError(cause)),
            // On exit, the compute is no longer in-flight. An interrupt (subscriber lost interest / a
            // superseding refresh) may leave the result at `waiting`; with `inFlight` back at 0 that
            // reads as "stuck", so the next mount recovers it (`recoverStuckWaitingOnMount`). We do not
            // re-fire here — the interrupt was intentional; recovery is driven by a genuine (re)mount.
            Effect.onExit(() =>
              Effect.sync(() => {
                fetchState.inFlight = Math.max(0, fetchState.inFlight - 1)
              })
            ),
            Effect.withSpan(`query ${self.id}`, {}, { captureStackTrace: false })
          )
        const parentSpan = takeAtomQueryParentSpan(atom)