Refactor Result

packages/effect-fc/src/Form.ts

@@ -65,7 +65,7 @@ export namespace make {
         readonly onSubmit: (
             this: Form<NoInfer<A>, NoInfer<I>, NoInfer<R>, unknown, unknown, unknown>,
             value: NoInfer<A>,
-        ) => Effect.Effect<SA, SE, Result.forkEffectPubSub.InputContext<SR, NoInfer<SP>>>
+        ) => Effect.Effect<SA, SE, Result.forkEffect.InputContext<SR, NoInfer<SP>>>
         readonly initialSubmitProgress?: SP
         readonly autosubmit?: boolean
         readonly debounce?: Duration.DurationInput
@@ -160,17 +160,15 @@ export const submit = <A, I, R, SA, SE, SR, SP>(
 > => Effect.whenEffect(
     self.valueRef.pipe(
         Effect.andThen(identity),
-        Effect.andThen(value => Result.forkEffectPubSub(
+        Effect.andThen(value => Result.forkEffect(
-            self.onSubmit(value) as Effect.Effect<SA, SE, Result.forkEffectPubSub.InputContext<SR, SP>>,
+            self.onSubmit(value) as Effect.Effect<SA, SE, Result.forkEffect.InputContext<SR, SP>>,
             { initialProgress: self.initialSubmitProgress },
         )),
-        Effect.andThen(identity),
+        Effect.andThen(([result]) => Stream.runFoldEffect(
-        Effect.andThen(Stream.fromQueue),
+            result.changes,
-        Stream.unwrapScoped,
-        Stream.runFoldEffect(
             Result.initial() as Result.Result<SA, SE, SP>,
             (_, result) => Effect.as(Ref.set(self.submitResultRef, result), result),
-        ),
+        )),
         Effect.tap(result => Result.isFailure(result)
             ? Option.match(
                 Chunk.findFirst(

packages/effect-fc/src/Query.ts

@@ -1,5 +1,5 @@
 import { Effect, Fiber, Option, Pipeable, Predicate, Stream, type Subscribable, type SubscriptionRef } from "effect"
-import type * as Result from "./Result.js"
+import * as Result from "./Result.js"
 
 
 export const QueryTypeId: unique symbol = Symbol.for("@effect-fc/Query/Query")
@@ -38,7 +38,8 @@ extends Pipeable.Class() implements Query<K, A, E, R, P> {
                 onSome: Fiber.interrupt,
                 onNone: () => Effect.void,
             })),
-            Effect.andThen(),
+            Effect.andThen(Result.forkEffect(this.f(key))),
+            Effect.tap(([result, fiber]) => ),
         )
     }
 }

packages/effect-fc/src/Result.ts

@@ -1,4 +1,4 @@
-import { Cause, Context, Data, Effect, Equal, Exit, Hash, Layer, Match, Option, Pipeable, Predicate, PubSub, pipe, type Queue, Ref, type Scope, type Subscribable, SubscriptionRef } from "effect"
+import { Cause, Context, Data, Effect, Equal, Exit, type Fiber, Hash, Layer, Match, Option, Pipeable, Predicate, PubSub, pipe, Ref, type Scope, Stream, Subscribable } from "effect"
 
 
 export const ResultTypeId: unique symbol = Symbol.for("@effect-fc/Result/Result")
@@ -186,78 +186,6 @@ export const makeProgressLayer = <A, E, P = never>(): Layer.Layer<
 }))
 
 
