feat(utils): createSignal & createKeyedSignal

packages/utils/src/createKeyedSignal.spec.ts

@@ -0,0 +1,55 @@
+import { createKeyedSignal } from "./"
+
+describe("createKeyedSignal", () => {
+  it("receives a key selector and a mapper and returns a tuple with an observable-getter and its corresponding event-emitter", () => {
+    const [getFooBar$, onFooBar] = createKeyedSignal(
+      (x) => x.key,
+      (foo: number, bar: string, key: string) => ({ foo, bar, key }),
+    )
+
+    let receivedValue1
+    let nHits1 = 0
+    const subscription1 = getFooBar$("key").subscribe((val) => {
+      receivedValue1 = val
+      nHits1++
+    })
+
+    expect(receivedValue1).toBe(undefined)
+    onFooBar(0, "1", "key")
+    expect(receivedValue1).toEqual({ foo: 0, bar: "1", key: "key" })
+    expect(nHits1).toBe(1)
+
+    let receivedValue2
+    let nHits2 = 0
+    const subscription2 = getFooBar$("key").subscribe((val) => {
+      receivedValue2 = val
+      nHits2++
+    })
+
+    expect(receivedValue2).toBe(undefined)
+
+    onFooBar(1, "2", "key")
+    expect(receivedValue1).toEqual({ foo: 1, bar: "2", key: "key" })
+    expect(nHits1).toBe(2)
+    expect(receivedValue2).toEqual({ foo: 1, bar: "2", key: "key" })
+    expect(nHits2).toBe(1)
+
+    onFooBar(1, "2", "key2")
+    expect(nHits1).toBe(2)
+    expect(nHits2).toBe(1)
+
+    subscription1.unsubscribe()
+    subscription2.unsubscribe()
+  })
+
+  it('returns a tuple with a typed observable and its corresponding event-emitter when no "event creator" is provided', () => {
+    const [foo$, onFoo] = createKeyedSignal<string>()
+    let receivedValue
+    foo$("foo").subscribe((val) => {
+      receivedValue = val
+    })
+    expect(receivedValue).toBe(undefined)
+    onFoo("foo")
+    expect(receivedValue).toEqual("foo")
+  })
+})

packages/utils/src/createKeyedSignal.ts

@@ -0,0 +1,72 @@
+import { GroupedObservable, identity, Observable, Observer } from "rxjs"
+
+/**
+ * Creates a "keyed" signal. It's sugar for splitting the Observer and the Observable of a keyed signal.
+ *
+ * @returns [1, 2]
+ * 1. The getter function that returns the GroupedObservable<T, T>
+ * 2. The emitter function.
+ */
+export function createKeyedSignal<T>(): [
+  (key: T) => GroupedObservable<T, T>,
+  (key: T) => void,
+]
+
+/**
+ * Creates a "keyed" signal. It's sugar for splitting the Observer and the Observable of a keyed signal.
+ *
+ * @param keySelector a function that extracts the key from the emitted value
+ * @returns [1, 2]
+ * 1. The getter function that returns the GroupedObservable<K, T>
+ * 2. The emitter function.
+ */
+export function createKeyedSignal<T, K, A extends any[]>(
+  keySelector: (signal: T) => K,
+): [(key: K) => GroupedObservable<K, T>, (key: K) => void]
+
+/**
+ * Creates a "keyed" signal. It's sugar for splitting the Observer and the Observable of a keyed signal.
+ *
+ * @param keySelector a function that extracts the key from the emitted value
+ * @param mapper a function that maps the arguments of the emitter function to the value of the GroupedObservable
+ * @returns [1, 2]
+ * 1. The getter function that returns the GroupedObservable<K, T>
+ * 2. The emitter function (...args: any[]) => T.
+ */
+export function createKeyedSignal<T, K, A extends any[]>(
+  keySelector: (signal: T) => K,
+  mapper: (...args: A) => T,
+): [(key: K) => GroupedObservable<K, T>, (...args: A) => void]
+
+export function createKeyedSignal<T, K, A extends any[]>(
+  keySelector: (signal: T) => K = identity as any,
+  mapper: (...args: A) => T = identity as any,
+): [(key: K) => GroupedObservable<K, T>, (...args: A) => void] {
+  const observersMap = new Map<K, Set<Observer<T>>>()
+
+  return [
+    (key: K) => {
+      const res = new Observable<T>((observer) => {
+        if (!observersMap.has(key)) {
+          observersMap.set(key, new Set())
+        }
+        const set = observersMap.get(key)!
+        set.add(observer)
+        return () => {
+          set.delete(observer)
+          if (set.size === 0) {
+            observersMap.delete(key)
+          }
+        }
+      }) as GroupedObservable<K, T>
+      res.key = key
+      return res
+    },
+    (...args: A) => {
+      const payload = mapper(...args)
+      observersMap.get(keySelector(payload))?.forEach((o) => {
+        o.next(payload)
+      })
+    },
+  ]
+}

