From ba993098770c68158484349b62f8e81d5e83fb3d Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Julien=20Valverd=C3=A9?= <julien.valverde@mailo.com>
Date: Wed, 30 Apr 2025 23:41:03 +0200
Subject: [PATCH] useSubscribePullStream

---
 packages/reffuse/src/ReffuseNamespace.ts | 15 +++++++++++++++
 1 file changed, 15 insertions(+)

diff --git a/packages/reffuse/src/ReffuseNamespace.ts b/packages/reffuse/src/ReffuseNamespace.ts
index c17b3d2..0066cfb 100644
--- a/packages/reffuse/src/ReffuseNamespace.ts
+++ b/packages/reffuse/src/ReffuseNamespace.ts
@@ -492,6 +492,21 @@ export abstract class ReffuseNamespace<R> {
         return reactStateValue as InitialA extends A ? Option.Some<A> : Option.Option<A>
     }
 
+    useSubscribePullStream<A, InitialA extends A | undefined, E, R>(
+        this: ReffuseNamespace<R>,
+        stream: Stream.Stream<A, E, R>,
+        initialValue?: InitialA,
+    ): [latest: InitialA extends A ? Option.Some<A> : Option.Option<A>, pull: () => void] {
+        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>
+    }
+
 
     SubRef<B, const P extends PropertyPath.Paths<B>, R>(
         this: ReffuseNamespace<R>,