From 7a12abdbdfc2056a97ee1b4a53a63ac3379b2ca3 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Julien=20Valverd=C3=A9?= <julien.valverde@mailo.com>
Date: Sun, 13 Apr 2025 02:30:29 +0200
Subject: [PATCH] useSubscribeStream

---
 packages/example/src/routes/tests.tsx    |  2 ++
 packages/reffuse/src/ReffuseNamespace.ts | 17 ++++++++++++++++-
 2 files changed, 18 insertions(+), 1 deletion(-)

diff --git a/packages/example/src/routes/tests.tsx b/packages/example/src/routes/tests.tsx
index 64bac4d..81d2fd4 100644
--- a/packages/example/src/routes/tests.tsx
+++ b/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)
diff --git a/packages/reffuse/src/ReffuseNamespace.ts b/packages/reffuse/src/ReffuseNamespace.ts
index 7bf1549..d2728a5 100644
--- a/packages/reffuse/src/ReffuseNamespace.ts
+++ b/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>[],