Fix

packages/example/src/routeTree.gen.ts

@@ -19,6 +19,7 @@ import { Route as LazyrefImport } from './routes/lazyref'
 import { Route as CountImport } from './routes/count'
 import { Route as BlankImport } from './routes/blank'
 import { Route as IndexImport } from './routes/index'
+import { Route as StreamsPullImport } from './routes/streams/pull'
 import { Route as QueryUsequeryImport } from './routes/query/usequery'
 import { Route as QueryUsemutationImport } from './routes/query/usemutation'
 import { Route as QueryServiceImport } from './routes/query/service'
@@ -73,6 +74,12 @@ const IndexRoute = IndexImport.update({
   getParentRoute: () => rootRoute,
 } as any)
 
+const StreamsPullRoute = StreamsPullImport.update({
+  id: '/streams/pull',
+  path: '/streams/pull',
+  getParentRoute: () => rootRoute,
+} as any)
+
 const QueryUsequeryRoute = QueryUsequeryImport.update({
   id: '/query/usequery',
   path: '/query/usequery',
@@ -172,6 +179,13 @@ declare module '@tanstack/react-router' {
       preLoaderRoute: typeof QueryUsequeryImport
       parentRoute: typeof rootRoute
     }
+    '/streams/pull': {
+      id: '/streams/pull'
+      path: '/streams/pull'
+      fullPath: '/streams/pull'
+      preLoaderRoute: typeof StreamsPullImport
+      parentRoute: typeof rootRoute
+    }
   }
 }
 
