Add pipe and operators

sketches/2024-05-30-common-frp/src/dom.ts

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

sketches/2024-05-30-common-frp/src/operators.ts

@@ -0,0 +1,111 @@
+import {
+  map as mapStream,
+  filter as filterStream,
+  zip as zipStreams,
+  scan as scanStream,
+  hold as holdStream,
+  Stream
+} from './stream'
+
+export type UnaryFn<A, B> = (a: A) => B
+
+export type Pipe = {
+  <A, B, C>(
+    value: A,
+    a2b: UnaryFn<A, B>,
+    b2c: UnaryFn<B, C>
+  ): C
+
+  <A, B, C, D>(
+    value: A,
+    a2b: UnaryFn<A, B>,
+    b2c: UnaryFn<B, C>,
+    c2d: UnaryFn<C, D>
+  ): D
+
+  <A, B, C, D, E>(
+    value: A,
+    a2b: UnaryFn<A, B>,
+    b2c: UnaryFn<B, C>,
+    c2d: UnaryFn<C, D>,
+    d2e: UnaryFn<D, E>
+  ): E
+
+  <A, B, C, D, E, F>(
+    value: A,
+    a2b: UnaryFn<A, B>,
+    b2c: UnaryFn<B, C>,
+    c2d: UnaryFn<C, D>,
+    d2e: UnaryFn<D, E>,
+    e2f: UnaryFn<E, F>
+  ): F
+
+  <A, B, C, D, E, F, G>(
+    value: A,
+    a2b: UnaryFn<A, B>,
+    b2c: UnaryFn<B, C>,
+    c2d: UnaryFn<C, D>,
+    d2e: UnaryFn<D, E>,
+    e2f: UnaryFn<E, F>,
+    f2g: UnaryFn<F, G>
+  ): G
+
+  <A, B, C, D, E, F, G, H>(
+    value: A,
+    a2b: UnaryFn<A, B>,
+    b2c: UnaryFn<B, C>,
+    c2d: UnaryFn<C, D>,
+    d2e: UnaryFn<D, E>,
+    e2f: UnaryFn<E, F>,
+    f2g: UnaryFn<F, G>,
+    g2h: UnaryFn<G, H>
+  ): H
+
+  <A, B, C, D, E, F, G, H, I>(
+    value: A,
+    a2b: UnaryFn<A, B>,
+    b2c: UnaryFn<B, C>,
+    c2d: UnaryFn<C, D>,
+    d2e: UnaryFn<D, E>,
+    e2f: UnaryFn<E, F>,
+    f2g: UnaryFn<F, G>,
+    g2h: UnaryFn<G, H>,
+    h2i: UnaryFn<H, I>
+  ): I
+}
+
+/** Pipe a value through a series of functions */
+export const pipe: Pipe = (
+  value: any,
+  ...fns: [(value: any) => any]
+): any => fns.reduce((value: any, fn: UnaryFn<any, any>) => fn(value), value)
+
+/** Map a stream of values */
+export const map = <T, U>(
+  transform: UnaryFn<T, U>
+) => (stream: Stream<T>) => mapStream(stream, transform)
+
+/** Filter a stream of values using a predicate function. */
+export const filter = <T>(
+  predicate: UnaryFn<T, boolean>
+) => (stream: Stream<T>) => filterStream(stream, predicate)
+
+/** Scan a stream, accumulating step state in a cell */
+export const scan = <T, U>(
+  step: (state: U, value: T) => U, initial: U
+) => (stream: Stream<T>) => scanStream(stream, step, initial)
+
+export const hold = <T>(
+  initial: T
+) => (stream: Stream<T>) => holdStream(stream, initial)
+
+export const zip = <T, U, V>(
+  right: Stream<U>
+) => (
+  left: Stream<T>,
+  combine: (left: T, right: U) => V,
+) => zipStreams(
+  left,
+  right,
+  combine
+)

sketches/2024-05-30-common-frp/src/signal.ts

@@ -89,7 +89,7 @@ const sample = <T>(container: Gettable<T>) => container.get()
 export type Signal<T> = Gettable<T> & Updates<void> & Sink<T> & Unsubscribable
 export type SignalSubject<T> = Gettable<T> & Updates<void> & Sink<T> & Subject<T>
 
-export const create = <T>(initial: T) => {
+export const createSignal = <T>(initial: T) => {
   const updates = createPublisher<void>()
 
   let state = initial
@@ -121,7 +121,7 @@ export const create = <T>(initial: T) => {
   }
 }
 
-export type computed = {
+export type createComputed = {
   <A, B, Z>(
     upstreams: [Signal<A>, Signal<B>],
     compute: (a: A, b: B) => Z
@@ -181,7 +181,7 @@ export type computed = {
   ): Signal<Z>
 }
 
-export const computed: computed = (
+export const computed: createComputed = (
   upstreams: Array<Signal<any>>,
   compute: (...values: Array<any>) => any
 ): Signal<any> => {

sketches/2024-05-30-common-frp/src/stream.ts

@@ -1,13 +1,13 @@
 import { debug } from "./shared"
 import { createPublisher, Send, Cancel, combineCancels } from "./publisher"
-import { create as createSignal, Signal, __updates__ } from "./signal"
+import { createSignal as createSignal, Signal, __updates__ } from "./signal"
 
 export type Stream<T> = {
   sink: (subscriber: Send<T>) => Cancel
   cancel?: Cancel
 }
 
-export const create = <T>(
+export const createStream = <T>(
   generate: (send: Send<T>) => Cancel|undefined
 ): Stream<T> => {
   const {pub, sub: sink} = createPublisher<T>()
@@ -18,14 +18,14 @@ export const create = <T>(
 export const map = <T, U>(
   stream: Stream<T>,
   transform: (value: T) => U
-) => create<U>(send => {
+) => createStream<U>(send => {
   return stream.sink((value: T) => send(transform(value)))
 })
 
 export const filter = <T>(
   stream: Stream<T>,
   predicate: (value: T) => boolean
-) => create<T>(send => {
+) => createStream<T>(send => {
   return stream.sink(value => {
     if (predicate(value)) {
       send(value)
@@ -37,7 +37,7 @@ export const zip = <T, U, V>(
   left: Stream<T>,
   right: Stream<U>,
   combine: (left: T, right: U) => V
-) => create<V>(send => {
+) => createStream<V>(send => {
   const leftQueue: Array<T> = []
   const rightQueue: Array<U> = []
 
@@ -84,6 +84,6 @@ export const scan = <T, U>(
  * Hold the latest value from a stream in a cell.
  */
 export const hold = <T>(
-  stream: Signal<T>,
+  stream: Stream<T>,
   initial: T
 ) => scan(stream, (_, value) => value, initial)