Refactor

packages/effect-fc/src/Component.ts

@@ -545,7 +545,7 @@ const runReactEffect = <E, R>(
     Effect.bind("exit", ({ scope }) => Effect.exit(Effect.provideService(f(), Scope.Scope, scope))),
     Effect.map(({ scope }) =>
         () => {
-            switch (options?.finalizerExecutionMode ?? "sync") {
+            switch (options?.finalizerExecutionMode ?? "fork") {
                 case "sync":
                     Runtime.runSync(runtime)(Scope.close(scope, Exit.void))
                     break

packages/effect-fc/src/Stream.ts

@@ -1,4 +1,4 @@
-import { Effect, Equivalence, Option, Stream } from "effect"
+import { Effect, Equivalence, Option, PubSub, Ref, type Scope, Stream } from "effect"
 import * as React from "react"
 import * as Component from "./Component.js"
 
@@ -25,9 +25,34 @@ export const useStream: {
             Stream.changesWith(stream, Equivalence.strict()),
             v => Effect.sync(() => setReactStateValue(Option.some(v))),
         )
-    ), [stream], { finalizerExecutionMode: "fork" })
+    ), [stream])
 
     return reactStateValue as Option.Some<A>
 })
 
+export const useStreamFromReactiveValues: {
+    <const A extends React.DependencyList>(
+        values: A
+    ): Effect.Effect<Stream.Stream<A>, never, Scope.Scope>
+} = Effect.fnUntraced(function* <const A extends React.DependencyList>(values: A) {
+    const { latest, pubsub, stream } = yield* Component.useOnMount(() => Effect.Do.pipe(
+        Effect.bind("latest", () => Ref.make(values)),
+        Effect.bind("pubsub", () => Effect.acquireRelease(PubSub.unbounded<A>(), PubSub.shutdown)),
+        Effect.let("stream", ({ latest, pubsub }) => latest.pipe(
+            Effect.flatMap(a => Effect.map(
+                Stream.fromPubSub(pubsub, { scoped: true }),
+                s => Stream.concat(Stream.make(a), s),
+            )),
+            Stream.unwrapScoped,
+        )),
+    ))
+
+    yield* Component.useReactEffect(() => Ref.set(latest, values).pipe(
+        Effect.andThen(PubSub.publish(pubsub, values)),
+        Effect.unlessEffect(PubSub.isShutdown(pubsub)),
+    ), values)
+
+    return stream
+})
+
 export * from "effect/Stream"

packages/effect-fc/src/Subscribable.ts

@@ -39,7 +39,7 @@ export const useSubscribables: {
         Stream.runForEach(v =>
             Effect.sync(() => setReactStateValue(v))
         ),
-    )), elements, { finalizerExecutionMode: "fork" })
+    )), elements)
 
     return reactStateValue as any
 })

packages/effect-fc/src/SubscriptionRef.ts

@@ -16,7 +16,7 @@ export const useSubscriptionRefState: {
             Stream.changesWith(ref.changes, Equivalence.strict()),
             v => Effect.sync(() => setReactStateValue(v)),
         )
-    ), [ref], { finalizerExecutionMode: "fork" })
+    ), [ref])
 
     const setValue = yield* Component.useCallbackSync((setStateAction: React.SetStateAction<A>) =>
         Effect.andThen(
@@ -38,7 +38,7 @@ export const useSubscriptionRefFromState: {
             Stream.changesWith(ref.changes, Equivalence.strict()),
             v => Effect.sync(() => setValue(v)),
         )
-    ), [setValue], { finalizerExecutionMode: "fork" })
+    ), [setValue])
     yield* Component.useReactEffect(() => Ref.set(ref, value), [value])
 
     return ref