@@ -189,6 +203,7 @@ export interface FileRoutesByFullPath {
   '/query/service': typeof QueryServiceRoute
   '/query/usemutation': typeof QueryUsemutationRoute
   '/query/usequery': typeof QueryUsequeryRoute
+  '/streams/pull': typeof StreamsPullRoute
 }
 
 export interface FileRoutesByTo {
@@ -203,6 +218,7 @@ export interface FileRoutesByTo {
   '/query/service': typeof QueryServiceRoute
   '/query/usemutation': typeof QueryUsemutationRoute
   '/query/usequery': typeof QueryUsequeryRoute
+  '/streams/pull': typeof StreamsPullRoute
 }
 
 export interface FileRoutesById {
@@ -218,6 +234,7 @@ export interface FileRoutesById {
   '/query/service': typeof QueryServiceRoute
   '/query/usemutation': typeof QueryUsemutationRoute
   '/query/usequery': typeof QueryUsequeryRoute
+  '/streams/pull': typeof StreamsPullRoute
 }
 
 export interface FileRouteTypes {
@@ -234,6 +251,7 @@ export interface FileRouteTypes {
     | '/query/service'
     | '/query/usemutation'
     | '/query/usequery'
+    | '/streams/pull'
   fileRoutesByTo: FileRoutesByTo
   to:
     | '/'
@@ -247,6 +265,7 @@ export interface FileRouteTypes {
     | '/query/service'
     | '/query/usemutation'
     | '/query/usequery'
+    | '/streams/pull'
   id:
     | '__root__'
     | '/'
@@ -260,6 +279,7 @@ export interface FileRouteTypes {
     | '/query/service'
     | '/query/usemutation'
     | '/query/usequery'
+    | '/streams/pull'
   fileRoutesById: FileRoutesById
 }
 
@@ -275,6 +295,7 @@ export interface RootRouteChildren {
   QueryServiceRoute: typeof QueryServiceRoute
   QueryUsemutationRoute: typeof QueryUsemutationRoute
   QueryUsequeryRoute: typeof QueryUsequeryRoute
+  StreamsPullRoute: typeof StreamsPullRoute
 }
 
 const rootRouteChildren: RootRouteChildren = {
@@ -289,6 +310,7 @@ const rootRouteChildren: RootRouteChildren = {
   QueryServiceRoute: QueryServiceRoute,
   QueryUsemutationRoute: QueryUsemutationRoute,
   QueryUsequeryRoute: QueryUsequeryRoute,
+  StreamsPullRoute: StreamsPullRoute,
 }
 
 export const routeTree = rootRoute
@@ -311,7 +333,8 @@ export const routeTree = rootRoute
         "/todos",
         "/query/service",
         "/query/usemutation",
-        "/query/usequery"
+        "/query/usequery",
+        "/streams/pull"
       ]
     },
     "/": {
@@ -346,6 +369,9 @@ export const routeTree = rootRoute
     },
     "/query/usequery": {
       "filePath": "query/usequery.tsx"
+    },
+    "/streams/pull": {
+      "filePath": "streams/pull.tsx"
     }
   }
 }

packages/example/src/routes/count.tsx

@@ -1,6 +1,6 @@
 import { R } from "@/reffuse"
 import { createFileRoute } from "@tanstack/react-router"
-import { Ref } from "effect"
+import { Effect, Ref } from "effect"
 
 
 export const Route = createFileRoute("/count")({
@@ -11,14 +11,13 @@ function Count() {
 
     const runSync = R.useRunSync()
 
-    const countRef = R.useRef(0)
+    const countRef = R.useRef(() => Effect.succeed(0))
-    const [count] = R.useRefState(countRef)
+    const [count] = R.useSubscribeRefs(countRef)
 
 
     return (
         <div className="container mx-auto">
-            {/* <button onClick={() => setCount((count) => count + 1)}> */}
+            <button onClick={() => runSync(Ref.update(countRef, count => count + 1))}>
-            <button onClick={() => Ref.update(countRef, count => count + 1).pipe(runSync)}>
                 count is {count}
             </button>
         </div>

packages/example/src/routes/query/service.tsx

@@ -14,7 +14,7 @@ export const Route = createFileRoute("/query/service")({
 
 function RouteComponent() {
     const query = R.useQuery({
-        key: R.useStreamFromValues(["uuid4", 10 as number]),
+        key: R.useStreamFromReactiveValues(["uuid4", 10 as number]),
         query: ([, count]) => Console.log(`Querying ${ count } IDs...`).pipe(
             Effect.andThen(Effect.sleep("500 millis")),
             Effect.andThen(HttpClient.get(`https://www.uuidtools.com/api/generate/v4/count/${ count }`)),

packages/example/src/routes/query/usequery.tsx

@@ -20,7 +20,7 @@ function RouteComponent() {
     const [count, setCount] = useState(1)
 
     const query = R.useQuery({
-        key: R.useStreamFromValues(["uuid4", count]),
+        key: R.useStreamFromReactiveValues(["uuid4", count]),
         query: ([, count]) => Console.log(`Querying ${ count } IDs...`).pipe(
             Effect.andThen(Effect.sleep("500 millis")),
             Effect.andThen(HttpClient.get(`https://www.uuidtools.com/api/generate/v4/count/${ count }`)),

packages/example/src/routes/streams/pull.tsx

@@ -0,0 +1,23 @@
+import { R } from "@/reffuse"
+import { Button, Flex, Text } from "@radix-ui/themes"
+import { createFileRoute } from "@tanstack/react-router"
+import { Option, Random, Stream } from "effect"
+import { useMemo } from "react"
+
+
+export const Route = createFileRoute("/streams/pull")({
+    component: RouteComponent,
+})
+
+function RouteComponent() {
+    const stream = useMemo(() => Stream.repeatEffect(Random.nextInt), [])
+    const [value, pull] = R.usePullStream(stream)
+    const pullNext = R.useCallbackSync(() => pull, [pull])
+
+    return (
+        <Flex direction="column" align="center" gap="2">
+            {Option.isSome(value) && <Text>{value.value}</Text>}
+            <Button onClick={pullNext}>Pull next</Button>
+        </Flex>
+    )
+}

packages/example/src/routes/tests.tsx

@@ -1,8 +1,12 @@
 import { R } from "@/reffuse"
-import { Button, Flex, Text } from "@radix-ui/themes"
+import { Button, Flex } from "@radix-ui/themes"
 import { createFileRoute } from "@tanstack/react-router"
 import { GetRandomValues, makeUuid4 } from "@typed/id"
-import { Console, Effect, Ref } from "effect"
+import { Console, Effect, Scope } from "effect"
+import { useEffect, useState } from "react"
+
+
+const makeUuid = Effect.provide(makeUuid4, GetRandomValues.CryptoRandom)
 
 
 export const Route = createFileRoute("/tests")({
@@ -10,48 +14,24 @@ export const Route = createFileRoute("/tests")({
 })
 
 function RouteComponent() {
-    const deepRef = R.useRef({ value: "poulet" })
+    const runSync = R.useRunSync()
-    const deepValueRef = R.useSubRef(deepRef, ["value"])
 
-    // const value = R.useMemoScoped(Effect.addFinalizer(() => Console.log("cleanup")).pipe(
+    const [uuid, setUuid] = useState(R.useMemo(() => makeUuid, []))
-    //     Effect.andThen(makeUuid4),
+    const generateUuid = R.useCallbackSync(() => makeUuid.pipe(
-    //     Effect.provide(GetRandomValues.CryptoRandom),
+        Effect.tap(v => Effect.sync(() => setUuid(v)))
-    // ), [])
-    // console.log(value)
-
-    R.useFork(() => Effect.addFinalizer(() => Console.log("cleanup")).pipe(
-        Effect.andThen(Console.log("ouient")),
-        Effect.delay("1 second"),
     ), [])
 
+    const scope = R.useScope([uuid])
 
-    const uuidRef = R.useRef("none")
+    useEffect(() => Effect.addFinalizer(() => Console.log("Scope cleanup!")).pipe(
-    const anotherRef = R.useRef(69)
+        Effect.andThen(Console.log("Scope changed")),
-
+        Effect.provideService(Scope.Scope, scope),
-
+        runSync,
-    const logValue = R.useCallbackSync(Effect.fn(function*(value: string) {
+    ), [scope, runSync])
-        yield* Effect.log(value)
-    }), [])
-
-    const generateUuid = R.useCallbackSync(() => makeUuid4.pipe(
-        Effect.provide(GetRandomValues.CryptoRandom),
-        Effect.tap(v => Ref.set(uuidRef, v)),
-        Effect.tap(v => Ref.set(deepValueRef, v)),
-    ), [])
-
 
     return (
         <Flex direction="row" justify="center" align="center" gap="2">
-            <R.SubscribeRefs refs={[uuidRef, anotherRef]}>
+            <Button onClick={generateUuid}>Generate UUID</Button>
-                {(uuid, anotherRef) => <Text>{uuid} / {anotherRef}</Text>}
-            </R.SubscribeRefs>
-
-            <R.SubscribeRefs refs={[deepRef, deepValueRef]}>
-                {(deep, deepValue) => <Text>{JSON.stringify(deep)} / {deepValue}</Text>}
-            </R.SubscribeRefs>
-
-            <Button onClick={() => logValue("test")}>Log value</Button>
-            <Button onClick={() => generateUuid()}>Generate UUID</Button>
         </Flex>
     )
 }

packages/example/src/todos/views/VTodos.tsx

@@ -16,7 +16,7 @@ export function VTodos() {
     ), [])
 
     const todosRef = R.useMemo(() => TodosState.TodosState.pipe(Effect.map(state => state.todos)), [])
-    const [todos] = R.useRefState(todosRef)
+    const [todos] = R.useSubscribeRefs(todosRef)
 
 
     return (

packages/reffuse/src/ReffuseNamespace.ts

@@ -1,4 +1,4 @@
-import { type Context, Effect, ExecutionStrategy, Exit, type Fiber, type Layer, Option, pipe, Pipeable, Queue, Ref, Runtime, Scope, Stream, SubscriptionRef } from "effect"
+import { Chunk, type Context, Effect, ExecutionStrategy, Exit, type Fiber, flow, type Layer, Match, 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"
@@ -14,6 +14,10 @@ export interface ScopeOptions {
     readonly finalizerExecutionStrategy?: ExecutionStrategy.ExecutionStrategy
 }
 
+export interface UseScopeOptions extends RenderOptions, ScopeOptions {
+    readonly finalizerExecutionMode?: "sync" | "fork"
+}
+
 export type RefsA<T extends readonly SubscriptionRef.SubscriptionRef<any>[]> = {
     [K in keyof T]: Effect.Effect.Success<T[K]>
 }
@@ -88,6 +92,55 @@ export abstract class ReffuseNamespace<R> {
         ), [runtime, context])
     }
 
+    useScope<R>(
+        this: ReffuseNamespace<R>,
+        deps: React.DependencyList = [],
+        options?: UseScopeOptions,
+    ): Scope.Scope {
+        const runSync = this.useRunSync()
+        const runFork = this.useRunFork()
+
+        const [isInitialRun, initialScope] = React.useMemo(() => runSync(Effect.all([
+            Ref.make(true),
+            Scope.make(options?.finalizerExecutionStrategy ?? ExecutionStrategy.sequential),
+        ])), [])
+
+        const [scope, setScope] = React.useState(initialScope)
+
+        React.useEffect(() => isInitialRun.pipe(
+            Effect.if({
+                onTrue: () => Effect.as(
+                    Ref.set(isInitialRun, false),
+                    () => Scope.close(initialScope, Exit.void).pipe(
+                        effect => Match.value(options?.finalizerExecutionMode ?? "sync").pipe(
+                            Match.when("sync", () => { runSync(effect) }),
+                            Match.when("fork", () => { runFork(effect) }),
+                            Match.exhaustive,
+                        )
+                    ),
+                ),
+
+                onFalse: () => Scope.make(options?.finalizerExecutionStrategy).pipe(
+                    Effect.tap(v => Effect.sync(() => setScope(v))),
+                    Effect.map(v => () => Scope.close(v, Exit.void).pipe(
+                        effect => Match.value(options?.finalizerExecutionMode ?? "sync").pipe(
+                            Match.when("sync", () => { runSync(effect) }),
+                            Match.when("fork", () => { runFork(effect) }),
+                            Match.exhaustive,
+                        )
+                    )),
+                ),
+            }),
+
+            runSync,
+        ), [
+            ...options?.doNotReExecuteOnRuntimeOrContextChange ? [] : [runSync, runFork],
+            ...deps,
+        ])
+
+        return scope
+    }
+
     /**
      * Reffuse equivalent to `React.useMemo`.
      *
@@ -111,53 +164,6 @@ export abstract class ReffuseNamespace<R> {
         ])
     }
 
-    useMemoScoped<A, E, R>(
-        this: ReffuseNamespace<R>,
-        effect: () => Effect.Effect<A, E, R | Scope.Scope>,
-        deps: React.DependencyList,
-        options?: RenderOptions & ScopeOptions,
-    ): A {
-        const runSync = this.useRunSync()
-
-        const [isInitialRun, initialScope, initialValue] = React.useMemo(() => Effect.Do.pipe(
-            Effect.bind("isInitialRun", () => Ref.make(true)),
-            Effect.bind("scope", () => Scope.make(options?.finalizerExecutionStrategy)),
-            Effect.bind("value", ({ scope }) => Effect.provideService(effect(), Scope.Scope, scope)),
-            Effect.map(({ isInitialRun, scope, value }) => [isInitialRun, scope, value] as const),
-            runSync,
-        ), [])
-
-        const [value, setValue] = React.useState(initialValue)
-
-        React.useEffect(() => isInitialRun.pipe(
-            Effect.if({
-                onTrue: () => Ref.set(isInitialRun, false).pipe(
-                    Effect.map(() =>
-                        () => runSync(Scope.close(initialScope, Exit.void))
-                    )
-                ),
-
-                onFalse: () => Effect.Do.pipe(
-                    Effect.bind("scope", () => Scope.make(options?.finalizerExecutionStrategy)),
-                    Effect.bind("value", ({ scope }) => Effect.provideService(effect(), Scope.Scope, scope)),
-                    Effect.tap(({ value }) =>
-                        Effect.sync(() => setValue(value))
-                    ),
-                    Effect.map(({ scope }) =>
-                        () => runSync(Scope.close(scope, Exit.void))
-                    ),
-                ),
-            }),
-
-            runSync,
-        ), [
-            ...options?.doNotReExecuteOnRuntimeOrContextChange ? [] : [runSync],
-            ...deps,
-        ])
-
-        return value
-    }
-
     /**
      * Reffuse equivalent to `React.useEffect`.
      *
@@ -378,17 +384,26 @@ export abstract class ReffuseNamespace<R> {
         ])
     }
 
-    useRef<A, R>(
+    useRef<A, E, R>(
         this: ReffuseNamespace<R>,
-        value: A,
+        initialValue: () => Effect.Effect<A, E, R>,
     ): SubscriptionRef.SubscriptionRef<A> {
         return this.useMemo(
-            () => SubscriptionRef.make(value),
+            () => Effect.flatMap(initialValue(), SubscriptionRef.make),
             [],
             { doNotReExecuteOnRuntimeOrContextChange: true }, // Do not recreate the ref when the context changes
         )
     }
 
+    useRefFromReactiveValue<A, R>(
+        this: ReffuseNamespace<R>,
+        value: A,
+    ): SubscriptionRef.SubscriptionRef<A> {
+        const ref = this.useRef(() => Effect.succeed(value))
+        this.useEffect(() => Ref.set(ref, value), [value], { doNotReExecuteOnRuntimeOrContextChange: true })
+        return ref
+    }
+
     useSubRef<B, const P extends PropertyPath.Paths<B>, R>(
         this: ReffuseNamespace<R>,
         parent: SubscriptionRef.SubscriptionRef<B>,
@@ -455,7 +470,7 @@ export abstract class ReffuseNamespace<R> {
         return [reactStateValue, setValue]
     }
 
-    useStreamFromValues<const A extends React.DependencyList, R>(
+    useStreamFromReactiveValues<const A extends React.DependencyList, R>(
         this: ReffuseNamespace<R>,
         values: A,
     ): Stream.Stream<A> {
@@ -464,23 +479,81 @@ export abstract class ReffuseNamespace<R> {
         ), [])
 
         this.useEffect(() => Queue.offer(queue, values), values)
-
         return stream
     }
 
-    useSubscribeStream<A, InitialA extends A | undefined, E, R>(
+    useSubscribeStream<A, E, R>(
         this: ReffuseNamespace<R>,
         stream: Stream.Stream<A, E, R>,
-        initialValue?: InitialA,
+    ): Option.Option<A>
-    ): InitialA extends A ? Option.Some<A> : Option.Option<A> {
+    useSubscribeStream<A, E, IE, R>(
-        const [reactStateValue, setReactStateValue] = React.useState<Option.Option<A>>(Option.fromNullable(initialValue))
+        this: ReffuseNamespace<R>,
+        stream: Stream.Stream<A, E, R>,
+        initialValue: () => Effect.Effect<A, IE, R>,
+    ): Option.Some<A>
+    useSubscribeStream<A, E, IE, R>(
+        this: ReffuseNamespace<R>,
+        stream: Stream.Stream<A, E, R>,
+        initialValue?: () => Effect.Effect<A, IE, R>,
+    ): Option.Option<A> {
+        const [reactStateValue, setReactStateValue] = React.useState(this.useMemo(
+            () => initialValue
+                ? Effect.map(initialValue(), Option.some)
+                : Effect.succeed(Option.none()),
+            [],
+            { doNotReExecuteOnRuntimeOrContextChange: true },
+        ))
 
         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>
+        return reactStateValue
+    }
+
+    usePullStream<A, E, IE, R>(
+        this: ReffuseNamespace<R>,
+        stream: Stream.Stream<A, E, R>,
+    ): [latestValue: Option.Option<A>, pull: Effect.Effect<Chunk.Chunk<A>, Option.Option<E>>]
+    usePullStream<A, E, IE, R>(
+        this: ReffuseNamespace<R>,
+        stream: Stream.Stream<A, E, R>,
+        initialValue: () => Effect.Effect<A, IE, R>,
+    ): [latestValue: Option.Some<A>, pull: Effect.Effect<Chunk.Chunk<A>, Option.Option<E>>]
+    usePullStream<A, E, IE, R>(
+        this: ReffuseNamespace<R>,
+        stream: Stream.Stream<A, E, R>,
+        initialValue?: () => Effect.Effect<A, IE, R>,
+    ): [latestValue: Option.Option<A>, pull: Effect.Effect<Chunk.Chunk<A>, Option.Option<E>>] {
+        const scope = this.useScope([stream])
+
+        const [reactStateValue, setReactStateValue] = React.useState(this.useMemo(
+            () => initialValue
+                ? Effect.map(initialValue(), Option.some)
+                : Effect.succeed(Option.none()),
+            [],
+            { doNotReExecuteOnRuntimeOrContextChange: true },
+        ))
+
+        const pull = this.useMemo(() => Effect.context<R>().pipe(
+            Effect.flatMap(context => Stream.toPull(Stream.changesWith(stream, (x, y) => x === y)).pipe(
+                Effect.map(effect => effect.pipe(
+                    Effect.tap(flow(
+                        Chunk.last,
+                        v => Option.match(v, {
+                            onSome: () => Effect.sync(() => setReactStateValue(v)),
+                            onNone: () => Effect.void,
+                        }),
+                    )),
+                    Effect.provide(context),
+                )),
+
+                Effect.provideService(Scope.Scope, scope),
+            ))
+        ), [stream, scope])
+
+        return [reactStateValue, pull]
     }
 
 
@@ -518,15 +591,30 @@ export abstract class ReffuseNamespace<R> {
         return props.children(this.useRefState(props.ref))
     }
 
-    SubscribeStream<A, InitialA extends A | undefined, E, R>(
+    SubscribeStream<A, E, R>(
         this: ReffuseNamespace<R>,
         props: {
             readonly stream: Stream.Stream<A, E, R>
-            readonly initialValue?: InitialA
+            readonly children: (latestValue: Option.Option<A>) => React.ReactNode
-            readonly children: (latestValue: InitialA extends A ? Option.Some<A> : Option.Option<A>) => React.ReactNode
+        },
+    ): React.ReactNode
+    SubscribeStream<A, E, IE, R>(
+        this: ReffuseNamespace<R>,
+        props: {
+            readonly stream: Stream.Stream<A, E, R>
+            readonly initialValue: () => Effect.Effect<A, IE, R>
+            readonly children: (latestValue: Option.Some<A>) => React.ReactNode
+        },
+    ): React.ReactNode
+    SubscribeStream<A, E, IE, R>(
+        this: ReffuseNamespace<R>,
+        props: {
+            readonly stream: Stream.Stream<A, E, R>
+            readonly initialValue?: () => Effect.Effect<A, IE, R>
+            readonly children: (latestValue: Option.Some<A>) => React.ReactNode
         },
     ): React.ReactNode {
-        return props.children(this.useSubscribeStream(props.stream, props.initialValue))
+        return props.children(this.useSubscribeStream(props.stream, props.initialValue as () => Effect.Effect<A, IE, R>))
     }
 }