usePullStream

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/streams/pull.tsx

@@ -0,0 +1,14 @@
+import { R } from "@/reffuse"
+import { createFileRoute } from "@tanstack/react-router"
+import { Stream } from "effect"
+
+
+export const Route = createFileRoute("/streams/pull")({
+    component: RouteComponent,
+})
+
+function RouteComponent() {
+    const stream = R.useMemo(() => Stream.)
+
+    return <div>Hello "/streams/pull"!</div>
+}

packages/reffuse/src/ReffuseNamespace.ts

@@ -495,27 +495,32 @@ export abstract class ReffuseNamespace<R> {
         return reactStateValue as InitialA extends A ? Option.Some<A> : Option.Option<A>
     }
 
-    usePullStream<A, InitialA extends A | undefined, E, R extends ContextR, ContextR>(
+    usePullStream<A, InitialA extends A | undefined, E, R>(
-        this: ReffuseNamespace<ContextR>,
+        this: ReffuseNamespace<R>,
         stream: Stream.Stream<A, E, R>,
         initialValue?: InitialA,
     ): [
         latestValue: InitialA extends A ? Option.Some<A> : Option.Option<A>,
-        pull: Effect.Effect<Chunk.Chunk<A>, Option.Option<E>, R>,
+        pull: Effect.Effect<Chunk.Chunk<A>, Option.Option<E>>,
     ] {
         const scope = this.useScope([stream])
 
         const [reactStateValue, setReactStateValue] = React.useState<Option.Option<A>>(Option.fromNullable(initialValue))
-        const pull = this.useMemo(() => Stream.toPull(stream).pipe(
+        const pull = this.useMemo(() => Effect.context<R>().pipe(
-            Effect.map(Effect.tap(flow(
+            Effect.flatMap(context => Stream.toPull(stream).pipe(
-                Chunk.last,
+                Effect.map(effect => effect.pipe(
-                v => Option.match(v, {
+                    Effect.tap(flow(
-                    onSome: () => Effect.sync(() => setReactStateValue(v)),
+                        Chunk.last,
-                    onNone: () => Effect.void,
+                        v => Option.match(v, {
-                }),
+                            onSome: () => Effect.sync(() => setReactStateValue(v)),
-            ))),
+                            onNone: () => Effect.void,
+                        }),
+                    )),
+                    Effect.provide(context),
+                )),
 
-            Effect.provideService(Scope.Scope, scope),
+                Effect.provideService(Scope.Scope, scope),
+            ))
         ), [stream, scope])
 
         return [