cleanup

Author: tim-smart

src/hooks/useResultCallback.ts

@@ -1,3 +1,4 @@
+import { pipe } from "@effect/data/Function"
 import * as Effect from "@effect/io/Effect"
 import type * as Fiber from "@effect/io/Fiber"
 import * as Ref from "@effect/io/Ref"
@@ -37,7 +38,7 @@ class FiberStore<R, E, A> {
   setResult(result: Result.Result<E, A>) {
     this.result = result
     updateTrackedProps(result, this.trackedProps)
-    this.resultBag = makeResultBag(this.result, this.trackedProps)
+    this.resultBag = makeResultBag(result, this.trackedProps)
 
     for (let i = 0; i < this.listeners.length; i++) {
       this.listeners[i]()
@@ -49,17 +50,20 @@ class FiberStore<R, E, A> {
 
   fiberState:
     | {
-      readonly fiber: Fiber.RuntimeFiber<E, void>
+      readonly fiber: Fiber.RuntimeFiber<never, void>
       readonly interruptedRef: Ref.Ref<boolean>
     }
     | undefined = undefined
 
   interruptIfRunning() {
     if (this.fiberState) {
-      Effect.runSync(Ref.set(this.fiberState.interruptedRef, true))
       Effect.runFork(
-        this.fiberState.fiber.interruptAsFork(this.fiberState.fiber.id())
+        Effect.zipRight(
+          Ref.set(this.fiberState.interruptedRef, true),
+          this.fiberState.fiber.interruptAsFork(this.fiberState.fiber.id())
+        )
       )
+      this.fiberState = undefined
     }
   }
 
@@ -73,18 +77,18 @@ class FiberStore<R, E, A> {
         (interrupted) =>
           interrupted
             ? Effect.unit
-            : Effect.sync(() => {
-              this.setResult(result)
-            })
+            : Effect.sync(() => this.setResult(result))
       )
 
-    const fiber = Stream.suspend(() => {
-      this.setResult(Result.waiting(this.result))
-      return stream
-    }).pipe(
-      Stream.tap((value) => maybeSetResult(Result.success(value))),
-      Stream.tapErrorCause((cause) => maybeSetResult(Result.failCause(cause))),
-      Stream.runDrain,
+    const fiber = pipe(
+      Stream.runForEach(
+        Stream.suspend(() => {
+          this.setResult(Result.waiting(this.result))
+          return stream
+        }),
+        (_) => maybeSetResult(Result.success(_))
+      ),
+      Effect.catchAllCause((cause) => maybeSetResult(Result.failCause(cause))),
       Runtime.runFork(this.runtime)
     )
 
@@ -108,16 +112,16 @@ export const makeUseResultCallback: <R>(
     f: (...args: Args) => Stream.Stream<R0, E, A>
   ) => {
     const runtime = useContext(runtimeContext)
-    const storeRef = useRef<FiberStore<R0, E, A> | undefined>(undefined)
+    const storeRef = useRef<FiberStore<R0, E, A>>(undefined as any)
     if (storeRef.current === undefined) {
       storeRef.current = new FiberStore(runtime)
     }
     const resultBag = useSyncExternalStore(
-      storeRef.current!.subscribe,
-      storeRef.current!.snapshot
+      storeRef.current.subscribe,
+      storeRef.current.snapshot
     )
     const run = useCallback((...args: Args) => {
-      storeRef.current!.run(f(...args))
+      storeRef.current.run(f(...args))
     }, [f])
     return [resultBag, run] as const
   }