allow intermediate updates in optimisticFn

.changeset/fruity-years-drive.md

@@ -0,0 +1,5 @@
+---
+"@effect-rx/rx": patch
+---
+
+allow intermediate updates in optimisticFn

docs/rx/Rx.ts.md

@@ -40,6 +40,7 @@ Added in v1.0.0
 - [batching](#batching)
   - [batch](#batch)
 - [combinators](#combinators)
+  - [autoDispose](#autodispose)
   - [debounce](#debounce)
   - [initialValue](#initialvalue)
   - [keepAlive](#keepalive)
@@ -243,9 +244,7 @@ Added in v1.0.0
 **Signature**
 
 ```ts
-export declare const optimistic: <A>(
-  self: Rx<A>
-) => Writable<A, Rx<Result.Result<A extends Result.Result<infer _A, infer _E> ? _A : A, unknown>>>
+export declare const optimistic: <A>(self: Rx<A>) => Writable<A, Rx<Result.Result<A, unknown>>>
 ```
 
 Added in v1.0.0
@@ -256,15 +255,19 @@ Added in v1.0.0
 
 ```ts
 export declare const optimisticFn: {
-  <A, W, XA, XE, OW = A extends Result.Result<infer _A, infer _E> ? _A : A>(options: {
-    readonly updateToValue: (value: OW, current: NoInfer<A>) => NoInfer<W>
-    readonly fn: RxResultFn<NoInfer<OW>, XA, XE>
+  <A, W, XA, XE, OW = W>(options: {
+    readonly reducer: (current: NoInfer<A>, update: OW) => NoInfer<W>
+    readonly fn:
+      | RxResultFn<NoInfer<OW>, XA, XE>
+      | ((set: (result: NoInfer<W>) => void) => RxResultFn<NoInfer<OW>, XA, XE>)
   }): (self: Writable<A, Rx<Result.Result<W, unknown>>>) => RxResultFn<OW, XA, XE>
-  <A, W, XA, XE, OW = A extends Result.Result<infer _A, infer _E> ? _A : A>(
+  <A, W, XA, XE, OW = W>(
     self: Writable<A, Rx<Result.Result<W, unknown>>>,
     options: {
-      readonly updateToValue: (value: OW, current: NoInfer<A>) => NoInfer<W>
-      readonly fn: RxResultFn<NoInfer<OW>, XA, XE>
+      readonly reducer: (current: NoInfer<A>, update: OW) => NoInfer<W>
+      readonly fn:
+        | RxResultFn<NoInfer<OW>, XA, XE>
+        | ((set: (result: NoInfer<W>) => void) => RxResultFn<NoInfer<OW>, XA, XE>)
     }
   ): RxResultFn<OW, XA, XE>
 }
@@ -353,6 +356,21 @@ Added in v1.0.0
 
 # combinators
 
+## autoDispose
+
+Reverts the `keepAlive` behavior of a reactive value, allowing it to be
+disposed of when not in use.
+
+Note that Rx's have this behavior by default.
+
+**Signature**
+
+```ts
+export declare const autoDispose: <A extends Rx<any>>(self: A) => A
+```
+
+Added in v1.0.0
+
 ## debounce
 
 **Signature**

packages/rx/src/Rx.ts

@@ -1192,6 +1192,21 @@ export const keepAlive = <A extends Rx<any>>(self: A): A =>
     keepAlive: true
   })
 
+/**
+ * Reverts the `keepAlive` behavior of a reactive value, allowing it to be
+ * disposed of when not in use.
+ *
+ * Note that Rx's have this behavior by default.
+ *
+ * @since 1.0.0
+ * @category combinators
+ */
+export const autoDispose = <A extends Rx<any>>(self: A): A =>
+  Object.assign(Object.create(Object.getPrototypeOf(self)), {
+    ...self,
+    keepAlive: false
+  })
+
 /**
  * @since 1.0.0
  * @category combinators
@@ -1433,7 +1448,9 @@ export const optimisticFn: {
   <A, W, XA, XE, OW = W>(
     options: {
       readonly reducer: (current: NoInfer<A>, update: OW) => NoInfer<W>
-      readonly fn: RxResultFn<NoInfer<OW>, XA, XE>
+      readonly fn:
+        | RxResultFn<NoInfer<OW>, XA, XE>
+        | ((set: (result: NoInfer<W>) => void) => RxResultFn<NoInfer<OW>, XA, XE>)
     }
   ): (
     self: Writable<A, Rx<Result.Result<W, unknown>>>
@@ -1442,14 +1459,18 @@ export const optimisticFn: {
     self: Writable<A, Rx<Result.Result<W, unknown>>>,
     options: {
       readonly reducer: (current: NoInfer<A>, update: OW) => NoInfer<W>
-      readonly fn: RxResultFn<NoInfer<OW>, XA, XE>
+      readonly fn:
+        | RxResultFn<NoInfer<OW>, XA, XE>
+        | ((set: (result: NoInfer<W>) => void) => RxResultFn<NoInfer<OW>, XA, XE>)
     }
   ): RxResultFn<OW, XA, XE>
 } = dual(2, <A, W, XA, XE, OW = W>(
   self: Writable<A, Rx<Result.Result<W, unknown>>>,
   options: {
     readonly reducer: (current: NoInfer<A>, update: OW) => NoInfer<W>
-    readonly fn: RxResultFn<OW, XA, XE>
+    readonly fn:
+      | RxResultFn<NoInfer<OW>, XA, XE>
+      | ((set: (result: NoInfer<W>) => void) => RxResultFn<NoInfer<OW>, XA, XE>)
   }
 ): RxResultFn<OW, XA, XE> => {
   const transition = state<Result.Result<W, unknown>>(Result.initial())
@@ -1460,8 +1481,16 @@ export const optimisticFn: {
     }
     get.set(transition, Result.success(value, { waiting: true }))
     get.set(self, transition)
-    get.set(options.fn, arg)
-    return Effect.onExit(get.result(options.fn, { suspendOnWaiting: true }), (exit) => {
+    const fn = typeof options.fn === "function"
+      ? autoDispose(options.fn((value) =>
+        get.set(
+          transition,
+          Result.success(Result.isResult(value) ? Result.waiting(value) : value, { waiting: true })
+        )
+      ))
+      : options.fn
+    get.set(fn, arg)
+    return Effect.onExit(get.result(fn, { suspendOnWaiting: true }), (exit) => {
       get.set(transition, Result.fromExit(Exit.as(exit, value)))
       return Effect.void
     })

packages/rx/test/Rx.test.ts

@@ -1171,6 +1171,44 @@ describe("Rx", () => {
       expect(r.get(rx)).toEqual(2)
       expect(r.get(optimisticRx)).toEqual(2)
     })
+
+    it("intermediate updates", async () => {
+      const latch = Effect.unsafeMakeLatch()
+      const r = Registry.make()
+      let i = 0
+      const rx = Rx.make(Effect.sync(() => i))
+      const optimisticRx = rx.pipe(
+        Rx.optimistic
+      )
+      const fn = optimisticRx.pipe(
+        Rx.optimisticFn({
+          reducer: (_current, update: number) => Result.success(update),
+          fn: (set) =>
+            Rx.fn(Effect.fnUntraced(function*() {
+              set(Result.success(123))
+              yield* latch.await
+            }))
+        }),
+        Rx.keepAlive
+      )
+
+      expect(r.get(rx)).toEqual(Result.success(0))
+      assert.deepStrictEqual(r.get(optimisticRx), Result.success(0))
+      r.set(fn, 1)
+      i = 2
+
+      // optimistic phase: the intermediate value is set, but the true value is
+      // not
+      assert.deepStrictEqual(r.get(rx), Result.success(0))
+      assert.deepStrictEqual(r.get(optimisticRx), Result.success(123, { waiting: true }))
+
+      latch.unsafeOpen()
+      await Effect.runPromise(Effect.yieldNow())
+
+      // commit phase: a refresh is triggered, the authoritative value is used
+      assert.deepStrictEqual(r.get(rx), Result.success(2))
+      assert.deepStrictEqual(r.get(optimisticRx), Result.success(2))
+    })
   })
 })