fix react store & remove some apis

.changeset/few-cooks-tap.md

@@ -0,0 +1,6 @@
+---
+"@effect-rx/rx-react": patch
+"@effect-rx/rx": patch
+---
+
+fix react store & remove some apis

packages/rx-react/src/index.ts

@@ -20,17 +20,19 @@ class RxStore<A> {
     readonly registry: Registry.Registry,
     readonly rx: Rx.Rx<A>
   ) {}
-
-  value!: A
-
+  value = this.registry.get(this.rx)
+  init = false
   subscribe = (f: () => void): () => void => {
-    this.value = this.registry.get(this.rx)
+    if (this.init === true) {
+      this.value = this.registry.get(this.rx)
+    } else {
+      this.init = true
+    }
     return this.registry.subscribe(this.rx, (a) => {
       this.value = a
       f()
     })
   }
-
   snapshot = (): A => this.value
 }
 

packages/rx/src/Registry.ts

@@ -27,8 +27,6 @@ export interface Registry {
   readonly refresh: Rx.Rx.Refresh
   readonly set: Rx.Rx.Set
   readonly subscribe: Rx.Rx.Subscribe
-  readonly subscribeWithPrevious: Rx.Rx.SubscribeWithPrevious
-  readonly queue: Rx.Rx.Queue
 }
 
 /**

packages/rx/src/Rx.ts

@@ -151,10 +151,6 @@ export interface Context<A> {
   readonly subscribe: <A>(rx: Rx<A>, f: (_: A) => void, options?: {
     readonly immediate?: boolean
   }) => void
-  readonly subscribeWithPrevious: <A>(rx: Rx<A>, f: (prev: Option.Option<A>, value: A) => void, options?: {
-    readonly immediate?: boolean
-  }) => void
-  readonly queue: <A>(rx: Rx<A>) => Effect.Effect<never, never, Queue_.Dequeue<A>>
 }
 
 /**

packages/rx/src/internal/registry.ts

@@ -2,10 +2,6 @@ import type * as Registry from "@effect-rx/rx/Registry"
 import type * as Rx from "@effect-rx/rx/Rx"
 import * as Equal from "@effect/data/Equal"
 import * as Option from "@effect/data/Option"
-import * as Effect from "@effect/io/Effect"
-import * as Exit from "@effect/io/Exit"
-import * as Queue from "@effect/io/Queue"
-import * as Scope from "@effect/io/Scope"
 
 const constImmediate = { immediate: true }
 function constListener(_: any) {}
@@ -63,40 +59,10 @@ class RegistryImpl implements Registry.Registry {
     }
   }
 
-  subscribeWithPrevious: Rx.Rx.SubscribeWithPrevious = <A>(
-    rx: Rx.Rx<A>,
-    f: (prev: Option.Option<A>, value: A) => void,
-    options?: { readonly immediate?: boolean }
-  ): () => void => {
-    let prev = Option.none<A>()
-    function listener(a: A) {
-      const old = prev
-      prev = Option.some(a)
-      f(old, a)
-    }
-    return this.subscribe(rx, listener, options)
-  }
-
   mount<A>(rx: Rx.Rx<A>) {
     return this.subscribe(rx, constListener, constImmediate)
   }
 
-  queue<A>(rx: Rx.Rx<A>) {
-    return Effect.tap(Queue.unbounded<A>(), (queue) => {
-      const offer = Effect.async<never, never, never>(() => {
-        const cancel = this.subscribe(rx, (a) => {
-          Queue.unsafeOffer(queue, a)
-        }, constImmediate)
-        return Effect.sync(cancel)
-      })
-      const shutdown = Queue.shutdown(queue)
-      return Effect.zipRight(
-        Effect.addFinalizer(() => shutdown),
-        Effect.forkScoped(offer)
-      )
-    })
-  }
-
   ensureNode<A>(rx: Rx.Rx<A>): Node<A> {
     let node = this.nodes.get(rx)
     if (node === undefined) {
@@ -358,24 +324,12 @@ class Lifetime<A> implements Rx.Context<A> {
     this.node.invalidate()
   }
 
-  queue<A>(rx: Rx.Rx<A>): Effect.Effect<never, never, Queue.Dequeue<A>> {
-    const scope = Effect.runSync(Scope.make())
-    this.addFinalizer(() => Effect.runFork(Scope.close(scope, Exit.unit)))
-    return Effect.provideService(this.node.registry.queue(rx), Scope.Scope, scope)
-  }
-
   subscribe<A>(rx: Rx.Rx<A>, f: (_: A) => void, options?: {
     readonly immediate?: boolean
   }): void {
     this.addFinalizer(this.node.registry.subscribe(rx, f, options))
   }
 
-  subscribeWithPrevious<A>(rx: Rx.Rx<A>, f: (prev: Option.Option<A>, value: A) => void, options?: {
-    readonly immediate?: boolean
-  }): void {
-    this.addFinalizer(this.node.registry.subscribeWithPrevious(rx, f, options))
-  }
-
   setSelf(a: A): void {
     this.node.setValue(a)
   }