useSubscribePullStream

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>,