useSubscribeStream

packages/example/src/routes/tests.tsx

@@ -25,6 +25,8 @@ function RouteComponent() {
     const uuidRef = R.useRef("none")
     const anotherRef = R.useRef(69)
 
+    console.log(R.useSubscribeStream(uuidRef.changes))
+
 
     const logValue = R.useCallbackSync(Effect.fn(function*(value: string) {
         yield* Effect.log(value)

packages/reffuse/src/ReffuseNamespace.ts

@@ -1,4 +1,4 @@
-import { Array, type Context, Effect, ExecutionStrategy, Exit, type Fiber, type Layer, pipe, Pipeable, Queue, Ref, Runtime, Scope, Stream, SubscriptionRef } from "effect"
+import { Array, type Context, Effect, ExecutionStrategy, Exit, type Fiber, type Layer, Option, pipe, Pipeable, Queue, Ref, Runtime, Scope, Stream, SubscriptionRef } from "effect"
 import * as React from "react"
 import * as ReffuseContext from "./ReffuseContext.js"
 import * as ReffuseRuntime from "./ReffuseRuntime.js"
@@ -448,6 +448,21 @@ export abstract class ReffuseNamespace<R> {
         return stream
     }
 
+    useSubscribeStream<A, InitialA extends A | undefined, E, R>(
+        this: ReffuseNamespace<R>,
+        stream: Stream.Stream<A, E, R>,
+        initialValue?: InitialA,
+    ): InitialA extends A ? Option.Some<A> : Option.Option<A> {
+        const [reactStateValue, setReactStateValue] = React.useState<Option.Option<A>>(Option.fromNullable(initialValue))
+
+        this.useFork(() => Stream.runForEach(
+            Stream.changesWith(stream, (x, y) => x === y),
+            v => Effect.sync(() => setReactStateValue(Option.some(v))),
+        ), [stream])
+
+        return reactStateValue as InitialA extends A ? Option.Some<A> : Option.Option<A>
+    }
+
 
     SubscribeRefs<
         const Refs extends readonly SubscriptionRef.SubscriptionRef<any>[],