packages/extension-query/src/QueryRunner.ts

@@ -5,11 +5,11 @@ import { Effect, Fiber, identity, Option, Ref, SubscriptionRef } from "effect"
 export interface QueryRunner<A, E, R> {
     readonly queryRef: SubscriptionRef.SubscriptionRef<Effect.Effect<A, E, R>>
     readonly stateRef: SubscriptionRef.SubscriptionRef<AsyncData.AsyncData<A, E>>
-    readonly fiberRef: SubscriptionRef.SubscriptionRef<Option.Option<Fiber.Fiber<void>>>
+    readonly fiberRef: SubscriptionRef.SubscriptionRef<Option.Option<Fiber.RuntimeFiber<void>>>
 
-    readonly interrupt: Effect.Effect<void>
+    readonly forkInterrupt: Effect.Effect<Fiber.RuntimeFiber<void>>
-    readonly forkFetch: Effect.Effect<void>
+    readonly forkFetch: Effect.Effect<Fiber.RuntimeFiber<void>>
-    readonly forkRefetch: Effect.Effect<void>
+    readonly forkRefresh: Effect.Effect<Fiber.RuntimeFiber<void>>
 }
 
 
@@ -20,66 +20,80 @@ export const make = <A, E, R>(
 
     const queryRef = yield* SubscriptionRef.make(query)
     const stateRef = yield* SubscriptionRef.make(AsyncData.noData<A, E>())
-    const fiberRef = yield* SubscriptionRef.make(Option.none<Fiber.Fiber<void>>())
+    const fiberRef = yield* SubscriptionRef.make(Option.none<Fiber.RuntimeFiber<void>>())
 
     const interrupt = fiberRef.pipe(
         Effect.flatMap(Option.match({
-            onSome: fiber => Ref.set(fiberRef, Option.none()).pipe(
+            onSome: Fiber.interrupt,
-                Effect.andThen(Fiber.interrupt(fiber))
-            ),
             onNone: () => Effect.void,
         }))
     )
 
-    const forkInterrupt = fiberRef.pipe(
+    const forkInterrupt = Effect.forkDaemon(interrupt)
-        Effect.flatMap(Option.match({
-            onSome: fiber => Ref.set(fiberRef, Option.none()).pipe(
-                Effect.andThen(Fiber.interruptFork(fiber))
-            ),
-            onNone: () => Effect.void,
-        }))
-    )
 
     const forkFetch = interrupt.pipe(
-        Effect.andThen(Ref.set(stateRef, AsyncData.loading())),
+        Effect.andThen(
-        Effect.andThen(queryRef.pipe(Effect.flatMap(identity))),
+            Effect.addFinalizer(() => Ref.set(fiberRef, Option.none())).pipe(
-        Effect.matchCauseEffect({
+                Effect.andThen(Ref.set(stateRef, AsyncData.loading())),
-            onSuccess: v => Ref.set(stateRef, AsyncData.success(v)),
+                Effect.andThen(queryRef.pipe(Effect.flatMap(identity))),
-            onFailure: c => Ref.set(stateRef, AsyncData.failure(c)),
+                Effect.matchCauseEffect({
-        }),
+                    onSuccess: v => Ref.set(stateRef, AsyncData.success(v)),
-        Effect.andThen(Ref.set(fiberRef, Option.none())),
+                    onFailure: c => Ref.set(stateRef, AsyncData.failure(c)),
-        Effect.provide(context),
+                }),
-        Effect.forkDaemon,
 
-        Effect.flatMap(fiber => Ref.set(fiberRef, Option.some(fiber))),
+                Effect.provide(context),
+                Effect.scoped,
+                Effect.fork,
+            )
+        ),
+
+        Effect.flatMap(fiber =>
+            Ref.set(fiberRef, Option.some(fiber)).pipe(
+                Effect.andThen(Fiber.join(fiber))
+            )
+        ),
+
+        Effect.forkDaemon,
     )
 
-    const forkRefetch = interrupt.pipe(
+    const forkRefresh = interrupt.pipe(
-        Effect.andThen(Ref.update(stateRef, previous => {
+        Effect.andThen(
-            if (AsyncData.isSuccess(previous) || AsyncData.isFailure(previous))
+            Effect.addFinalizer(() => Ref.set(fiberRef, Option.none())).pipe(
-                return AsyncData.refreshing(previous)
+                Effect.andThen(Ref.update(stateRef, previous => {
-            if (AsyncData.isRefreshing(previous))
+                    if (AsyncData.isSuccess(previous) || AsyncData.isFailure(previous))
-                return AsyncData.refreshing(previous.previous)
+                        return AsyncData.refreshing(previous)
-            return AsyncData.loading()
+                    if (AsyncData.isRefreshing(previous))
-        })),
+                        return AsyncData.refreshing(previous.previous)
-        Effect.andThen(queryRef.pipe(Effect.flatMap(identity))),
+                    return AsyncData.loading()
-        Effect.matchCauseEffect({
+                })),
-            onSuccess: v => Ref.set(stateRef, AsyncData.success(v)),
+                Effect.andThen(queryRef.pipe(Effect.flatMap(identity))),
-            onFailure: c => Ref.set(stateRef, AsyncData.failure(c)),
+                Effect.matchCauseEffect({
-        }),
+                    onSuccess: v => Ref.set(stateRef, AsyncData.success(v)),
-        Effect.andThen(Ref.set(fiberRef, Option.none())),
+                    onFailure: c => Ref.set(stateRef, AsyncData.failure(c)),
-        Effect.provide(context),
+                }),
-        Effect.forkDaemon,
 
-        Effect.flatMap(fiber => Ref.set(fiberRef, Option.some(fiber))),
+                Effect.provide(context),
+                Effect.scoped,
+                Effect.fork,
+            )
+        ),
+
+        Effect.flatMap(fiber =>
+            Ref.set(fiberRef, Option.some(fiber)).pipe(
+                Effect.andThen(Fiber.join(fiber))
+            )
+        ),
+
+        Effect.forkDaemon,
     )
 
     return {
         queryRef,
         stateRef,
         fiberRef,
-        interrupt,
+
+        forkInterrupt,
         forkFetch,
-        forkRefetch,
+        forkRefresh,
     }
 })

