feature: change to stream-based implementation for greated flexibility

Author: Datner

src/RuntimeProvider.tsx

@@ -1,43 +1,39 @@
 "use client"
 import { pipe } from "@effect/data/Function"
-import * as Option from "@effect/data/Option"
 import * as Effect from "@effect/io/Effect"
 import * as Layer from "@effect/io/Layer"
-import * as LogLevel from "@effect/io/Logger/Level"
 import type * as Runtime from "@effect/io/Runtime"
 import * as Scope from "@effect/io/Scope"
 import { createContext } from "react"
-import type { UseEffectCallback } from "react-effect/hooks/useEffectCallback"
-import { makeUseEffectCallback } from "react-effect/hooks/useEffectCallback"
-import type { UseEffectResult } from "react-effect/hooks/useEffectResult"
-import { makeUseEffectResult } from "react-effect/hooks/useEffectResult"
+import type { UseResult } from "react-effect/hooks/useResult"
+import { makeUseResult } from "react-effect/hooks/useResult"
+import type { UseResultCallback } from "react-effect/hooks/useResultCallback"
+import { makeUseResultCallback } from "react-effect/hooks/useResultCallback"
 
 export { RuntimeContext } from "react-effect/internal/runtimeContext"
 
-export interface ReactEffectBag<R, D> {
-  readonly RuntimeContext: React.Context<Effect.Effect<never, D, Runtime.Runtime<R>>>
-  readonly useEffectCallback: UseEffectCallback<R, D>
-  readonly useEffectResult: UseEffectResult<R, D>
+export interface ReactEffectBag<R> {
+  readonly RuntimeContext: React.Context<Runtime.Runtime<R>>
+  readonly useResultCallback: UseResultCallback<R>
+  readonly useResult: UseResult<R>
 }
 
 export const makeFromLayer = <R, E>(
   layer: Layer.Layer<never, E, R>
-): ReactEffectBag<R, E> => {
+): ReactEffectBag<R> => {
   const scope = Effect.runSync(Scope.make())
 
   const runtime = pipe(
     Layer.toRuntime(layer),
     Effect.provideService(Scope.Scope, scope),
-    Effect.withUnhandledErrorLogLevel(Option.some(LogLevel.Fatal)),
-    Effect.cached,
     Effect.runSync
   )
 
   const RuntimeContext = createContext(runtime)
 
   return {
     RuntimeContext,
-    useEffectCallback: makeUseEffectCallback(RuntimeContext),
-    useEffectResult: makeUseEffectResult(RuntimeContext)
+    useResultCallback: makeUseResultCallback(RuntimeContext),
+    useResult: makeUseResult(RuntimeContext)
   }
 }

src/hooks/useEffectCallback.ts

@@ -0,0 +1,50 @@
+import * as Effect from "@effect/io/Effect"
+import * as Fiber from "@effect/io/Fiber"
+import * as Runtime from "@effect/io/Runtime"
+import type * as Schedule from "@effect/io/Schedule"
+import * as Stream from "@effect/stream/Stream"
+import { useContext, useRef, useState } from "react"
+import type { ResultBag } from "react-effect/hooks/useResultBag"
+import { updateNext, useResultBag } from "react-effect/hooks/useResultBag"
+import type { RuntimeContext } from "react-effect/internal/runtimeContext"
+import * as Result from "react-effect/Result"
+
+export type UseResult<R> = <R0 extends R, E, A>(
+  effect: Effect.Effect<R0, E, A>
+) => ResultBag<E, A>
+
+export const makeUseResult: <R>(
+  runtimeContext: RuntimeContext<R>
+) => UseResult<R> = <R>(runtimeContext: RuntimeContext<R>) =>
+  <R0 extends R, E, A>(stream: Stream.Stream<R0, E, A>) => {
+    const runtime = useContext(runtimeContext)
+    const prevRef = useRef<Stream.Stream<R0, E, A>>()
+    const fiberRef = useRef<Fiber.RuntimeFiber<E, void>>()
+    const [result, setResult] = useState<Result.Result<E, A>>(Result.waiting(Result.initial()))
+    const [trackRef, resultBag] = useResultBag(result)
+
+    if (prevRef.current !== stream) {
+      prevRef.current = stream
+      if (fiberRef.current) {
+        Effect.runSync(Fiber.interruptFork(fiberRef.current))
+      }
+      fiberRef.current = stream.pipe(
+        Stream.tap((value) =>
+          Effect.sync(() => {
+            setResult(updateNext(Result.success(value), trackRef))
+          })
+        ),
+        Stream.tapErrorCause((cause) =>
+          Effect.sync(() => {
+            setResult(updateNext(Result.failCause(cause), trackRef))
+          })
+        ),
+        Stream.runDrain,
+        Runtime.runFork(runtime)
+      )
+    }
+
+    trackRef.current.currentStatus = result._tag
+
+    return resultBag
+  }

src/hooks/useEffectResult.ts

@@ -7,13 +7,13 @@ import type { MutableRefObject } from "react"
 import { useMemo, useRef } from "react"
 import * as Result from "react-effect/Result"
 
-export interface ResultBag<D, E, A> {
-  readonly result: Result.Result<D, E, A>
+export interface ResultBag<E, A> {
+  readonly result: Result.Result<E, A>
   readonly dataUpdatedAt: Option.Option<Date>
   readonly errorUpdatedAt: Option.Option<Date>
   readonly errorRunningCount: number
   readonly failureCount: number
-  readonly failureCause: Cause.Cause<D | E>
+  readonly failureCause: Cause.Cause<E>
   readonly isError: boolean
   readonly isRefreshing: boolean
   readonly isRetrying: boolean
@@ -36,7 +36,7 @@ export interface TrackedProperties {
   runningErrorCount: number
   invocationCount: number
   interruptCount: number
-  currentStatus: Result.Result<any, any, any>["_tag"]
+  currentStatus: Result.Result<any, any>["_tag"]
   failureCause: Cause.Cause<unknown>
 }
 
@@ -60,28 +60,31 @@ const optionDateGreaterThan = pipe(
   Order.greaterThan
 )
 
-export const updateNext = <D, E, A>(
-  next: Result.Result<D, E, A>,
+export const updateNext = <E, A>(
+  next: Result.Result<E, A>,
   ref: MutableRefObject<TrackedProperties>
-): Result.Result<D, E, A> => {
+): Result.Result<E, A> => {
   switch (next._tag) {
     case "Initial": {
       break
     }
     case "Waiting": {
       break
     }
-    case "Defect": {
-      ref.current.currentDefectCount++
-      ref.current.currentErrorCount++
-      ref.current.runningErrorCount++
-      break
-    }
-    case "Fail": {
-      ref.current.currentFailureCount++
-      ref.current.currentErrorCount++
-      ref.current.runningErrorCount++
-      ref.current.errorUpdatedAt = Option.some(new Date())
+    case "Failure": {
+      if (Cause.isFailure(next.cause)) {
+        ref.current.currentFailureCount++
+        ref.current.currentErrorCount++
+        ref.current.runningErrorCount++
+        ref.current.errorUpdatedAt = Option.some(new Date())
+        break
+      }
+      if (!Cause.isInterruptedOnly(next.cause)) {
+        ref.current.currentDefectCount++
+        ref.current.currentErrorCount++
+        ref.current.runningErrorCount++
+      }
+
       break
     }
     case "Success": {
@@ -94,10 +97,10 @@ export const updateNext = <D, E, A>(
   return next
 }
 
-export const useResultBag = <D, E, A>(result: Result.Result<D, E, A>) => {
+export const useResultBag = <E, A>(result: Result.Result<E, A>) => {
   const trackedPropsRef = useRef<TrackedProperties>(initial)
 
-  const resultBag = useMemo((): ResultBag<D, E, A> => ({
+  const resultBag = useMemo((): ResultBag<E, A> => ({
     result,
     get isLoading() {
       return Result.isLoading(result)
@@ -131,7 +134,7 @@ export const useResultBag = <D, E, A>(result: Result.Result<D, E, A>) => {
       return trackedPropsRef.current.currentFailureCount
     },
     get failureCause() {
-      return trackedPropsRef.current.failureCause as Cause.Cause<D | E>
+      return trackedPropsRef.current.failureCause as Cause.Cause<E>
     },
     get errorRunningCount() {
       return trackedPropsRef.current.runningErrorCount

src/hooks/useResult.ts

@@ -0,0 +1,66 @@
+import * as Effect from "@effect/io/Effect"
+import * as Fiber from "@effect/io/Fiber"
+import * as Queue from "@effect/io/Queue"
+import * as Runtime from "@effect/io/Runtime"
+import * as Stream from "@effect/stream/Stream"
+import { useCallback, useContext, useRef, useState } from "react"
+import type { ResultBag } from "react-effect/hooks/useResultBag"
+import { updateNext, useResultBag } from "react-effect/hooks/useResultBag"
+import type { RuntimeContext } from "react-effect/internal/runtimeContext"
+import * as Result from "react-effect/Result"
+
+export type UseResultCallback<R> = <Args extends Array<any>, R0 extends R, E, A>(
+  callback: (...args: Args) => Effect.Effect<R0, E, A>
+) => readonly [ResultBag<E, A>, (...args: Args) => void, Queue.Queue<Args>]
+
+export const makeUseResultCallback: <R>(
+  runtimeContext: RuntimeContext<R>
+) => UseResultCallback<R> = <R>(
+  runtimeContext: RuntimeContext<R>
+) =>
+  <Args extends Array<any>, R0 extends R, E, A>(f: (...args: Args) => Stream.Stream<R0, E, A>) => {
+    const runtime = useContext(runtimeContext)
+    const fiberRef = useRef<Fiber.RuntimeFiber<E, void>>()
+    const prevRef = useRef<typeof f>()
+    const queueRef = useRef<Queue.Queue<Args>>()
+    if (!queueRef.current) {
+      queueRef.current = Effect.runSync(Queue.unbounded())
+    }
+    const [result, setResult] = useState<Result.Result<E, A>>(Result.initial())
+    const [trackRef, resultBag] = useResultBag(result)
+
+    if (prevRef.current !== f) {
+      prevRef.current = f
+      if (fiberRef.current) {
+        Effect.runSync(Fiber.interruptFork(fiberRef.current))
+      }
+      fiberRef.current = Stream.fromQueue(queueRef.current).pipe(
+        Stream.tap(() =>
+          Effect.sync(() => {
+            setResult((prev) => updateNext(Result.waiting(prev), trackRef))
+          })
+        ),
+        Stream.flatMap((args) => f(...args)),
+        Stream.tap((value) =>
+          Effect.sync(() => {
+            setResult(updateNext(Result.success(value), trackRef))
+          })
+        ),
+        Stream.tapErrorCause((cause) =>
+          Effect.sync(() => {
+            setResult(updateNext(Result.failCause(cause), trackRef))
+          })
+        ),
+        Stream.runDrain,
+        Runtime.runFork(runtime)
+      )
+    }
+    trackRef.current.currentStatus = result._tag
+
+    const run = useCallback((...args: Args) => {
+      trackRef.current.invocationCount++
+      queueRef.current!.unsafeOffer(args)
+    }, [])
+
+    return [resultBag, run, queueRef.current!] as const
+  }

src/hooks/useResultBag.ts

@@ -1,7 +1,4 @@
-import type * as Effect from "@effect/io/Effect"
 import type * as Runtime from "@effect/io/Runtime"
 
 /* @internal */
-export type RuntimeContext<R, E> = React.Context<
-  Effect.Effect<never, E, Runtime.Runtime<R>>
->
+export type RuntimeContext<R> = React.Context<Runtime.Runtime<R>>