diff --git a/.changeset/vast-crabs-swim.md b/.changeset/vast-crabs-swim.md
new file mode 100644
index 00000000..95eb2b96
--- /dev/null
+++ b/.changeset/vast-crabs-swim.md
@@ -0,0 +1,6 @@
+---
+"@effect-rx/rx-react": patch
+"@effect-rx/rx": patch
+---
+
+add Reactive module
diff --git a/docs/rx-react/index.ts.md b/docs/rx-react/index.ts.md
index 89360c48..bf7e921f 100644
--- a/docs/rx-react/index.ts.md
+++ b/docs/rx-react/index.ts.md
@@ -12,6 +12,10 @@ Added in v1.0.0
Table of contents
+- [Reactive](#reactive)
+ - [ReactiveComponent (interface)](#reactivecomponent-interface)
+ - [action](#action)
+ - [component](#component)
- [context](#context)
- [RegistryContext](#registrycontext)
- [RegistryProvider](#registryprovider)
@@ -31,6 +35,8 @@ Added in v1.0.0
- [useRxSuspenseSuccess](#userxsuspensesuccess)
- [useRxValue](#userxvalue)
- [modules](#modules)
+ - [From "@effect-rx/rx/Reactive"](#from-effect-rxrxreactive)
+ - [From "@effect-rx/rx/ReactiveRef"](#from-effect-rxrxreactiveref)
- [From "@effect-rx/rx/Registry"](#from-effect-rxrxregistry)
- [From "@effect-rx/rx/Result"](#from-effect-rxrxresult)
- [From "@effect-rx/rx/Rx"](#from-effect-rxrxrx)
@@ -38,6 +44,62 @@ Added in v1.0.0
---
+# Reactive
+
+## ReactiveComponent (interface)
+
+**Signature**
+
+```ts
+export interface ReactiveComponent, R = never> {
+ (
+ props: Props & [R] extends [never]
+ ? {
+ readonly context?: Context.Context | undefined
+ }
+ : { readonly context: Context.Context }
+ ): React.ReactNode
+
+ provide(layer: Layer.Layer): ReactiveComponent | RL>
+
+ render: Effect.Effect<(props: Props) => React.ReactNode, never, R | Reactive.Reactive>
+}
+```
+
+Added in v1.0.0
+
+## action
+
+**Signature**
+
+```ts
+export declare const action: , A, E, R>(
+ f: (...args: Args) => Effect.Effect
+) => Effect.Effect<(...args: Args) => Promise, never, R | Scope>
+```
+
+Added in v1.0.0
+
+## component
+
+**Signature**
+
+```ts
+export declare const component: = {}>(
+ name: string,
+ build: (
+ props: Props,
+ emit: (_: React.ReactNode) => Effect.Effect
+ ) => Effect.Effect,
+ options?: {
+ readonly onInitial?: (props: Props) => React.ReactNode
+ readonly deps?: (props: Props) => ReadonlyArray
+ }
+) => ReactiveComponent>
+```
+
+Added in v1.0.0
+
# context
## RegistryContext
@@ -224,6 +286,30 @@ Added in v1.0.0
# modules
+## From "@effect-rx/rx/Reactive"
+
+Re-exports all named exports from the "@effect-rx/rx/Reactive" module as `Reactive`.
+
+**Signature**
+
+```ts
+export * as Reactive from "@effect-rx/rx/Reactive"
+```
+
+Added in v1.0.0
+
+## From "@effect-rx/rx/ReactiveRef"
+
+Re-exports all named exports from the "@effect-rx/rx/ReactiveRef" module as `ReactiveRef`.
+
+**Signature**
+
+```ts
+export * as ReactiveRef from "@effect-rx/rx/ReactiveRef"
+```
+
+Added in v1.0.0
+
## From "@effect-rx/rx/Registry"
Re-exports all named exports from the "@effect-rx/rx/Registry" module as `Registry`.
diff --git a/docs/rx/Reactive.ts.md b/docs/rx/Reactive.ts.md
new file mode 100644
index 00000000..70666498
--- /dev/null
+++ b/docs/rx/Reactive.ts.md
@@ -0,0 +1,177 @@
+---
+title: Reactive.ts
+nav_order: 2
+parent: "@effect-rx/rx"
+---
+
+## Reactive overview
+
+Added in v1.0.0
+
+---
+
+Table of contents
+
+- [Handle](#handle)
+ - [Handle (interface)](#handle-interface)
+- [Layer](#layer)
+ - [layerTimeToLive](#layertimetolive)
+- [Reactive](#reactive)
+ - [Reactive (class)](#reactive-class)
+ - [with (static method)](#with-static-method)
+ - [cache](#cache)
+ - [emit](#emit)
+- [Readable](#readable)
+ - [Readable (interface)](#readable-interface)
+ - [readable](#readable-1)
+- [Subscribable](#subscribable)
+ - [Subscribable (interface)](#subscribable-interface)
+ - [toSubscribable](#tosubscribable)
+ - [toSubscribableWith](#tosubscribablewith)
+
+---
+
+# Handle
+
+## Handle (interface)
+
+**Signature**
+
+```ts
+export interface Handle {
+ unsubscribe(reactive: Reactive["Type"]): void
+}
+```
+
+Added in v1.0.0
+
+# Layer
+
+## layerTimeToLive
+
+**Signature**
+
+```ts
+export declare const layerTimeToLive: {
+ (duration: Duration.DurationInput): (self: Layer.Layer) => Layer.Layer
+ (self: Layer.Layer, duration: Duration.DurationInput): Layer.Layer
+}
+```
+
+Added in v1.0.0
+
+# Reactive
+
+## Reactive (class)
+
+**Signature**
+
+```ts
+export declare class Reactive
+```
+
+Added in v1.0.0
+
+### with (static method)
+
+**Signature**
+
+```ts
+static with(f: (reactive: Reactive["Type"]) => Effect.Effect): Effect.Effect
+```
+
+Added in v1.0.0
+
+## cache
+
+Cache an Effect for the lifetime of the current reactive context.
+
+**Signature**
+
+```ts
+export declare const cache: (
+ ...key: ReadonlyArray
+) => (effect: Effect.Effect) => Effect.Effect | Reactive>
+```
+
+Added in v1.0.0
+
+## emit
+
+Immediately emits a value for the current reactive context.
+
+**Signature**
+
+```ts
+export declare const emit: (value: A) => Effect.Effect
+```
+
+Added in v1.0.0
+
+# Readable
+
+## Readable (interface)
+
+**Signature**
+
+```ts
+export interface Readable {
+ readonly effect: Effect.Effect
+ readonly result: Effect.Effect, never, R | Reactive>
+ readonly success: Effect.Effect, E, R | Reactive>
+}
+```
+
+Added in v1.0.0
+
+## readable
+
+**Signature**
+
+```ts
+export declare const readable: (
+ effect: Effect.Effect
+) => Readable>
+```
+
+Added in v1.0.0
+
+# Subscribable
+
+## Subscribable (interface)
+
+**Signature**
+
+```ts
+export interface Subscribable {
+ subscribe(onResult: (result: Result.Result) => void): () => void
+}
+```
+
+Added in v1.0.0
+
+## toSubscribable
+
+**Signature**
+
+```ts
+export declare const toSubscribable: (
+ layer: Layer.Layer,
+ memoMap?: Layer.MemoMap | undefined
+) => (effect: Effect.Effect) => Subscribable
+```
+
+Added in v1.0.0
+
+## toSubscribableWith
+
+**Signature**
+
+```ts
+export declare const toSubscribableWith: (
+ build: (scope: Scope.Scope, memoMap: Layer.MemoMap) => Effect.Effect, ER>,
+ memoMap?: Layer.MemoMap | undefined
+) => (effect: Effect.Effect) => Subscribable
+```
+
+Added in v1.0.0
diff --git a/docs/rx/ReactiveRef.ts.md b/docs/rx/ReactiveRef.ts.md
new file mode 100644
index 00000000..39936b93
--- /dev/null
+++ b/docs/rx/ReactiveRef.ts.md
@@ -0,0 +1,272 @@
+---
+title: ReactiveRef.ts
+nav_order: 3
+parent: "@effect-rx/rx"
+---
+
+## ReactiveRef overview
+
+Added in v1.0.0
+
+---
+
+Table of contents
+
+- [Combinators](#combinators)
+ - [get](#get)
+ - [modify](#modify)
+ - [set](#set)
+ - [subscribe](#subscribe)
+ - [unsafeGet](#unsafeget)
+ - [unsafeModify](#unsafemodify)
+ - [unsafeSet](#unsafeset)
+ - [unsafeUpdate](#unsafeupdate)
+ - [update](#update)
+- [Constructors](#constructors)
+ - [make](#make)
+ - [unsafeMake](#unsafemake)
+- [Conversions](#conversions)
+ - [fromStream](#fromstream)
+ - [fromStreamPull](#fromstreampull)
+ - [intoStream](#intostream)
+ - [intoStreamPull](#intostreampull)
+- [Models](#models)
+ - [ReactiveRef (interface)](#reactiveref-interface)
+- [Symbols](#symbols)
+ - [TypeId](#typeid)
+ - [TypeId (type alias)](#typeid-type-alias)
+
+---
+
+# Combinators
+
+## get
+
+**Signature**
+
+```ts
+export declare const get: (self: ReactiveRef) => Effect.Effect
+```
+
+Added in v1.0.0
+
+## modify
+
+**Signature**
+
+```ts
+export declare const modify: {
+ (f: (_: A) => readonly [returnValue: R, nextValue: A]): (self: ReactiveRef) => Effect.Effect
+ (self: ReactiveRef, f: (_: A) => readonly [returnValue: R, nextValue: A]): Effect.Effect
+}
+```
+
+Added in v1.0.0
+
+## set
+
+**Signature**
+
+```ts
+export declare const set: {
+ (value: A): (self: ReactiveRef) => Effect.Effect
+ (self: ReactiveRef, value: A): Effect.Effect
+}
+```
+
+Added in v1.0.0
+
+## subscribe
+
+**Signature**
+
+```ts
+export declare const subscribe: (self: ReactiveRef) => Effect.Effect
+```
+
+Added in v1.0.0
+
+## unsafeGet
+
+**Signature**
+
+```ts
+export declare const unsafeGet: (self: ReactiveRef) => A
+```
+
+Added in v1.0.0
+
+## unsafeModify
+
+**Signature**
+
+```ts
+export declare const unsafeModify: {
+ (f: (_: A) => readonly [returnValue: R, nextValue: A]): (self: ReactiveRef) => R
+ (self: ReactiveRef, f: (_: A) => readonly [returnValue: R, nextValue: A]): R
+}
+```
+
+Added in v1.0.0
+
+## unsafeSet
+
+**Signature**
+
+```ts
+export declare const unsafeSet: {
+ (value: A): (self: ReactiveRef) => void
+ (self: ReactiveRef, value: A): void
+}
+```
+
+Added in v1.0.0
+
+## unsafeUpdate
+
+**Signature**
+
+```ts
+export declare const unsafeUpdate: {
+ (f: (_: A) => A): (self: ReactiveRef) => void
+ (self: ReactiveRef, f: (_: A) => A): void
+}
+```
+
+Added in v1.0.0
+
+## update
+
+**Signature**
+
+```ts
+export declare const update: {
+ (f: (_: A) => A): (self: ReactiveRef) => Effect.Effect
+ (self: ReactiveRef, f: (_: A) => A): Effect.Effect
+}
+```
+
+Added in v1.0.0
+
+# Constructors
+
+## make
+
+**Signature**
+
+```ts
+export declare const make: (value: A) => Effect.Effect>
+```
+
+Added in v1.0.0
+
+## unsafeMake
+
+**Signature**
+
+```ts
+export declare const unsafeMake: (value: A) => ReactiveRef
+```
+
+Added in v1.0.0
+
+# Conversions
+
+## fromStream
+
+**Signature**
+
+```ts
+export declare const fromStream: (
+ stream: Stream.Stream
+) => Effect.Effect>, never, R | Scope.Scope>
+```
+
+Added in v1.0.0
+
+## fromStreamPull
+
+**Signature**
+
+```ts
+export declare const fromStreamPull: (
+ stream: Stream.Stream
+) => Effect.Effect<{ readonly ref: ReactiveRef>; readonly pull: () => void }, never, R | Scope.Scope>
+```
+
+Added in v1.0.0
+
+## intoStream
+
+**Signature**
+
+```ts
+export declare const intoStream: {
+ (
+ self: ReactiveRef>
+ ): (stream: Stream.Stream) => Effect.Effect
+ (
+ self: ReactiveRef>,
+ stream: Stream.Stream
+ ): Effect.Effect
+}
+```
+
+Added in v1.0.0
+
+## intoStreamPull
+
+**Signature**
+
+```ts
+export declare const intoStreamPull: {
+ (
+ self: ReactiveRef>
+ ): (stream: Stream.Stream) => Effect.Effect<() => void, never, R | Scope.Scope>
+ (
+ self: ReactiveRef>,
+ stream: Stream.Stream
+ ): Effect.Effect<() => void, never, R | Scope.Scope>
+}
+```
+
+Added in v1.0.0
+
+# Models
+
+## ReactiveRef (interface)
+
+**Signature**
+
+```ts
+export interface ReactiveRef extends Handle, Effect.Effect {
+ readonly [TypeId]: TypeId
+ subscribe(reactive: Reactive.Reactive["Type"]): void
+ unsafeSet(value: A): void
+ value: A
+}
+```
+
+Added in v1.0.0
+
+# Symbols
+
+## TypeId
+
+**Signature**
+
+```ts
+export declare const TypeId: typeof TypeId
+```
+
+Added in v1.0.0
+
+## TypeId (type alias)
+
+**Signature**
+
+```ts
+export type TypeId = typeof TypeId
+```
+
+Added in v1.0.0
diff --git a/docs/rx/Registry.ts.md b/docs/rx/Registry.ts.md
index 19e39ef9..24756bf9 100644
--- a/docs/rx/Registry.ts.md
+++ b/docs/rx/Registry.ts.md
@@ -1,6 +1,6 @@
---
title: Registry.ts
-nav_order: 2
+nav_order: 4
parent: "@effect-rx/rx"
---
diff --git a/docs/rx/Result.ts.md b/docs/rx/Result.ts.md
index 0ef5e5ec..abddaaf1 100644
--- a/docs/rx/Result.ts.md
+++ b/docs/rx/Result.ts.md
@@ -1,6 +1,6 @@
---
title: Result.ts
-nav_order: 3
+nav_order: 5
parent: "@effect-rx/rx"
---
diff --git a/docs/rx/Rx.ts.md b/docs/rx/Rx.ts.md
index b04a820b..72b6b6c4 100644
--- a/docs/rx/Rx.ts.md
+++ b/docs/rx/Rx.ts.md
@@ -1,6 +1,6 @@
---
title: Rx.ts
-nav_order: 4
+nav_order: 6
parent: "@effect-rx/rx"
---
diff --git a/docs/rx/RxRef.ts.md b/docs/rx/RxRef.ts.md
index 6ad57a4f..e6019b06 100644
--- a/docs/rx/RxRef.ts.md
+++ b/docs/rx/RxRef.ts.md
@@ -1,6 +1,6 @@
---
title: RxRef.ts
-nav_order: 5
+nav_order: 7
parent: "@effect-rx/rx"
---
diff --git a/docs/rx/index.ts.md b/docs/rx/index.ts.md
index 4bf23e52..438bfac8 100644
--- a/docs/rx/index.ts.md
+++ b/docs/rx/index.ts.md
@@ -13,6 +13,8 @@ Added in v1.0.0
Table of contents
- [exports](#exports)
+ - [From "./Reactive.js"](#from-reactivejs)
+ - [From "./ReactiveRef.js"](#from-reactiverefjs)
- [From "./Registry.js"](#from-registryjs)
- [From "./Result.js"](#from-resultjs)
- [From "./Rx.js"](#from-rxjs)
@@ -22,6 +24,30 @@ Added in v1.0.0
# exports
+## From "./Reactive.js"
+
+Re-exports all named exports from the "./Reactive.js" module as `Reactive`.
+
+**Signature**
+
+```ts
+export * as Reactive from "./Reactive.js"
+```
+
+Added in v1.0.0
+
+## From "./ReactiveRef.js"
+
+Re-exports all named exports from the "./ReactiveRef.js" module as `ReactiveRef`.
+
+**Signature**
+
+```ts
+export * as ReactiveRef from "./ReactiveRef.js"
+```
+
+Added in v1.0.0
+
## From "./Registry.js"
Re-exports all named exports from the "./Registry.js" module as `Registry`.
diff --git a/package.json b/package.json
index 73a21b8a..41b47e22 100644
--- a/package.json
+++ b/package.json
@@ -18,36 +18,36 @@
"docgen-cp": "node scripts/docs-cp.js"
},
"devDependencies": {
- "@babel/cli": "^7.25.7",
- "@babel/core": "^7.25.8",
- "@babel/plugin-transform-export-namespace-from": "^7.25.8",
- "@babel/plugin-transform-modules-commonjs": "^7.25.7",
- "@changesets/changelog-github": "^0.5.0",
- "@changesets/cli": "^2.27.9",
- "@effect/build-utils": "^0.7.8",
- "@effect/docgen": "^0.4.5",
+ "@babel/cli": "^7.27.0",
+ "@babel/core": "^7.26.10",
+ "@babel/plugin-transform-export-namespace-from": "^7.25.9",
+ "@babel/plugin-transform-modules-commonjs": "^7.26.3",
+ "@changesets/changelog-github": "^0.5.1",
+ "@changesets/cli": "^2.28.1",
+ "@effect/build-utils": "^0.7.9",
+ "@effect/docgen": "^0.5.2",
"@effect/eslint-plugin": "^0.2.0",
- "@effect/language-service": "^0.2.0",
- "@eslint/compat": "^1.2.1",
- "@eslint/eslintrc": "^3.1.0",
- "@eslint/js": "^9.13.0",
- "@typescript-eslint/eslint-plugin": "^8.10.0",
- "@typescript-eslint/parser": "^8.10.0",
- "@vitest/coverage-v8": "^2.1.3",
- "babel-plugin-annotate-pure-calls": "^0.4.0",
- "eslint": "^9.13.0",
- "eslint-import-resolver-typescript": "^3.6.3",
- "eslint-plugin-codegen": "^0.29.0",
+ "@effect/language-service": "^0.5.1",
+ "@eslint/compat": "^1.2.8",
+ "@eslint/eslintrc": "^3.3.1",
+ "@eslint/js": "^9.23.0",
+ "@typescript-eslint/eslint-plugin": "^8.29.0",
+ "@typescript-eslint/parser": "^8.29.0",
+ "@vitest/coverage-v8": "^3.1.1",
+ "babel-plugin-annotate-pure-calls": "^0.5.0",
+ "eslint": "^9.23.0",
+ "eslint-import-resolver-typescript": "^4.3.1",
+ "eslint-plugin-codegen": "^0.30.0",
"eslint-plugin-deprecation": "^3.0.0",
"eslint-plugin-import": "^2.31.0",
"eslint-plugin-simple-import-sort": "^12.1.1",
"eslint-plugin-sort-destructure-keys": "^2.0.0",
- "fast-check": "^3.22.0",
- "glob": "^11.0.0",
+ "fast-check": "^4.0.1",
+ "glob": "^11.0.1",
"madge": "^8.0.0",
- "prettier": "^3.3.3",
- "tsx": "^4.19.1",
- "typescript": "^5.6.3",
- "vitest": "^2.1.3"
+ "prettier": "^3.5.3",
+ "tsx": "^4.19.3",
+ "typescript": "^5.8.2",
+ "vitest": "^3.1.1"
}
}
diff --git a/packages/rx-react/package.json b/packages/rx-react/package.json
index 9dacbd9a..98f21e75 100644
--- a/packages/rx-react/package.json
+++ b/packages/rx-react/package.json
@@ -25,12 +25,12 @@
"devDependencies": {
"@types/react": "^18.3.11",
"@types/scheduler": "^0.23.0",
- "effect": "^3.9.2",
+ "effect": "^3.14.0",
"react": "^18.3.1",
"scheduler": "^0.23"
},
"peerDependencies": {
- "effect": "^3.8.0",
+ "effect": "^3.14.0",
"react": ">=18 <20",
"scheduler": "*"
},
diff --git a/packages/rx-react/src/index.ts b/packages/rx-react/src/index.ts
index 58160ea7..0411cd99 100644
--- a/packages/rx-react/src/index.ts
+++ b/packages/rx-react/src/index.ts
@@ -1,13 +1,21 @@
/**
* @since 1.0.0
*/
+/* eslint-disable @typescript-eslint/no-empty-object-type */
+import * as Reactive from "@effect-rx/rx/Reactive"
import * as Registry from "@effect-rx/rx/Registry"
import * as Result from "@effect-rx/rx/Result"
import * as Rx from "@effect-rx/rx/Rx"
import type * as RxRef from "@effect-rx/rx/RxRef"
import * as Cause from "effect/Cause"
+import * as Context from "effect/Context"
+import * as Effect from "effect/Effect"
import type * as Exit from "effect/Exit"
+import * as FiberSet from "effect/FiberSet"
+import { constNull } from "effect/Function"
import { globalValue } from "effect/GlobalValue"
+import * as Layer from "effect/Layer"
+import type { Scope } from "effect/Scope"
import * as React from "react"
import * as Scheduler from "scheduler"
@@ -31,6 +39,16 @@ export * as Rx from "@effect-rx/rx/Rx"
* @category modules
*/
export * as RxRef from "@effect-rx/rx/RxRef"
+/**
+ * @since 1.0.0
+ * @category modules
+ */
+export * as Reactive from "@effect-rx/rx/Reactive"
+/**
+ * @since 1.0.0
+ * @category modules
+ */
+export * as ReactiveRef from "@effect-rx/rx/ReactiveRef"
/**
* @since 1.0.0
@@ -349,3 +367,157 @@ export const useRxRefProp = (ref: RxRef.RxRef, prop: K)
*/
export const useRxRefPropValue = (ref: RxRef.RxRef, prop: K): A[K] =>
useRxRef(useRxRefProp(ref, prop))
+
+/**
+ * @since 1.0.0
+ * @category Reactive
+ */
+export interface ReactiveComponent, R = never> {
+ (
+ props: Props & [R] extends [never] ? {
+ readonly context?: Context.Context | undefined
+ } :
+ { readonly context: Context.Context }
+ ): React.ReactNode
+
+ provide(layer: Layer.Layer): ReactiveComponent | RL>
+
+ render: Effect.Effect<(props: Props) => React.ReactNode, never, R | Reactive.Reactive>
+}
+
+const depsAreEqual = (a: ReadonlyArray, b: ReadonlyArray) => {
+ if (a.length !== b.length) {
+ return false
+ }
+ for (let i = 0; i < a.length; i++) {
+ if (!Object.is(a[i], b[i])) {
+ return false
+ }
+ }
+ return true
+}
+
+const makeReactiveComponent = , E, R>(options: {
+ readonly name: string
+ readonly build: (
+ props: Props,
+ emit: (_: React.ReactNode) => Effect.Effect
+ ) => Effect.Effect
+ readonly layer: Layer.Layer
+ readonly onInitial: (props: Props) => React.ReactNode
+ readonly deps: (props: Props) => ReadonlyArray
+}): ReactiveComponent => {
+ function ReactiveComponent(props: Props & { readonly context?: Context.Context }) {
+ const [result, setResult] = React.useState>(Result.initial(true))
+
+ const ref = React.useRef<{
+ readonly subscribable: Reactive.Subscribable
+ previousDeps: ReadonlyArray
+ cancel: () => void
+ timeout: number | undefined
+ }>(undefined as any)
+
+ const deps = options.deps(props)
+
+ if (ref.current === undefined || !depsAreEqual(ref.current.previousDeps, deps)) {
+ if (ref.current) {
+ ref.current.cancel()
+ if (ref.current.timeout !== undefined) {
+ clearTimeout(ref.current.timeout)
+ ref.current.timeout = undefined
+ }
+ }
+
+ const layer = props.context
+ ? Layer.provideMerge(options.layer, Layer.succeedContext(props.context))
+ : options.layer
+ const subscribable = Reactive.toSubscribable(layer)(
+ options.build(props, Reactive.emit) as Effect.Effect
+ )
+ const cancel = subscribable.subscribe(setResult)
+ ref.current = {
+ subscribable,
+ cancel,
+ previousDeps: deps,
+ timeout: undefined
+ }
+ }
+
+ React.useEffect(() => {
+ if (ref.current.timeout !== undefined) {
+ clearTimeout(ref.current.timeout)
+ ref.current.timeout = undefined
+ }
+ return () => {
+ ref.current.timeout = setTimeout(ref.current.cancel, 100)
+ }
+ }, [ref.current])
+ if (result._tag === "Initial") {
+ return options.onInitial(props)
+ } else if (result._tag === "Failure") {
+ throw Cause.squash(result.cause)
+ }
+ return result.value
+ }
+ ReactiveComponent.displayName = options.name
+ ReactiveComponent.provide = function provide(layer: Layer.Layer) {
+ return makeReactiveComponent({
+ ...options,
+ layer: options.layer === Layer.empty ? layer : Layer.provideMerge(options.layer, layer) as any
+ })
+ }
+ let renderCache: WeakMap React.ReactNode> | undefined
+ ReactiveComponent.render = Effect.contextWith((context: Context.Context) => {
+ const reactive = Context.unsafeGet(context, Reactive.Reactive)
+ if (renderCache?.has(reactive.parentScope)) {
+ return renderCache.get(reactive.parentScope)!
+ }
+ function Wrapped(props: Props) {
+ return React.createElement(ReactiveComponent, { ...props, context })
+ }
+ Wrapped.displayName = `${options.name}.render`
+ if (renderCache === undefined) {
+ renderCache = new WeakMap()
+ }
+ renderCache.set(reactive.parentScope, Wrapped)
+ return Wrapped
+ })
+ return ReactiveComponent as any
+}
+
+const defaultDeps = (props: Record) => Object.values(props)
+
+/**
+ * @since 1.0.0
+ * @category Reactive
+ */
+export const component = = {}>(
+ name: string,
+ build: (
+ props: Props,
+ emit: (_: React.ReactNode) => Effect.Effect
+ ) => Effect.Effect,
+ options?: {
+ readonly onInitial?: (props: Props) => React.ReactNode
+ readonly deps?: (props: Props) => ReadonlyArray
+ }
+): ReactiveComponent> =>
+ makeReactiveComponent({
+ name,
+ build,
+ layer: Layer.empty,
+ onInitial: options?.onInitial ?? constNull,
+ deps: options?.deps ?? defaultDeps
+ }) as any
+
+/**
+ * @since 1.0.0
+ * @category Reactive
+ */
+export const action = , A, E, R>(
+ f: (...args: Args) => Effect.Effect
+): Effect.Effect<(...args: Args) => Promise, never, R | Scope> =>
+ Effect.map(
+ FiberSet.makeRuntimePromise(),
+ (runPromise) => (...args) => runPromise(f(...args))
+ )
diff --git a/packages/rx/package.json b/packages/rx/package.json
index 4d182261..d843760a 100644
--- a/packages/rx/package.json
+++ b/packages/rx/package.json
@@ -24,9 +24,9 @@
"license": "MIT",
"sideEffects": [],
"devDependencies": {
- "effect": "^3.9.2"
+ "effect": "^3.14.0"
},
"peerDependencies": {
- "effect": "^3.8.0"
+ "effect": "^3.14.0"
}
}
diff --git a/packages/rx/src/Reactive.ts b/packages/rx/src/Reactive.ts
new file mode 100644
index 00000000..036cfd6d
--- /dev/null
+++ b/packages/rx/src/Reactive.ts
@@ -0,0 +1,453 @@
+/**
+ * @since 1.0.0
+ */
+import * as Context from "effect/Context"
+import * as Data from "effect/Data"
+import * as Deferred from "effect/Deferred"
+import * as Duration from "effect/Duration"
+import * as Effect from "effect/Effect"
+import * as Exit from "effect/Exit"
+import * as Fiber from "effect/Fiber"
+import { dual } from "effect/Function"
+import { globalValue } from "effect/GlobalValue"
+import * as Layer from "effect/Layer"
+import * as MutableHashMap from "effect/MutableHashMap"
+import * as Option from "effect/Option"
+import * as Runtime from "effect/Runtime"
+import * as Scope from "effect/Scope"
+import { runCallbackSync } from "./internal/runtime.js"
+import * as Result from "./Result.js"
+
+/**
+ * @since 1.0.0
+ * @category Reactive
+ */
+export class Reactive extends Context.Tag("@effect-rx/rx/Reactive")<
+ Reactive,
+ {
+ parentScope: Scope.Scope
+ disposed: boolean
+ notify(): void
+ emit(value: unknown): void
+ addHandle(handle: Handle): void
+ cache(key: unknown, value: Effect.Effect): Effect.Effect>
+ }
+>() {
+ /**
+ * @since 1.0.0
+ */
+ static with(f: (reactive: Reactive["Type"]) => Effect.Effect): Effect.Effect {
+ return Effect.withFiberRuntime((fiber) => {
+ const reactive = fiber.currentContext.unsafeMap.get(Reactive.key) as Reactive["Type"]
+ return f(reactive)
+ })
+ }
+}
+
+/**
+ * Immediately emits a value for the current reactive context.
+ *
+ * @since 1.0.0
+ * @category Reactive
+ */
+export const emit = (value: A): Effect.Effect =>
+ Reactive.with((reactive) => {
+ reactive.emit(value)
+ return Effect.void
+ })
+
+/**
+ * Cache an Effect for the lifetime of the current reactive context.
+ *
+ * @since 1.0.0
+ * @category Reactive
+ */
+export const cache =
+ (...key: ReadonlyArray) =>
+ (effect: Effect.Effect): Effect.Effect | Reactive> =>
+ Reactive.with((reactive) => reactive.cache(Data.array(key), effect))
+
+/**
+ * @since 1.0.0
+ * @category Handle
+ */
+export interface Handle {
+ unsubscribe(reactive: Reactive["Type"]): void
+}
+
+/**
+ * @since 1.0.0
+ * @category Subscribable
+ */
+export interface Subscribable {
+ subscribe(onResult: (result: Result.Result) => void): () => void
+}
+
+const defaultMemoMap = globalValue(
+ "@effect-rx/rx/Reactive/defaultMemoMap",
+ () => Effect.runSync(Layer.makeMemoMap)
+)
+
+/**
+ * @since 1.0.0
+ * @category Subscribable
+ */
+export const toSubscribableWith = (
+ build: (scope: Scope.Scope, memoMap: Layer.MemoMap) => Effect.Effect, ER>,
+ memoMap?: Layer.MemoMap | undefined
+) => {
+ const buildMemoMap = memoMap ?? defaultMemoMap
+
+ return (effect: Effect.Effect): Subscribable => {
+ let result: Result.Result = Result.initial(true)
+ let waiting = false
+ const callbacks = new Set<(result: Result.Result) => void>()
+ let scope: Scope.CloseableScope | undefined
+ let cancel: (() => void) | undefined
+ let currentScope: Scope.CloseableScope | undefined
+ let handles: Array = []
+ let reactive: Reactive["Type"] | undefined
+
+ function setResult(newResult: Result.Result) {
+ waiting = false
+ result = newResult
+ for (const callback of callbacks) {
+ callback(result)
+ }
+ }
+
+ function onExit(exit: Exit.Exit) {
+ waiting = false
+ cancel = undefined
+ setResult(Result.fromExitWithPrevious(exit, Option.some(result)))
+ }
+
+ function start() {
+ scope = Effect.runSync(Scope.make())
+
+ const initial = Effect.flatMap(
+ Effect.uninterruptible(build(scope, buildMemoMap)),
+ (context) => {
+ let pending = false
+ function run() {
+ pending = false
+
+ if (cancel) cancel()
+ Effect.runFork(Scope.close(currentScope!, Exit.void))
+
+ const prevHandles = handles
+ handles = []
+ waiting = true
+ currentScope = Effect.runSync(Scope.make())
+ cancel = runCallback(Scope.extend(effect, currentScope), onExit)
+
+ if (waiting) {
+ setResult(Result.waitingFrom(Option.some(result)))
+ }
+
+ for (const handle of prevHandles) {
+ if (handles.includes(handle)) continue
+ handle.unsubscribe(reactive!)
+ }
+ }
+
+ let cache: MutableHashMap.MutableHashMap> | undefined
+ reactive = {
+ disposed: false,
+ parentScope: scope!,
+ notify() {
+ if (pending || this.disposed) return
+ pending = true
+ queueMicrotask(run)
+ },
+ emit(value: A) {
+ if (this.disposed) return
+ setResult(Result.success(value, true))
+ },
+ addHandle(handle) {
+ if (this.disposed) {
+ queueMicrotask(() => handle.unsubscribe(reactive!))
+ return
+ }
+ handles.push(handle)
+ },
+ cache(key: unknown, effect: Effect.Effect): Effect.Effect> {
+ return Effect.withFiberRuntime((fiber) => {
+ if (!cache) {
+ cache = MutableHashMap.empty()
+ }
+ const maybeValue = MutableHashMap.get(cache, key)
+ if (Option.isSome(maybeValue)) {
+ return maybeValue.value
+ }
+ const deferred = Deferred.unsafeMake(fiber.id())
+ MutableHashMap.set(cache, key, Deferred.await(deferred))
+ return effect.pipe(
+ Scope.extend(scope!),
+ Effect.onExit((exit) => Deferred.done(deferred, exit))
+ )
+ })
+ }
+ }
+
+ const fiber = Option.getOrThrow(Fiber.getCurrentFiber())
+ const contextReactive = Context.add(context, Reactive, reactive)
+ const runtime = Runtime.make({
+ context: contextReactive,
+ runtimeFlags: Runtime.defaultRuntime.runtimeFlags,
+ fiberRefs: fiber.getFiberRefs()
+ })
+ const runCallback = runCallbackSync(runtime)
+ currentScope = Effect.runSync(Scope.make())
+ return Effect.provide(effect, Context.add(contextReactive, Scope.Scope, currentScope))
+ }
+ )
+
+ waiting = true
+ cancel = runCallbackSync(Runtime.defaultRuntime)(initial, onExit)
+ }
+
+ return {
+ subscribe(onResult) {
+ callbacks.add(onResult)
+
+ if (!scope) {
+ start()
+ if (waiting) {
+ onResult(result)
+ }
+ } else {
+ onResult(result)
+ }
+
+ return () => {
+ callbacks.delete(onResult)
+ if (callbacks.size > 0) return
+ result = Result.initial(true)
+ if (cancel) cancel()
+ if (reactive) {
+ reactive.disposed = true
+ Effect.runFork(Scope.close(currentScope!, Exit.void))
+ }
+ reactive = cancel = currentScope = undefined
+ for (const handle of handles) {
+ handle.unsubscribe(reactive!)
+ }
+ handles = []
+ Effect.runFork(Scope.close(scope!, Exit.void))
+ scope = undefined
+ }
+ }
+ }
+ }
+}
+
+/**
+ * @since 1.0.0
+ * @category Subscribable
+ */
+export const toSubscribable = (
+ layer: Layer.Layer,
+ memoMap?: Layer.MemoMap | undefined
+): (effect: Effect.Effect) => Subscribable =>
+ toSubscribableWith((scope, memoMap) => Layer.buildWithMemoMap(layer, memoMap, scope), memoMap)
+
+/**
+ * @since 1.0.0
+ * @category Readable
+ */
+export interface Readable {
+ readonly effect: Effect.Effect
+ readonly result: Effect.Effect, never, R | Reactive>
+ readonly success: Effect.Effect, E, R | Reactive>
+}
+
+/**
+ * @since 1.0.0
+ * @category Readable
+ */
+export const readable = (
+ effect: Effect.Effect
+): Readable> => {
+ let context: Context.Context | undefined
+ const sub = toSubscribableWith(() => Effect.succeed(context!))(effect)
+
+ const subscribers = new Set()
+ const pending = new Set()
+ let cancel: (() => void) | undefined
+ const handle: Handle = {
+ unsubscribe(reactive) {
+ subscribers.delete(reactive)
+ pending.delete(reactive)
+ if (subscribers.size === 0 && cancel) {
+ cancel()
+ cancel = context = currentResult = undefined
+ }
+ }
+ }
+
+ const resumes = {
+ effect: new Set<(_: Effect.Effect) => void>(),
+ success: new Set<(_: Effect.Effect, E>) => void>()
+ }
+
+ let currentResult: Result.Result | undefined
+ function start() {
+ cancel = sub.subscribe((result) => {
+ currentResult = result
+ if (result._tag !== "Initial") {
+ if (resumes.effect.size > 0) {
+ const exit = Result.toExit(result)
+ for (const resume of resumes.effect) {
+ resume(exit as any)
+ }
+ resumes.effect.clear()
+ }
+ if (resumes.success.size > 0) {
+ const exit: Exit.Exit, E> = result._tag === "Success"
+ ? Exit.succeed(result)
+ : Exit.failCause(result.cause)
+ for (const resume of resumes.success) {
+ resume(exit as any)
+ }
+ resumes.success.clear()
+ }
+ }
+ for (const reactive of pending) {
+ reactive.notify()
+ }
+ pending.clear()
+ })
+ }
+
+ return {
+ effect: Effect.withFiberRuntime((fiber) => {
+ const reactive = fiber.currentContext.unsafeMap.get(Reactive.key) as Reactive["Type"]
+ context = fiber.currentContext as any
+ reactive.addHandle(handle)
+ subscribers.add(reactive)
+ if (!cancel) {
+ start()
+ }
+ if (currentResult!._tag !== "Initial") {
+ pending.add(reactive)
+ return Result.toExit(currentResult!) as Effect.Effect
+ }
+ return Effect.tap(
+ Effect.async((resume) => {
+ if (currentResult!._tag !== "Initial") {
+ return resume(Result.toExit(currentResult!) as any)
+ }
+ resumes.effect.add(resume)
+ return Effect.sync(() => {
+ resumes.effect.delete(resume)
+ })
+ }),
+ () => {
+ pending.add(reactive)
+ }
+ )
+ }),
+ success: Effect.withFiberRuntime((fiber) => {
+ const reactive = fiber.currentContext.unsafeMap.get(Reactive.key) as Reactive["Type"]
+ context = fiber.currentContext as any
+ reactive.addHandle(handle)
+ subscribers.add(reactive)
+ if (!cancel) {
+ start()
+ }
+ if (currentResult!._tag !== "Initial") {
+ pending.add(reactive)
+ return currentResult!._tag === "Success"
+ ? Effect.succeed(currentResult as Result.Success)
+ : Effect.failCause(currentResult!.cause)
+ }
+ return Effect.tap(
+ Effect.async, E>((resume) => {
+ if (currentResult!._tag !== "Initial") {
+ return resume(
+ currentResult!._tag === "Success"
+ ? Effect.succeed(currentResult as Result.Success)
+ : Effect.failCause(currentResult!.cause)
+ )
+ }
+ resumes.success.add(resume)
+ return Effect.sync(() => {
+ resumes.success.delete(resume)
+ })
+ }),
+ () => {
+ pending.add(reactive)
+ }
+ )
+ }),
+ result: Effect.withFiberRuntime((fiber) => {
+ const reactive = fiber.currentContext.unsafeMap.get(Reactive.key) as Reactive["Type"]
+ context = fiber.currentContext as any
+ reactive.addHandle(handle)
+ subscribers.add(reactive)
+ if (!cancel) {
+ start()
+ }
+ pending.add(reactive)
+ return Effect.succeed(currentResult!)
+ })
+ }
+}
+
+/**
+ * @since 1.0.0
+ * @category Layer
+ */
+export const layerTimeToLive: {
+ (duration: Duration.DurationInput): (self: Layer.Layer) => Layer.Layer
+ (self: Layer.Layer, duration: Duration.DurationInput): Layer.Layer
+} = dual(2, (self: Layer.Layer, duration: Duration.DurationInput): Layer.Layer => {
+ const cache = new Map
+ readonly finalizer: Effect.Effect
+ refCount: number
+ fiber: Fiber.Fiber | undefined
+ }>()
+ const resolvedDuration = Duration.decode(duration)
+
+ return Layer.scopedContext(Effect.gen(function*() {
+ const context = yield* Effect.context()
+ const memoMap = Context.get(context, Layer.CurrentMemoMap)
+ const layerScope = Context.get(context, Scope.Scope)
+
+ if (cache.has(memoMap)) {
+ const entry = cache.get(memoMap)!
+ if (entry.fiber) {
+ yield* Fiber.interrupt(entry.fiber)
+ entry.fiber = undefined
+ }
+ if (Duration.isFinite(resolvedDuration)) {
+ yield* Scope.addFinalizer(layerScope, entry.finalizer)
+ entry.refCount++
+ }
+ return entry.context
+ }
+
+ const childScope = yield* Scope.make()
+ const built = yield* Layer.buildWithMemoMap(self, memoMap, childScope)
+ const entry = {
+ context: built,
+ finalizer: Effect.sync(() => {
+ entry.refCount--
+ if (entry.refCount > 0) return
+ entry.fiber = Effect.runFork(Effect.flatMap(Effect.sleep(resolvedDuration), () => {
+ cache.delete(memoMap)
+ return Scope.close(childScope, Exit.void)
+ }))
+ }),
+ refCount: 1,
+ fiber: undefined as Fiber.Fiber | undefined
+ }
+ if (Duration.isFinite(resolvedDuration)) {
+ yield* Scope.addFinalizer(layerScope, entry.finalizer)
+ }
+ cache.set(memoMap, entry)
+ return built
+ }))
+})
diff --git a/packages/rx/src/ReactiveRef.ts b/packages/rx/src/ReactiveRef.ts
new file mode 100644
index 00000000..4c04bbd9
--- /dev/null
+++ b/packages/rx/src/ReactiveRef.ts
@@ -0,0 +1,332 @@
+/**
+ * @since 1.0.0
+ */
+import * as Cause from "effect/Cause"
+import * as Chunk from "effect/Chunk"
+import * as Context from "effect/Context"
+import * as Effect from "effect/Effect"
+import * as Effectable from "effect/Effectable"
+import * as Fiber from "effect/Fiber"
+import { dual } from "effect/Function"
+import * as Option from "effect/Option"
+import * as Runtime from "effect/Runtime"
+import * as Scope from "effect/Scope"
+import * as Stream from "effect/Stream"
+import type { Handle } from "./Reactive.js"
+import * as Reactive from "./Reactive.js"
+import * as Result from "./Result.js"
+import type { PullResult } from "./Rx.js"
+
+/**
+ * @since 1.0.0
+ * @category Symbols
+ */
+export const TypeId: unique symbol = Symbol.for("@effect-rx/rx/ReactiveRef")
+
+/**
+ * @since 1.0.0
+ * @category Symbols
+ */
+export type TypeId = typeof TypeId
+
+/**
+ * @since 1.0.0
+ * @category Models
+ */
+export interface ReactiveRef extends Handle, Effect.Effect {
+ readonly [TypeId]: TypeId
+ subscribe(reactive: Reactive.Reactive["Type"]): void
+ unsafeSet(value: A): void
+ value: A
+}
+
+const proto = {
+ ...Effectable.CommitPrototype,
+ [TypeId]: TypeId,
+ commit(this: ReactiveRef) {
+ return subscribe(this)
+ }
+}
+
+/**
+ * @since 1.0.0
+ * @category Constructors
+ */
+export const unsafeMake = (value: A): ReactiveRef => {
+ const subscribers = new Set()
+ const self = {
+ ...proto,
+ subscribe(reactive) {
+ subscribers.add(reactive)
+ },
+ unsubscribe(reactive) {
+ subscribers.delete(reactive)
+ },
+ unsafeSet(value) {
+ if (self.value === value) return
+ self.value = value
+ for (const subscriber of subscribers) {
+ subscriber.notify()
+ }
+ },
+ value
+ } as ReactiveRef
+ return self
+}
+
+/**
+ * @since 1.0.0
+ * @category Constructors
+ */
+export const make = (value: A): Effect.Effect> => Effect.sync(() => unsafeMake(value))
+
+/**
+ * @since 1.0.0
+ * @category Combinators
+ */
+export const unsafeGet = (self: ReactiveRef): A => self.value
+
+/**
+ * @since 1.0.0
+ * @category Combinators
+ */
+export const get = (self: ReactiveRef): Effect.Effect => Effect.sync(() => self.value)
+
+/**
+ * @since 1.0.0
+ * @category Combinators
+ */
+export const subscribe = (self: ReactiveRef): Effect.Effect =>
+ Reactive.Reactive.with((reactive) => {
+ reactive.addHandle(self)
+ self.subscribe(reactive)
+ return Effect.succeed(self.value)
+ })
+
+/**
+ * @since 1.0.0
+ * @category Combinators
+ */
+export const unsafeSet: {
+ (value: A): (self: ReactiveRef) => void
+ (self: ReactiveRef, value: A): void
+} = dual(2, (self: ReactiveRef, value: A) => self.unsafeSet(value))
+
+/**
+ * @since 1.0.0
+ * @category Combinators
+ */
+export const set: {
+ (value: A): (self: ReactiveRef) => Effect.Effect
+ (self: ReactiveRef, value: A): Effect.Effect
+} = dual(2, (self: ReactiveRef, value: A) => Effect.sync(() => unsafeSet(self, value)))
+
+/**
+ * @since 1.0.0
+ * @category Combinators
+ */
+export const unsafeModify: {
+ (f: (_: A) => readonly [returnValue: R, nextValue: A]): (self: ReactiveRef) => R
+ (self: ReactiveRef, f: (_: A) => readonly [returnValue: R, nextValue: A]): R
+} = dual(2, (self: ReactiveRef, f: (_: A) => readonly [returnValue: R, nextValue: A]): R => {
+ const [returnValue, nextValue] = f(self.value)
+ unsafeSet(self, nextValue)
+ return returnValue
+})
+
+/**
+ * @since 1.0.0
+ * @category Combinators
+ */
+export const modify: {
+ (f: (_: A) => readonly [returnValue: R, nextValue: A]): (self: ReactiveRef) => Effect.Effect
+ (self: ReactiveRef, f: (_: A) => readonly [returnValue: R, nextValue: A]): Effect.Effect
+} = dual(
+ 2,
+ (self: ReactiveRef, f: (_: A) => readonly [returnValue: R, nextValue: A]): Effect.Effect =>
+ Effect.sync(() => unsafeModify(self, f))
+)
+
+/**
+ * @since 1.0.0
+ * @category Combinators
+ */
+export const unsafeUpdate: {
+ (f: (_: A) => A): (self: ReactiveRef) => void
+ (self: ReactiveRef, f: (_: A) => A): void
+} = dual(2, (self: ReactiveRef, f: (_: A) => A) => {
+ unsafeSet(self, f(self.value))
+})
+
+/**
+ * @since 1.0.0
+ * @category Combinators
+ */
+export const update: {
+ (f: (_: A) => A): (self: ReactiveRef) => Effect.Effect
+ (self: ReactiveRef, f: (_: A) => A): Effect.Effect
+} = dual(2, (self: ReactiveRef, f: (_: A) => A) => Effect.sync(() => unsafeUpdate(self, f)))
+
+/**
+ * @since 1.0.0
+ * @category Conversions
+ */
+export const intoStream: {
+ (
+ self: ReactiveRef>
+ ): (stream: Stream.Stream) => Effect.Effect
+ (
+ self: ReactiveRef>,
+ stream: Stream.Stream
+ ): Effect.Effect
+} = function() {
+ if (arguments.length === 1) {
+ const ref = arguments[0] as ReactiveRef>
+ return (stream: Stream.Stream) => intoStreamImpl(ref, stream)
+ }
+ return intoStreamImpl(arguments[0], arguments[1]) as any
+}
+
+const intoStreamImpl = (
+ self: ReactiveRef>,
+ stream: Stream.Stream
+) =>
+ Effect.uninterruptible(Effect.withFiberRuntime((parentFiber) => {
+ const scope = Context.unsafeGet(parentFiber.currentContext, Scope.Scope)
+ const runFork = Runtime.runFork(Runtime.make({
+ context: parentFiber.currentContext as Context.Context,
+ runtimeFlags: Runtime.defaultRuntime.runtimeFlags,
+ fiberRefs: parentFiber.getFiberRefs()
+ }))
+ const fiber = runFork(Stream.runForEachChunk(stream, (chunk) =>
+ Effect.sync(() => {
+ if (Chunk.isNonEmpty(chunk)) {
+ self.unsafeSet(Result.success(Chunk.lastNonEmpty(chunk), true))
+ }
+ })))
+ fiber.addObserver((exit) => {
+ if (exit._tag === "Failure") {
+ unsafeUpdate(self, (result) => Result.failureWithPrevious(exit.cause, Option.some(result)))
+ } else {
+ unsafeUpdate(
+ self,
+ Result.match({
+ onInitial(_) {
+ return Result.fail(new Cause.NoSuchElementException("No value was emitted"))
+ },
+ onFailure(_) {
+ return Result.failure(_.cause)
+ },
+ onSuccess(_) {
+ return Result.success(_.value)
+ }
+ })
+ )
+ }
+ })
+ return Scope.addFinalizer(scope, Fiber.interrupt(parentFiber))
+ }))
+
+/**
+ * @since 1.0.0
+ * @category Conversions
+ */
+export const fromStream = (stream: Stream.Stream): Effect.Effect<
+ ReactiveRef>,
+ never,
+ R | Scope.Scope
+> =>
+ Effect.suspend(() => {
+ const ref = unsafeMake>(Result.initial(true))
+ return Effect.as(intoStreamImpl(ref, stream), ref)
+ })
+
+const intoStreamPullImpl =