Rx.family

.changeset/lovely-parents-do.md

@@ -0,0 +1,5 @@
+---
+"@effect-rx/rx": patch
+---
+
+Rx.family

docs/rx/Rx.ts.md

@@ -21,6 +21,7 @@ Added in v1.0.0
 - [constructors](#constructors)
   - [effect](#effect)
   - [effectFn](#effectfn)
+  - [family](#family)
   - [fn](#fn)
   - [readable](#readable)
   - [runtime](#runtime)
@@ -137,6 +138,16 @@ export declare const effectFn: {
 
 Added in v1.0.0
 
+## family
+
+**Signature**
+
+```ts
+export declare const family: <Arg, T extends Rx<any>>(f: (arg: Arg) => T) => (arg: Arg) => T
+```
+
+Added in v1.0.0
+
 ## fn
 
 **Signature**

packages/rx/src/Rx.ts

@@ -607,6 +607,30 @@ export const streamPull: {
   })
 }
 
+/**
+ * @since 1.0.0
+ * @category constructors
+ */
+export const family = <Arg, T extends Rx<any>>(
+  f: (arg: Arg) => T
+): (arg: Arg) => T => {
+  const atoms = new Map<number, WeakRef<T>>()
+  const registry = new FinalizationRegistry<number>((hash) => {
+    atoms.delete(hash)
+  })
+  return function(arg) {
+    const hash = Hash.hash(arg)
+    const atom = atoms.get(hash)?.deref()
+    if (atom !== undefined) {
+      return atom
+    }
+    const newAtom = f(arg)
+    atoms.set(hash, new WeakRef(newAtom))
+    registry.register(newAtom, hash)
+    return newAtom
+  }
+}
+
 /**
  * @since 1.0.0
  * @category combinators

packages/rx/src/internal/registry.ts

@@ -18,7 +18,7 @@ class RegistryImpl implements Registry.Registry {
     this[TypeId] = TypeId
   }
 
-  private readonly nodes = new WeakMap<Rx.Rx<any>, Node<any>>()
+  private readonly nodes = new Map<Rx.Rx<any>, Node<any>>()
 
   get = <A>(rx: Rx.Rx<A>): A => {
     return this.ensureNode(rx).value()

packages/rx/test/Rx.test.ts

@@ -2,6 +2,7 @@ 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 * as Context from "@effect/data/Context"
+import * as Hash from "@effect/data/Hash"
 import * as Option from "@effect/data/Option"
 import * as Effect from "@effect/io/Effect"
 import * as Layer from "@effect/io/Layer"
@@ -221,6 +222,29 @@ describe("Rx", () => {
     assert(Result.isWaiting(result))
     assert(Option.isNone(Result.value(result)))
   })
+
+  it("family", async () => {
+    const r = Registry.make()
+
+    const count = Rx.family((n: number) => Rx.state(n))
+    const hash = Hash.hash(count(1))
+    assert.strictEqual(count(1), count(1))
+    r.set(count(1), 2)
+    assert.strictEqual(r.get(count(1)), 2)
+
+    const countKeep = Rx.family((n: number) => Rx.state(n).pipe(Rx.keepAlive))
+    assert.strictEqual(countKeep(1), countKeep(1))
+    r.get(countKeep(1))
+    const hashKeep = Hash.hash(countKeep(1))
+
+    if (global.gc) {
+      vi.useRealTimers()
+      await new Promise((resolve) => setTimeout(resolve, 0))
+      global.gc()
+      assert.notEqual(hash, Hash.hash(count(1)))
+      assert.strictEqual(hashKeep, Hash.hash(countKeep(1)))
+    }
+  })
 })
 
 interface Counter {