Inspired by Cycle.js and Motorcycle.js

0-description.md

I was playing around with functional and reactive programming in Javascript, and the UI framework Cycle.js. In order to better understand how that framework works, I pared down the core functionality to a single concept: given a function that takes in an input stream and returns an output stream, resolve the circular dependency of feeding its output back into itself as input. I also built a naive re-implementation of the core Cycle.run command on top of that.

1-ouroboros.js

/*****************************************************************************
  This is the function doing all the heavy lifting.
 
  It takes a function and returns a stream generated by feeding its output
  back into itself as input. (It doesn't start consuming the stream.)
 
  It's basically a pared-down version of Cycle.run that forgets about the
  application architecture of separating out a pure main from the effectful
  drivers and just resolves a single circularly dependent stream. The other
  files build up more API-compatible versions of Cycle.run from this.
 *****************************************************************************/

import most from 'most'
import hold from '@most/hold'

const ouroboros = f => {
  const input = hold(most.create((add, end, error) => {
    output.observe(add).then(end).catch(error)
  }))

  const output = f(input)

  return input
}

export default ouroboros

2-utils.js

/*****************************************************************************
  Some utilities for manipulating most.js streams.
 
  pluck is pluck.
 
  mergeObjOfStreams and sift are inverses of each other.
 
  mergeObjOfStreams takes an object of streams, e.g.
  {
    a:  -------1---------------------------------|,
    b:  ----------------------------------2----------------------|
  }
  and merges them together as the single stream of objects
        -------{key: `a`, value: 1}-------{key: `b`, value: 2}---|.
 
  sift does the (quasi)-inverse: given a stream of {key, value} objects, and a
  list of keys, it separate the stream back out to an object of streams that is
  indexed by the keys.
 *****************************************************************************/
 
import compose from '@your-favorite-utility-library/compose'
import mapObj from '@your-favorite-utility-library/mapObj'
import zipObj from '@your-favorite-utility-library/zipObj'

import most from 'most'

export function pluck(key, stream) {
  return (stream || this).map(obj => obj[key])
}

export const mergeObjOfStreams = compose(
  array => most.merge(...array),
  Object.values,
  mapObj.bind(null, (stream, key) => stream.map(value => {key, value})))

export const sift = (keys, stream) => 
  zipObj(keys, keys.map(k => stream
    .filter({key} => key === k)
    ::pluck(`value`)))
  
export {compose, mapObj, mergeObjOfStreams, pluck, sift, zipObj}

3-ouroboros-obj.js

import {compose, mergeObjOfStreams, sift} from './utils'

const ouroborosObj = (f, ...keys) => otherStreams => sift(
  Object.keys(otherStreams).concat(keys),
  ouroboros(
    compose(
      mergeObjOfStreams,
      streams => f({...streams, ...(otherStreams || {})}),
      sift.bind(null, keys)
  )))

export default ouroborosObj

4-run.js

/*****************************************************************************
  A naive re-implementation of Cycle.run.
 
  It doesn't return anything, just runs the program. See below for an
  implementation that actually returns the {sources, sinks} object.
 *****************************************************************************/

import ouroborosObj from './ouroboros-obj'
import {compose, mapObj} from './utils'

export const run = (main, drivers) => ouroborosObj(
  compose(main, mapObj.bind(null, (proxy, key) => drivers[key](proxy))),
  ...Object.keys(drivers))().drain()

5-run-with-return-value.js

/*****************************************************************************
  Another naive re-implementation of Cycle.run.
 
  This one actually returns the {sources, sinks} object. 
 *****************************************************************************/
import hold from '@most/hold'

import {compose, mapObj, mergeObjOfStreams, sift} from './utils'
import ouroboros from './ouroboros'

const run = (main, drivers) => {
  const keys = Object.keys(drivers)
  const {sources: sourcesStream, sinks: sinksStream} = ouroborosObj(
    compose(
      mapObj.bind(null, mergeObjOfStreams),
      sources => {sources, sinks: main(sources)},
      mapObj.bind(null, (proxy, key) => drivers[key](proxy)),
      mapObj.bind(null, sift.bind(null, keys))))()
  
  const {sources, sinks} = mapObj(
    sift.bind(null, keys),
    {sources: sourcesStream, sinks: sinksStream})
    
  Object.values(sinks).forEach(sink => sink.drain())
  return {sources, sinks}
}