some codata examples in javascript/typescript

codata.ts

interface BtreeVisitor<T, TT> {
  leaf: (value: T) => TT;
  branch: (left: TT, right: TT) => TT;
}

interface Btree<T> {
  <TT>(visitor: BtreeVisitor<T, TT>): TT;
}

const leaf = <T>(value: T): Btree<T> => {
  return (visitor) => visitor.leaf(value);
};

const branch = <T>(left: Btree<T>, right: Btree<T>): Btree<T> => {
  return (visitor) => {
    return visitor.branch(left(visitor), right(visitor));
  };
};

// reverse binary tree
const reverse = <T>(btree: Btree<T>): Btree<T> => {
  return btree({
    leaf: (value) => leaf(value),
    branch: (left, right) => branch(right, left),
  });
};

type Data_BTree<T> =
  | {
      type: 'leaf';
      value: T;
    }
  | {
      type: 'branch';
      left: Data_BTree<T>;
      right: Data_BTree<T>;
    };

// data constructors
const dataBtreeConstructors = {
  leaf: <T>(value: T): Data_BTree<T> => ({
    type: 'leaf',
    value,
  }),
  branch: <T>(left: Data_BTree<T>, right: Data_BTree<T>): Data_BTree<T> => ({
    type: 'branch',
    left,
    right,
  }),
};

const toDataBTree = <T>(btree: Btree<T>): Data_BTree<T> => {
  // codata to data is easy
  return btree({
    leaf: dataBtreeConstructors.leaf,
    branch: dataBtreeConstructors.branch,
  });
};

const toSourceCode = <T>(btree: Btree<T>): string => {
  return btree({
    leaf: (value) => `leaf(${value})`,
    branch: (left, right) => `branch(${left}, ${right})`,
  });
};

const btree = branch(
  branch(leaf(0), branch(leaf(1), leaf(2))),
  branch(leaf(3), leaf(4))
);

const btreeData = toDataBTree(btree);
const btreeCode = toSourceCode(btree);

const reversedBtree = reverse(btree);
const reversedBtreeData = toDataBTree(reversedBtree);
const reversedBtreeCode = toSourceCode(reversedBtree);

console.log('btreeData', JSON.stringify(btreeData, null, 2));
console.log('btreeCode', btreeCode);
console.log('reversedBtreeData', JSON.stringify(reversedBtreeData, null, 2));
console.log('reversedBtreeCode', reversedBtreeCode);

streaming via codata

interface UnaryFunction<T, TT> {
  (arg: T): TT;
}

function pipe<T, A>(a: T, f1: UnaryFunction<T, A>): UnaryFunction<T, A>;
function pipe<T, A, B>(
  a: T,
  f1: UnaryFunction<T, A>,
  f2: UnaryFunction<A, B>
): UnaryFunction<T, B>;
function pipe<T, A, B, C>(
  a: T,
  f1: UnaryFunction<T, A>,
  f2: UnaryFunction<A, B>,
  f3: UnaryFunction<B, C>
): UnaryFunction<T, C>;
function pipe<T, A, B, C>(
  a: T,
  f1: UnaryFunction<T, A>,
  f2: UnaryFunction<A, B>,
  f3: UnaryFunction<B, C>
): UnaryFunction<T, C>;
function pipe<T, A, B, C, D>(
  a: T,
  f1: UnaryFunction<T, A>,
  f2: UnaryFunction<A, B>,
  f3: UnaryFunction<B, C>,
  f4: UnaryFunction<C, D>
): UnaryFunction<T, D>;
function pipe<T, A, B, C, D, E>(
  a: T,
  f1: UnaryFunction<T, A>,
  f2: UnaryFunction<A, B>,
  f3: UnaryFunction<B, C>,
  f4: UnaryFunction<C, D>,
  f5: UnaryFunction<D, E>
): UnaryFunction<T, E>;

