Function asStreamResult

packages/vue/src/mutate.ts:223–263

(
  handler: (...args: Args) => Stream.Stream<A, E, R>
)

        ))
      )

  return tuple(computed(() => state.value), act)
}

/**
 * Like `asResult`, but for streams. The ref is updated with each emitted value
 * (keeping `waiting: true`) and is finalised (with `waiting: false`) once the
 * stream terminates successfully. Errors are surfaced as `AsyncResult.failure`.
 */
export const asStreamResult = <Args extends readonly any[], A, E, R>(
  handler: (...args: Args) => Stream.Stream<A, E, R>
): readonly [ComputedRef<AsyncResult.AsyncResult<A, E>>, (...args: Args) => Effect.Effect<void, never, R>] => {
  const state = shallowRef<AsyncResult.AsyncResult<A, E>>(AsyncResult.initial())

  const runStream = (stream: Stream.Stream<A, E, R>): Effect.Effect<void, never, R> =>
    Effect
      .sync(() => {
        state.value = AsyncResult.initial(true)
      })
      .pipe(
        Effect.andThen(
          stream.pipe(
            Stream.runForEach((value) =>
              Effect.sync(() => {
                state.value = AsyncResult.success(value, { waiting: true })
              })
            ),
            Effect.exit,
            Effect.flatMap((exit) =>
              Effect.sync(() => {
                if (exit._tag === "Success") {
                  const current = state.value
                  if (AsyncResult.isSuccess(current)) {
                    state.value = AsyncResult.success(current.value, { waiting: false })
                  } else {
                    state.value = AsyncResult.initial(false)
                  }
                } else {
                  state.value = AsyncResult.failure(exit.cause)
                }
              })
            )
          )
        )