Inspired by Cycle.js and Motorcycle.js

0-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 just 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) => {
    setTimeout(() => output.observe(add).then(end).catch(error), 1)
  }))

  const output = f(input)

  return output
}

export default ouroboros

1-utils.js

/*****************************************************************************
 Some utilities for manipulating most.js streams.
 
 pluck is pluck.
 
 mergeObjOfStreams and siftStreamOfObjs 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}---|.
 
 siftStreamOfObjs separates a stream of {key, value} objects back out into 
 an object of streams of the values, indexed by the keys.
 *****************************************************************************/
 
import compose from '@your-favorite-utility-library/compose'
import curry from '@your-favorite-utility-library/curry'
import mapObj from '@your-favorite-utility-library/mapObj'
import values from '@your-favorite-utility-libarary/values'
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),
  values,
  curry(mapObj)((stream, key) => stream.map(value => {key, value})))

export const siftStreamOfObjs = (keys, stream) => zipObj(keys, keys.map(k => stream
  .filter({key} => key === k)
  ::pluck(`value`)))
  
export {compose, curry, mapObj, mergeObj, pluck, siftObj, values, zipObj}

2-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 ouroboros from './ouroboros'
import {compose, curry, mapObj, mergeObjOfStreams, siftStreamOfObjs} from './utils'

export const run = (main, drivers) => {
  const f = compose(
    mergeObjOfStreams,
    main,
    curry(mapObj)((sinkProxy, key) => drivers[key](sinkProxy)),
    curry(siftStreamOfObjs)(Object.keys(drivers)))

  const stream = ouroboros(f)
  stream.drain()
}

3-run-with-return-value.js

/*****************************************************************************
 Another naive re-implementation of Cycle.run.
 
 This one actually returns the {sources, sinks} object. 
 *****************************************************************************/
import ouroboros from './ouroboros'
import {compose, curry, mapObj, mergeObjOfStreams, pluck, siftStreamOfObjs} from './utils'
 
export const run = (main, drivers) => {
  const keys = Object.keys(drivers)

  const f = stream => {
    const sinkProxies = siftStreamOfObjs(
      keys,
      stream.filter({key} => key === `sinks`)::pluck(`value`))
    const sources = mapObj((proxy, key) => drivers[key](proxy), sinkProxies)
    const sinks = main(sources)

    return mergeObjOfStreams(mapObj(mergeObjOfStreams, {sources, sinks}))
  }

  const stream = ouroboros(f)
  stream.drain()

  return mapObj(
    curry(siftStreamOfObjs)(keys),
    siftStreamOfObjs([`sources`, `sinks`], stream))
}