-export namespace forkEffectSubscriptionRef {
-    export type InputContext<R, P> = R extends Progress<infer X> ? [X] extends [P] ? R : never : R
-    export type OutputContext<R> = Scope.Scope | Exclude<R, Progress<any> | Progress<never>>
-
-    export interface Options<P> {
-        readonly initialProgress?: P
-    }
-}
-
-export const forkEffectSubscriptionRef = <A, E, R, P = never>(
-    effect: Effect.Effect<A, E, forkEffectSubscriptionRef.InputContext<R, NoInfer<P>>>,
-    options?: forkEffectSubscriptionRef.Options<P>,
-): Effect.Effect<
-    Subscribable.Subscribable<Result<A, E, P>>,
-    never,
-    forkEffectSubscriptionRef.OutputContext<R>
-> => Effect.tap(
-    SubscriptionRef.make<Result<A, E, P>>(initial()),
-    ref => Effect.forkScoped(State<A, E, P>().pipe(
-        Effect.andThen(state => state.set(running(options?.initialProgress)).pipe(
-            Effect.andThen(effect),
-            Effect.onExit(exit => state.set(fromExit(exit))),
-        )),
-        Effect.provide(Layer.empty.pipe(
-            Layer.provideMerge(makeProgressLayer<A, E, P>()),
-            Layer.provideMerge(Layer.succeed(State<A, E, P>(), {
-                get: ref,
-                set: v => Ref.set(ref, v),
-            })),
-        )),
-    )),
-) as Effect.Effect<Subscribable.Subscribable<Result<A, E, P>>, never, Scope.Scope>
-
-export namespace forkEffectPubSub {
-    export type InputContext<R, P> = R extends Progress<infer X> ? [X] extends [P] ? R : never : R
-    export type OutputContext<R> = Scope.Scope | Exclude<R, Progress<any> | Progress<never>>
-
-    export interface Options<P> {
-        readonly initialProgress?: P
-    }
-}
-
-export const forkEffectPubSub = <A, E, R, P = never>(
-    effect: Effect.Effect<A, E, forkEffectPubSub.InputContext<R, NoInfer<P>>>,
-    options?: forkEffectPubSub.Options<P>,
-): Effect.Effect<
-    Effect.Effect<Queue.Dequeue<Result<A, E, P>>, never, Scope.Scope>,
-    never,
-    forkEffectPubSub.OutputContext<R>
-> => Effect.all([
-    Ref.make<Result<A, E, P>>(initial()),
-    PubSub.unbounded<Result<A, E, P>>(),
-]).pipe(
-    Effect.tap(([ref, pubsub]) => Effect.forkScoped(State<A, E, P>().pipe(
-        Effect.andThen(state => state.set(running(options?.initialProgress)).pipe(
-            Effect.andThen(effect),
-            Effect.onExit(exit => Effect.andThen(
-                state.set(fromExit(exit)),
-                Effect.forkScoped(PubSub.shutdown(pubsub)),
-            )),
-        )),
-        Effect.provide(Layer.empty.pipe(
-            Layer.provideMerge(makeProgressLayer<A, E, P>()),
-            Layer.provideMerge(Layer.succeed(State<A, E, P>(), {
-                get: ref,
-                set: v => Effect.andThen(Ref.set(ref, v), PubSub.publish(pubsub, v))
-            })),
-        )),
-    ))),
-    Effect.map(([, pubsub]) => pubsub.subscribe),
-) as Effect.Effect<Effect.Effect<Queue.Dequeue<Result<A, E, P>>, never, Scope.Scope>, never, Scope.Scope>
-
 export namespace forkEffect {
     export type InputContext<R, P> = R extends Progress<infer X> ? [X] extends [P] ? R : never : R
     export type OutputContext<R> = Scope.Scope | Exclude<R, Progress<any> | Progress<never>>
@@ -270,11 +198,14 @@ export namespace forkEffect {
 export const forkEffect = <A, E, R, P = never>(
     effect: Effect.Effect<A, E, forkEffect.InputContext<R, NoInfer<P>>>,
     options?: forkEffect.Options<P>,
-) => Effect.all([
+): Effect.Effect<
-    Ref.make<Result<A, E, P>>(initial()),
+    readonly [result: Subscribable.Subscribable<Result<A, E, P>, never, never>, fiber: Fiber.Fiber<A, E>],
-    PubSub.unbounded<Result<A, E, P>>(),
+    never,
-]).pipe(
+    Scope.Scope | forkEffect.OutputContext<R>
-    Effect.tap(([ref, pubsub]) => Effect.forkScoped(State<A, E, P>().pipe(
+> => Effect.Do.pipe(
+    Effect.bind("ref", () => Ref.make<Result<A, E, P>>(initial())),
+    Effect.bind("pubsub", () => PubSub.unbounded<Result<A, E, P>>()),
+    Effect.bind("fiber", ({ ref, pubsub }) => Effect.forkScoped(State<A, E, P>().pipe(
         Effect.andThen(state => state.set(running(options?.initialProgress)).pipe(
             Effect.andThen(effect),
             Effect.onExit(exit => Effect.andThen(
@@ -290,5 +221,18 @@ export const forkEffect = <A, E, R, P = never>(
             })),
         )),
     ))),
-    Effect.map(([, pubsub]) => pubsub.subscribe),
+    Effect.map(({ ref, pubsub, fiber }) => [
-) as Effect.Effect<Effect.Effect<Queue.Dequeue<Result<A, E, P>>, never, Scope.Scope>, never, Scope.Scope>
+        Subscribable.make({
+            get: ref,
+            changes: Stream.unwrapScoped(Effect.map(
+                Effect.all([ref, Stream.fromPubSub(pubsub, { scoped: true })]),
+                ([latest, stream]) => Stream.concat(Stream.make(latest), stream),
+            )),
+        }),
+        fiber,
+    ]),
+) as Effect.Effect<
+    readonly [result: Subscribable.Subscribable<Result<A, E, P>, never, never>, fiber: Fiber.Fiber<A, E>],
+    never,
+    Scope.Scope | forkEffect.OutputContext<R>
+>