diff --git a/packages/extension-query/src/FailureHandler.ts b/packages/extension-query/src/FailureHandler.ts index 96d784b..e1f33ad 100644 --- a/packages/extension-query/src/FailureHandler.ts +++ b/packages/extension-query/src/FailureHandler.ts @@ -1,4 +1,4 @@ -import { Effect, type Queue, type Stream } from "effect" +import { type Context, Effect, Layer, Queue, Stream } from "effect" export interface FailureHandler { @@ -9,3 +9,12 @@ export interface FailureHandler { export const Tag = (id: Id) => < Self, E = never, >() => Effect.Tag(id)>() + +export const layer = ( + tag: Context.TagClass> +): Layer.Layer => Layer.effect(tag, Queue.unbounded().pipe( + Effect.map(queue => ({ + failures: Stream.fromQueue(queue), + queue, + })) +)) diff --git a/packages/extension-query/src/QueryRunner.ts b/packages/extension-query/src/QueryRunner.ts index a403672..c1401ba 100644 --- a/packages/extension-query/src/QueryRunner.ts +++ b/packages/extension-query/src/QueryRunner.ts @@ -54,22 +54,23 @@ export const make = ( })) ) - const forkFetch = interrupt.pipe( - Effect.andThen( - Ref.set(stateRef, AsyncData.loading()).pipe( - Effect.andThen(latestKeyRef), - Effect.flatMap(identity), - Effect.flatMap(key => query(key).pipe( - Effect.matchCauseEffect({ - onSuccess: v => Ref.set(stateRef, AsyncData.success(v)), - onFailure: c => Ref.set(stateRef, AsyncData.failure(c)), - }) - )), + const run = latestKeyRef.pipe( + Effect.flatMap(identity), + Effect.flatMap(key => query(key).pipe( + Effect.matchCauseEffect({ + onSuccess: v => Ref.set(stateRef, AsyncData.success(v)), + onFailure: c => Ref.set(stateRef, AsyncData.failure(c)), + }) + )), - Effect.provide(context), - Effect.fork, - ) - ), + Effect.provide(context), + ) + + const forkFetch = interrupt.pipe( + Effect.andThen(Ref.set(stateRef, AsyncData.loading()).pipe( + Effect.andThen(run), + Effect.fork, + )), Effect.flatMap(fiber => Ref.set(fiberRef, Option.some(fiber)).pipe( @@ -82,27 +83,16 @@ export const make = ( ) const forkRefresh = interrupt.pipe( - Effect.andThen( - Ref.update(stateRef, previous => { - if (AsyncData.isSuccess(previous) || AsyncData.isFailure(previous)) - return AsyncData.refreshing(previous) - if (AsyncData.isRefreshing(previous)) - return AsyncData.refreshing(previous.previous) - return AsyncData.loading() - }).pipe( - Effect.andThen(latestKeyRef), - Effect.flatMap(identity), - Effect.flatMap(key => query(key).pipe( - Effect.matchCauseEffect({ - onSuccess: v => Ref.set(stateRef, AsyncData.success(v)), - onFailure: c => Ref.set(stateRef, AsyncData.failure(c)), - }) - )), - - Effect.provide(context), - Effect.fork, - ) - ), + Effect.andThen(Ref.update(stateRef, previous => { + if (AsyncData.isSuccess(previous) || AsyncData.isFailure(previous)) + return AsyncData.refreshing(previous) + if (AsyncData.isRefreshing(previous)) + return AsyncData.refreshing(previous.previous) + return AsyncData.loading() + }).pipe( + Effect.andThen(run), + Effect.fork, + )), Effect.flatMap(fiber => Ref.set(fiberRef, Option.some(fiber)).pipe(