Build FRP stream graph out of subscribers (#44)

Stream publishers now hold a list of Sendables (subscribers with a send method).

Holding a list of subscribers allows us to provide traversal and serialization methods on subscribers, while returning a readonly handle to derived streams.

Author: gordonbrander

typescript/packages/common-frp/src/dom.ts

@@ -1,9 +1,13 @@
-import { generate } from './stream.js'
+import { subject, readonly, __sink__ } from './stream.js'
 
 export const events = (
   element: HTMLElement,
   name: string
-) => generate((send: (value: Event) => void) => {
-  element.addEventListener(name, send)
-  return () => element.removeEventListener(name, send)
-})
+) => {
+  const event = subject()
+  element.addEventListener(name, event.send)
+  return readonly({
+    ...event,
+    cancel: () => element.removeEventListener(name, event.send)
+  })
+}

typescript/packages/common-frp/src/operators.ts

@@ -1,6 +1,5 @@
 import {
   map as mapStream,
-  select as selectStream,
   filter as filterStream,
   join as joinStreams,
   chooseLeft,
@@ -99,12 +98,6 @@ export const map = <T, U>(
   stream: Stream<T>
 ) => mapStream(stream, transform)
 
-export const select = <T extends object, U extends keyof T & string>(
-  key: U
-) => (
-  stream: Stream<T>
-) => selectStream(stream, key)
-
 /** Filter a stream of values using a predicate function. */
 export const filter = <T>(
   predicate: UnaryFn<T, boolean>

typescript/packages/common-frp/src/publisher.ts

@@ -13,13 +13,19 @@ export type Sendable<T> = {
   send: Send<T>
 }
 
+export type Publisher<T> = {
+  [Symbol.iterator]: () => Iterator<Sendable<T>>;
+  send: (value: T) => void;
+  sink: (subscriber: Sendable<T>) => Cancel;
+}
+
 /** Low-level pub-sub channel used under the hood. */
-export const publisher = <T>() => {
-  const subscribers = new Set<Send<T>>()
+export const publisher = <T>(): Publisher<T> => {
+  const subscribers = new Set<Sendable<T>>()
 
-  const pub = (value: T) => {
+  const send = (value: T) => {
     for (const subscriber of subscribers) {
-      subscriber(value)
+      subscriber.send(value)
     }
   }
 
@@ -28,7 +34,7 @@ export const publisher = <T>() => {
    * @param subscriber the function to call when a new value is published
    * @returns Unsubscribe function
    */
-  const sub = (subscriber: Send<T>): Cancel => {
+  const sink = (subscriber: Sendable<T>): Cancel => {
     debug('sub', 'subscribing', subscriber)
     subscribers.add(subscriber)
     return () => {
@@ -40,8 +46,8 @@ export const publisher = <T>() => {
   return {
     // Allow iterating over subscribers
     [Symbol.iterator]: () => subscribers.values(),
-    pub,
-    sub
+    send,
+    sink
   }
 }
 

typescript/packages/common-frp/src/signal.ts

@@ -1,7 +1,6 @@
 import { debug } from './shared.js'
 import {
   publisher,
-  Send,
   Sendable,
   Cancel,
   Cancellable,
@@ -65,7 +64,7 @@ const { withUpdates, withReads } = createTransactionManager()
 export const __updates__ = Symbol('updates')
 
 export type Updates<T> = {
-  [__updates__]: (subscriber: Send<T>) => Cancel
+  [__updates__]: (subscriber: Sendable<T>) => Cancel
 }
 
 const isEqual = Object.is
@@ -160,8 +159,10 @@ export const effect: Effect = (
 
   job()
 
+  const subscriber = { send: schedule }
+
   return combineCancels(
-    upstreams.map(signal => signal[__updates__](schedule))
+    upstreams.map(signal => signal[__updates__](subscriber))
   )
 }
 
@@ -177,7 +178,7 @@ export const state = <T>(initial: T) => {
     if (!isEqual(state, value)) {
       state = value
       debug('state', 'value updated', state)
-      updates.pub()
+      updates.send()
     }
   }
 
@@ -186,7 +187,7 @@ export const state = <T>(initial: T) => {
   return {
     get,
     send,
-    [__updates__]: updates.sub
+    [__updates__]: updates.sink
   }
 }
 
@@ -280,11 +281,13 @@ export const computed: Computed = (
   const performUpdate = () => {
     debug('computed', 'mark dirty')
     isDirty = true
-    updates.pub()
+    updates.send()
   }
 
+  const subscriber = { send: performUpdate }
+
   const cancel = combineCancels(
-    upstreams.map(signal => signal[__updates__](performUpdate))
+    upstreams.map(signal => signal[__updates__](subscriber))
   )
 
   const get = () => {
@@ -298,7 +301,7 @@ export const computed: Computed = (
 
   return {
     get,
-    [__updates__]: updates.sub,
+    [__updates__]: updates.sink,
     cancel
   }
 }