function pipe(a: any, ...args: UnaryFunction<any, any>[]) {
  return args.reduce((a, f) => f(a), a);
}

interface Unsubscribe {
  (): any;
}

interface Subscription {
  unsubscribe: Unsubscribe;
}

interface StreamVisitor<T> {
  next: (value: T) => any;
}

interface Stream<T> {
  (visitor: StreamVisitor<T>): Subscription;
}

interface Consumer<T> {
  next: (value: T) => void;
}

interface Producer<T> {
  (consumer: Consumer<T>): Unsubscribe | void;
}

const create = <T>(producer: Producer<T>): Stream<T> => (visitor) => {
  let next = (value: T) => {
    visitor.next(value);
  };
  let consumer: Consumer<T> = {
    next: next,
  };

  let unsubscribe = producer(consumer);

  return {
    unsubscribe() {
      if (unsubscribe) unsubscribe();
    },
  };
};

interface MapFn<T, TT> {
  (value: T): TT;
}

const map = <T, TT>(f: MapFn<T, TT>) => (stream: Stream<T>): Stream<TT> => {
  return (visitor) => {
    return stream({
      ...visitor,
      next: (value) => {
        visitor.next(f(value));
      },
    });
  };
};

interface FilterFn<T> {
  (value: T): boolean;
}

const filter = <T>(f: FilterFn<T>) => (stream: Stream<T>): Stream<T> => {
  return (visitor) => {
    return stream({
      ...visitor,
      next: (value) => {
        if (f(value)) {
          visitor.next(value);
        }
      },
    });
  };
};

const take = (n: number = 0) => <T>(stream: Stream<T>): Stream<T> => {
  return (visitor) => {
    let subscription = stream({
      ...visitor,
      next: (value) => {
        if (n-- <= 0) {
          subscription.unsubscribe();
        } else {
          visitor.next(value);
        }
      },
    });
    return subscription;
  };
};

const foreach = <T>(f: (value: T) => any) => (stream: Stream<T>) => {
  return stream({
    next: f,
  });
};

const interval = (period: number = 1000) =>
  create<number>((consumer) => {
    let i = 0;
    let tid = setInterval(() => {
      consumer.next(i++);
    }, period);

    return () => {
      clearInterval(tid);
    };
  });

pipe(
  interval(100),
  filter((n) => n % 2 === 0),
  map((n) => `count: ${n}`),
  take(10),
  foreach(console.log)
);

pullable & pushable streaming

interface UnaryFunction<T, TT> {
  (arg: T): TT;
}

function pipe<T, A>(a: T, f1: UnaryFunction<T, A>): A;
function pipe<T, A, B>(
  a: T,
  f1: UnaryFunction<T, A>,
  f2: UnaryFunction<A, B>
): B;
function pipe<T, A, B, C>(
  a: T,
  f1: UnaryFunction<T, A>,
  f2: UnaryFunction<A, B>,
  f3: UnaryFunction<B, C>
): C;
function pipe<T, A, B, C, D>(
  a: T,
  f1: UnaryFunction<T, A>,
  f2: UnaryFunction<A, B>,
  f3: UnaryFunction<B, C>,
  f4: UnaryFunction<C, D>
): D;
function pipe<T, A, B, C, D, E>(
  a: T,
  f1: UnaryFunction<T, A>,
  f2: UnaryFunction<A, B>,
  f3: UnaryFunction<B, C>,
  f4: UnaryFunction<C, D>,
  f5: UnaryFunction<D, E>
): E;

function pipe(a: any, ...args: UnaryFunction<any, any>[]) {
  return args.reduce((a, f) => f(a), a);
}

interface Subscriber<A> {
  start: () => void;
  next: (value: A) => any;
  complete: () => void;
}

interface ProducerHandler {
  start: () => void;
  next: () => void;
  complete: () => void;
}

interface Producer<A> {
  (subscriber: Subscriber<A>): ProducerHandler;
}