packages/extension-query/src/index.ts

@@ -1,5 +1,5 @@
 import * as AsyncData from "@typed/async-data"
-import { Console, Effect, Ref, Stream, SubscriptionRef } from "effect"
+import { Effect, Fiber, Ref, SubscriptionRef } from "effect"
 import * as React from "react"
 import { ReffuseExtension, type ReffuseHelpers } from "reffuse"
 import * as QueryRunner from "./QueryRunner.js"
@@ -12,7 +12,7 @@ export interface UseQueryProps<A, E, R> {
 
 export interface UseQueryResult<A, E> {
     readonly state: SubscriptionRef.SubscriptionRef<AsyncData.AsyncData<A, E>>
-    readonly refresh: Effect.Effect<void>
+    readonly refresh: Effect.Effect<Fiber.RuntimeFiber<void>>
 }
 
 
@@ -23,16 +23,14 @@ export const QueryExtension = ReffuseExtension.make(() => ({
     ): UseQueryResult<A, E> {
         const runner = this.useMemo(() => QueryRunner.make(props.effect()), [])
 
-        this.useFork(() => Stream.runForEach(runner.fiberRef.changes, Console.log), [])
+        this.useEffect(() => Effect.addFinalizer(() => runner.forkInterrupt).pipe(
-
-        this.useEffect(() => Effect.addFinalizer(() => Effect.void).pipe(
             Effect.andThen(Ref.set(runner.queryRef, props.effect())),
             Effect.andThen(runner.forkFetch),
         ), [runner, ...props.deps])
 
         return React.useMemo(() => ({
             state: runner.stateRef,
-            refresh: runner.forkRefetch,
+            refresh: runner.forkRefresh,
         }), [runner])
     }
 }))