packages/utils/src/createListener.ts

@@ -1,16 +1,24 @@
-import { Observable, Subject } from "rxjs"
-
-const defaultMapper: any = (v: unknown) => v
+import { Observable } from "rxjs"
+import { createSignal } from "./createSignal"
 
+/** @deprecated createListener is deprecated and it will be removed in the next version, please use createSignal. */
 export function createListener<A extends unknown[], T>(
   mapper: (...args: A) => T,
 ): [Observable<T>, (...args: A) => void]
+
+/** @deprecated createListener is deprecated and it will be removed in the next version, please use createSignal. */
 export function createListener(): [Observable<void>, () => void]
+
+/** @deprecated createListener is deprecated and it will be removed in the next version, please use createSignal. */
 export function createListener<T>(): [Observable<T>, (payload: T) => void]
 
-export function createListener<A extends unknown[], T>(
-  mapper: (...args: A) => T = defaultMapper,
-): [Observable<T>, (...args: A) => void] {
-  const subject = new Subject<T>()
-  return [subject.asObservable(), (...args: A) => subject.next(mapper(...args))]
+/**
+ * Creates a void signal. It's sugar for splitting the Observer and the Observable of a signal.
+ *
+ * @returns [1, 2]
+ * 1. The Observable
+ * 2. The emitter function.
+ */
+export function createListener(...args: any[]) {
+  return (createSignal as any)(...args)
 }

packages/utils/src/createSignal.spec.ts

@@ -1,8 +1,9 @@
+import { createSignal } from "./"
 import { createListener } from "./"
 
-describe("createListener", () => {
+describe("createSignal", () => {
   it('receives an "event creator" and it returns a tuple with an observable and its corresponding event-emitter', () => {
-    const [fooBar$, onFooBar] = createListener((foo: number, bar: string) => ({
+    const [fooBar$, onFooBar] = createSignal((foo: number, bar: string) => ({
       foo,
       bar,
     }))
@@ -15,7 +16,7 @@ describe("createListener", () => {
     expect(receivedValue).toEqual({ foo: 0, bar: "1" })
   })
   it('returns a tuple with a typed observable and its corresponding event-emitter when no "event creator" is provided', () => {
-    const [foo$, onFoo] = createListener<string>()
+    const [foo$, onFoo] = createSignal<string>()
     let receivedValue
     foo$.subscribe((val) => {
       receivedValue = val

packages/utils/src/createSignal.ts

@@ -0,0 +1,39 @@
+import { identity, Observable, Subject } from "rxjs"
+
+/**
+ * Creates a signal. It's sugar for splitting the Observer and the Observable of a signal.
+ *
+ * @param mapper a mapper function, for mapping the arguments of the emitter function into
+ * the value of the Observable.
+ * @returns [1, 2]
+ * 1. The Observable<T>
+ * 2. The emitter function.
+ */
+export function createSignal<A extends unknown[], T>(
+  mapper: (...args: A) => T,
+): [Observable<T>, (...args: A) => void]
+
+/**
+ * Creates a void signal. It's sugar for splitting the Observer and the Observable of a signal.
+ *
+ * @returns [1, 2]
+ * 1. The Observable<void>
+ * 2. The emitter function.
+ */
+export function createSignal(): [Observable<void>, () => void]
+
+/**
+ * Creates a signal. It's sugar for splitting the Observer and the Observable of a signal.
+ *
+ * @returns [1, 2]
+ * 1. The Observable<T>
+ * 2. The emitter function.
+ */
+export function createSignal<T>(): [Observable<T>, (payload: T) => void]
+
+export function createSignal<A extends unknown[], T>(
+  mapper: (...args: A) => T = identity as any,
+): [Observable<T>, (...args: A) => void] {
+  const subject = new Subject<T>()
+  return [subject.asObservable(), (...args: A) => subject.next(mapper(...args))]
+}

packages/utils/src/index.tsx

@@ -1,7 +1,8 @@
 export { collectValues } from "./collectValues"
 export { collect } from "./collect"
 export { getGroupedObservable } from "./getGroupedObservable"
-export { createListener } from "./createListener"
+export { createSignal } from "./createSignal"
+export { createKeyedSignal } from "./createKeyedSignal"
 export { mergeWithKey } from "./mergeWithKey"
 export { split } from "./split"
 export { suspend } from "./suspend"
@@ -9,3 +10,5 @@ export { suspended } from "./suspended"
 export { switchMapSuspended } from "./switchMapSuspended"
 export { selfDependant } from "./selfDependant"
 export { contextBinder } from "./contextBinder"
+
+export { createListener } from "./createListener"