const range = (from: number, to: number): Producer<number> => (subscriber) => {
  let start = () => {
    subscriber.start();
  };
  let next = () => {
    if (from <= to) {
      subscriber.next(from++);
    } else {
      complete();
    }
  };
  let complete = () => {
    subscriber.complete();
  };
  return { start, next, complete };
};

const interval = (period: number): Producer<number> => (subscriber) => {
  let tid: any;
  let i = 0;
  let start = () => {
    subscriber.start();
    tid = setInterval(() => subscriber.next(i++), period);
  };
  let next = () => {};
  let complete = () => {
    clearInterval(tid);
    subscriber.complete();
  };

  return { start, next, complete };
};

const delay = (duration: number) => <T>(producer: Producer<T>): Producer<T> => {
  return (subscriber) => {
    let timer = interval(duration)({
      start: () => {},
      next: () => {
        parent.next();
      },
      complete: () => {},
    });

    let parent = producer({
      ...subscriber,
    });

    let start = () => {
      timer.start();
      parent.start();
    };

    let next = () => {};

    let complete = () => {
      timer.complete();
      parent.complete();
    };

    return { start, next, complete };
  };
};

const toPullable = <T>(producer: Producer<T>): Producer<T> => {
  return (subsciber) => {
    let count = 0;
    let valueList: T[] = [];

    let handler = producer({
      ...subsciber,
      next: (value) => {
        if (count) {
          count -= 1;
          subsciber.next(value);
        } else {
          valueList.push(value);
        }
      },
    });

    let next = () => {
      if (valueList.length) {
        let value = valueList.shift();
        subsciber.next(value);
      } else {
        count += 1;
      }
    };

    return { ...handler, next };
  };
};

const map = <A, B>(f: UnaryFunction<A, B>) => (
  producer: Producer<A>
): Producer<B> => {
  return (subscriber) =>
    producer({
      ...subscriber,
      next: (value) => subscriber.next(f(value)),
    });
};

const filter = <A>(f: (value: A) => boolean) => (
  producer: Producer<A>
): Producer<A> => {
  return (subscriber) => {
    let handler = producer({
      ...subscriber,
      next: (value) => {
        if (f(value)) {
          subscriber.next(value);
        } else {
          handler.next();
        }
      },
    });

    return handler;
  };
};

const take = (count: number) => <A>(producer: Producer<A>): Producer<A> => {
  return (subscriber) => {
    let handler = producer({
      ...subscriber,
      next: (value) => {
        if (count-- <= 0) {
          handler.complete();
        } else {
          subscriber.next(value);
        }
      },
    });
    return handler;
  };
};

const foreach = <A>(f: (value: A) => any) => (
  producer: Producer<A>
): ProducerHandler => {
  let handler = producer({
    start: () => {
      handler.next();
    },
    next: (value) => {
      f(value);
      handler.next();
    },
    complete: () => {},
  });
  return handler;
};

let handler0 = pipe(
  range(10, Number.POSITIVE_INFINITY),
  filter((n) => n % 2 === 0),
  map((n) => `range count: ${n}`),
  take(10),
  foreach(console.log)
);

handler0.start();

let handler1 = pipe(
  interval(10),
  filter((n) => n % 2 === 0),
  map((n) => `interval count: ${n}`),
  take(10),
  foreach(console.log)
);

handler1.start();

let handler2 = pipe(
  range(10, Number.POSITIVE_INFINITY),
  delay(1000),
  map((n) => `delay count: ${n}`),
  foreach(console.log)
);

handler2.start();

let handler3 = pipe(interval(10), toPullable, (source) =>
  source({
    start: () => {},
    next: (n) => {
      console.log(`pullable count: ${n}`);
    },
    complete: () => {},
  })
);

handler3.start();

setTimeout(() => {
  handler3.next();
  handler3.next();
  handler3.next();
}, 1000);