Cleanup

packages/example/src/VQueryErrorHandler.tsx

@@ -1,7 +1,7 @@
 import { AlertDialog, Button, Flex, Text } from "@radix-ui/themes"
 import { Cause, Console, Effect, Either, flow, Match, Option, Stream } from "effect"
 import { useState } from "react"
-import { AppQueryErrorHandler } from "./query"
+import { AppQueryClient } from "./query"
 import { R } from "./reffuse"
 
 
@@ -9,8 +9,8 @@ export function VQueryErrorHandler() {
     const [open, setOpen] = useState(false)
 
     const error = R.useSubscribeStream(
-        R.useMemo(() => AppQueryErrorHandler.pipe(
-            Effect.map(handler => handler.errors.pipe(
+        R.useMemo(() => AppQueryClient.pipe(
+            Effect.map(client => client.errorHandler.errors.pipe(
                 Stream.changes,
                 Stream.tap(Console.error),
                 Stream.tap(() => Effect.sync(() => setOpen(true))),

packages/example/src/reffuse.ts

@@ -3,12 +3,11 @@ import { Clipboard, Geolocation, Permissions } from "@effect/platform-browser"
 import { LazyRefExtension } from "@reffuse/extension-lazyref"
 import { QueryExtension } from "@reffuse/extension-query"
 import { Reffuse, ReffuseContext } from "reffuse"
-import { AppQueryClient, AppQueryErrorHandler } from "./query"
+import { AppQueryClient } from "./query"
 
 
 export const RootContext = ReffuseContext.make<
     | AppQueryClient
-    | AppQueryErrorHandler
     | Clipboard.Clipboard
     | Geolocation.Geolocation
     | Permissions.Permissions

packages/extension-query/src/QueryErrorHandler.ts

@@ -1,4 +1,4 @@
-import { Cause, Context, Effect, identity, Layer, Queue, Stream } from "effect"
+import { Cause, Context, Effect, identity, Layer, PubSub, Stream } from "effect"
 import type { Mutable } from "effect/Types"
 
 
@@ -36,17 +36,17 @@ export const Service = <Self, HandledE = never>() => (
         const TagClass = Context.Tag(id)() as ServiceResult<Self, Id, FallbackA, HandledE>
 
         (TagClass as Mutable<typeof TagClass>).Live = Layer.effect(TagClass, Effect.gen(function*() {
-            const queue = yield* Queue.unbounded<Cause.Cause<HandledE>>()
-            const errors = Stream.fromQueue(queue)
+            const pubsub = yield* PubSub.unbounded<Cause.Cause<HandledE>>()
+            const errors = Stream.fromPubSub(pubsub)
 
             const handle = <A, E, R>(
                 self: Effect.Effect<A, E, R>
             ): Effect.Effect<A | FallbackA, Exclude<E, HandledE>, R> => f(
                 self as unknown as Effect.Effect<never, HandledE, never>,
-                (failure: HandledE) => Queue.offer(queue, Cause.fail(failure)).pipe(
+                (failure: HandledE) => PubSub.publish(pubsub, Cause.fail(failure)).pipe(
                     Effect.andThen(Effect.failCause(Cause.empty))
                 ),
-                (defect: unknown) => Queue.offer(queue, Cause.die(defect)).pipe(
+                (defect: unknown) => PubSub.publish(pubsub, Cause.die(defect)).pipe(
                     Effect.andThen(Effect.failCause(Cause.empty))
                 ),
             )

packages/reffuse/src/ReffuseNamespace.ts

@@ -1,4 +1,4 @@
-import { Chunk, type Context, Effect, ExecutionStrategy, Exit, type Fiber, flow, type Layer, Match, Option, pipe, Pipeable, Queue, Ref, Runtime, Scope, Stream, SubscriptionRef } from "effect"
+import { Chunk, type Context, Effect, ExecutionStrategy, Exit, type Fiber, flow, type Layer, Match, Option, pipe, Pipeable, PubSub, Ref, Runtime, Scope, Stream, SubscriptionRef } from "effect"
 import * as React from "react"
 import * as ReffuseContext from "./ReffuseContext.js"
 import * as ReffuseRuntime from "./ReffuseRuntime.js"
@@ -474,11 +474,26 @@ export abstract class ReffuseNamespace<R> {
         this: ReffuseNamespace<R>,
         values: A,
     ): Stream.Stream<A> {
-        const [queue, stream] = this.useMemo(() => Queue.unbounded<A>().pipe(
-            Effect.map(queue => [queue, Stream.fromQueue(queue)] as const)
-        ), [])
+        const scope = this.useScope()
+
+        const { latest, pubsub, stream } = this.useMemo(() => Effect.Do.pipe(
+            Effect.bind("latest", () => Ref.make(values)),
+            Effect.bind("pubsub", () => Effect.acquireRelease(PubSub.unbounded<A>(), PubSub.shutdown)),
+            Effect.let("stream", ({ latest, pubsub }) => Ref.get(latest).pipe(
+                Effect.flatMap(a => Effect.map(
+                    Stream.fromPubSub(pubsub, { scoped: true }),
+                    s => Stream.concat(Stream.make(a), s),
+                )),
+                Stream.unwrapScoped,
+            )),
+            Effect.provideService(Scope.Scope, scope),
+        ), [scope], { doNotReExecuteOnRuntimeOrContextChange: true })
+
+        this.useEffect(() => Ref.set(latest, values).pipe(
+            Effect.andThen(PubSub.publish(pubsub, values)),
+            Effect.unlessEffect(PubSub.isShutdown(pubsub)),
+        ), values, { doNotReExecuteOnRuntimeOrContextChange: true })
 
-        this.useEffect(() => Queue.offer(queue, values), values)
         return stream
     }