API improvements

- Add 2-argument variant to pipe
- Add stream.subject
- Rename stream.stream to stream.generate
- Rename signal.signal to signal.state (similar to TC39 signals)

Author: gordonbrander

typescript/packages/common-frp/README.md

@@ -12,7 +12,6 @@ Common functional reactive programming utilities.
     - [x] All graph dependencies held in array (in contrast to S.js style or Signals style where graph is defined via VM execution trace)
 - [x] Glitch free: updates happen during discrete moments using a transaction system
 - [x] Transformations should happen during same transaction
-- [ ] Transactions
-    - [ ] (see experimental for WIP)
-- [ ] Cycles
-    - [ ] TODO: switch from microtask-based topological sorting to transaction-based push-pull (see experimental for WIP)
+- [x] Transactions
+- [ ] Cycles supported for Signals
+    - [ ] TODO need to improve [push-pull dirty marking logic](https://github.com/tc39/proposal-signals?tab=readme-ov-file#common-algorithms).

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

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

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

@@ -11,6 +11,11 @@ import {
 export type UnaryFn<A, B> = (a: A) => B
 
 export type Pipe = {
+  <A, B>(
+    value: A,
+    a2b: UnaryFn<A, B>
+  ): B
+
   <A, B, C>(
     value: A,
     a2b: UnaryFn<A, B>,

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

@@ -92,7 +92,7 @@ export const effect = <T>(
   return signal[__updates__](() => withReads(job))
 }
 
-export const signal = <T>(initial: T) => {
+export const state = <T>(initial: T) => {
   const updates = publisher<void>()
 
   let state = initial

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

@@ -1,6 +1,6 @@
 import { debug } from "./shared.js"
 import { publisher, Send, Cancel, combineCancels } from "./publisher.js"
-import { signal as signal, Signal, __updates__ } from "./signal.js"
+import { state, Signal, __updates__ } from "./signal.js"
 
 const __sink__ = Symbol('sink')
 
@@ -18,20 +18,33 @@ export const sink = <T>(
   subscriber: Send<T>
 ) => upstream[__sink__](subscriber)
 
-/** Create a new stream source */
-export const stream = <T>(
+/**
+ * Create a stream subject - a source for a stream that has a send method
+ * for publishing new items to stream.
+ */
+export const subject = <T>() => {
+  const { pub, sub } = publisher<T>()
+  return { [__sink__]: sub, send: pub }
+}
+
+/**
+ * Create a new stream source using a closure to publish new items to stream
+ * Closure receives a single argument, a send function to publish new items,
+ * and may return a cancel function to stop generation and clean up resources.
+ */
+export const generate = <T>(
   generate: (send: Send<T>) => Cancel | undefined
 ): Stream<T> => {
-  const { pub, sub } = publisher<T>()
-  const cancel = generate(pub)
-  return { [__sink__]: sub, cancel }
+  const { [__sink__]: sink, send } = subject<T>()
+  const cancel = generate(send)
+  return { [__sink__]: sink, cancel }
 }
 
 /** Map a stream of values */
 export const map = <T, U>(
   upstream: Stream<T>,
   transform: (value: T) => U
-) => stream<U>(send => {
+) => generate<U>(send => {
   return sink(upstream, value => send(transform(value)))
 })
 
@@ -51,7 +64,7 @@ export const select = <T extends object, U extends keyof T & string>(
 export const filter = <T>(
   upstream: Stream<T>,
   predicate: (value: T) => boolean
-) => stream<T>(send => {
+) => generate<T>(send => {
   return sink(upstream, value => {
     if (predicate(value)) {
       send(value)
@@ -67,7 +80,7 @@ export const zip = <T, U, V>(
   left: Stream<T>,
   right: Stream<U>,
   combine: (left: T, right: U) => V
-) => stream<V>(send => {
+) => generate<V>(send => {
   const leftQueue: Array<T> = []
   const rightQueue: Array<U> = []
 
@@ -100,7 +113,7 @@ export const scan = <T, U>(
   step: (state: U, value: T) => U,
   initial: U
 ): Signal<U> => {
-  const { get, [__updates__]: updates, send } = signal(initial)
+  const { get, [__updates__]: updates, send } = state(initial)
   const cancel = sink(upstream, (value: T) => {
     send(step(get(), value))
   })