Lazy, functional streams library for Node.js

Question

I’ve written a small library that implements lazy streams in (ES2016) JavaScript for use in Node.js. The library could work in the browser as well with some minor modifications, but that isn’t something I’m concerned with at the moment.

Here is the library code, in its entirety:

import { curry } from 'ramda'
const util = require('util')

const toString = (x) =>
  (x && typeof x.toString === 'function') ? x.toString() : Object.prototype.toString.call(x)
const hasOwnProperty = (x, prop) =>
  x != null && Object.prototype.hasOwnProperty.call(x, prop)

const consTag = Symbol('Stream.cons')
const emptyTag = Symbol('Stream.empty')
const delayTag = Symbol('Stream.delay')

export const isCons = (x) => hasOwnProperty(force(x), consTag)
export const isEmpty = (x) => hasOwnProperty(force(x), emptyTag)
export const isDelay = (x) => hasOwnProperty(x, delayTag)
export const is = (x) => isCons(x) || isEmpty(x)

const assertIs = (fnName, x) => {
  if (!is(x)) {
    throw new TypeError(`Stream.${fnName}: expected a Stream, but was given ${toString(x)}`)
  }
  return x
}

const assertFunction = (fnName, x) => {
  if (typeof x !== 'function') {
    throw new TypeError(`Stream.${fnName}: expected a Function, but was given ${toString(x)}`)
  }
  return x
}

// The empty stream.
export const empty = Object.freeze(Object.create(null, {
  [emptyTag]: { value: true },
  toString: { value: () => 'Stream.empty' },
  [util.inspect.custom]: { value: (depth, opts) => {
    const special = (x) => opts.stylize(x, 'special')
    if (depth < 0) {
      return special('[Stream]')
    }
    return `${special('Stream')}.${special('empty')}`
  } }
}))

// Strictly produces a new stream by prepending an element onto an existing
// stream. It is often useful to be lazy in the second argument, in which case
// you should wrap it with Stream.delay.
export const cons = curry((x, xs) => Object.freeze(Object.create(null, {
  [consTag]: { value: delay(() => [x, assertIs('cons', xs)]) },
  toString: { value: () => `Stream.cons(${toString(x)}, ${toString(assertIs('cons', xs))})` },
  [util.inspect.custom]: { value: (depth, opts) => {
    const special = (x) => opts.stylize(x, 'special')
    if (depth < 0) {
      return special('[Stream]')
    }
    const newOpts = Object.assign({}, opts, {
      depth: opts.depth == null ? null : opts.depth - 1 })
    const y = util.inspect(x, newOpts)
    const ys = util.inspect(assertIs('cons', xs), newOpts)
    return `${special('Stream')}.${special('cons')}(${y}, ${ys})`
  } }
})))

// Creates a delayed value from a nullary function. When the value is first
// evaluated using Stream.force, the function will be invoked to produce a
// value. The resulting value will be cached, and subsequent invocations of
// Stream.force will produce the same value.
//
// Delayed values are forced recursively, so if a delayed value evaluates to
// another delayed value, Stream.force will force the result until a non-delayed
// value is produced.
export const delay = (thunk) => {
  assertFunction('delay', thunk)
  let forced = false
  let val = undefined
  return Object.freeze(Object.create(null, {
    [delayTag]: { value: () => {
      if (forced) {
        return val
      } else {
        val = thunk()
        forced = true
        return val
      }
    } },
    toString: { value: () =>
      forced ? `Stream.delay(${toString(val)})` : 'Stream.delay(...)'
    },
    [util.inspect.custom]: { value: (depth, opts) => {
      const special = (x) => opts.stylize(x, 'special')
      if (depth < 0) {
        return special('[Stream.delay]')
      }
      if (forced) {
        const newOpts = Object.assign({}, opts, {
          depth: opts.depth == null ? null : opts.depth - 1 })
        const str = util.inspect(val, newOpts)
        return `${special('Stream')}.${special('delay')}(${str})`
      }
      return `${special('Stream')}.${special('delay')}(${special('...')})`
    } }
  }))
}

// Forces evaluation of a value. If the provided argument is a value produced
// by Stream.delay, returns the value as described in the documentation for
// Stream.delay. Otherwise, returns the argument unchanged.
export const force = (x) =>
  isDelay(x) ? force(x[delayTag]()) : x

// The basic stream consumer. Must be called on a nonempty stream. The result
// is a an array of length 2 containing the first element of the stream and the
// remainder of the stream.
export const uncons = (xs) => {
  if (!isCons(assertIs('uncons', xs))) {
    throw new TypeError(`Stream.uncons: expected nonempty Stream, but was given ${toString(xs)}`)
  }
  return force(force(xs)[consTag])
}

// Returns the first element of a stream. The stream must be nonempty.
export const first = (xs) => {
  const [y, _ys] = uncons(assertIs('first', xs))
  return y
}

// Returns the remainder of a stream without its first element. The stream must
// be nonempty.
export const rest = (xs) => {
  const [_y, ys] = uncons(assertIs('rest', xs))
  return ys
}

// Lazily applies a function to each element of a stream to produce a new
// stream.
export const map = curry((f, xs) =>
  delay(() => {
    const [y, ys] = uncons(assertIs('map', xs))
    return cons(assertFunction('map', f)(y), map(f, ys))
  }))

// Strictly applies a function to each element of a stream, and always returns
// undefined.
export const forEach_ = curry((xs, f) => {
  let ys = xs
  while (!isEmpty(ys)) {
    const [z, zs] = uncons(assertIs('forEach_', ys))
    assertFunction('forEach_', f)(z)
    ys = zs
  }
})

export const forEach = curry((f, xs) => forEach_(assertIs('forEach', xs), assertFunction('forEach', f)))
export const map_ = curry((xs, f) => map(assertFunction('map_', f), assertIs('map_', xs)))

// The strict left fold over streams.
export const foldl = curry((f, acc, xs) => {
  assertFunction('foldl', f)
  if (isEmpty(assertIs('foldl', xs))) {
    return acc
  } else {
    const [y, ys] = uncons(xs)
    return foldl(f, f(acc, y), ys)
  }
})

// The lazy right fold over streams. The accumulator function will be provided
// a delayed value as its second argument, so if the accumulator function is
// lazy in its second argument, the overall fold will also be lazy.
//
// If the accumulator function is strict, consider using Stream.foldl instead.
export const foldr = curry((f, acc, xs) => {
  assertFunction('foldr', f)
  if (isEmpty(assertIs('foldr', xs))) {
    return acc
  } else {
    const [y, ys] = uncons(xs)
    return f(y, delay(() => foldr(f, acc, ys)))
  }
})

// Converts a stream to an ordinary JavaScript array.
export const toArray = (xs) => {
  const arr = []
  forEach_(assertIs('toArray', xs), (x) => {
    arr.push(x)
  })
  return arr
}

// Lazily converts a JavaScript array to a stream. Since the stream is produced
// lazily, the source array should not be mutated, or the behavior of the result
// could be unpredictable. Additionally, a reference to the source array will be
// retained until the stream is fully realized, so it will not be garbage
// collected until the stream is fully constructed.
export const fromArray = (arr) => {
  return (function next(i) {
    return i >= arr.length
      ? empty : delay(() => cons(arr[i], next(i + 1)))
  })(0)
}

// Produces an infinite stream of natural numbers, starting from the provided
// argument. If no argument is provided, the stream starts from 0.
export const naturals = (n = 0) =>
  delay(() => cons(n, naturals(n + 1)))

// Lazily produces a new stream from an existing one that contains the first n
// elements from the original stream. If the original stream has fewer than n
// elements, the result is the original stream.
export const take = curry((n, xs) => {
  if (typeof n !== 'number') {
    throw new TypeError(`Stream.take: number of elements must be a number, but was given ${toString(n)}`)
  }
  if (n < 0) {
    throw new Error(`Stream.take: number of elements must be nonnegative, but was given ${toString(n)}`)
  }
  if (n == 0 || isEmpty(xs)) {
    return empty
  } else {
    const [y, ys] = uncons(assertIs('take', xs))
    return delay(() => cons(y, take(n - 1, ys)))
  }
})

// Produces a new stream from an existing one without the first n elements of
// the original stream. If the original stream has fewer than n elements, the
// result is the empty stream.
export const drop = curry((n, xs) => {
  if (typeof n !== 'number') {
    throw new TypeError(`Stream.drop: number of elements must be a number, but was given ${toString(n)}`)
  }
  if (n < 0) {
    throw new Error(`Stream.drop: number of elements must be nonnegative, but was given ${toString(n)}`)
  }
  assertIs('drop', xs)
  if (n == 0 || isEmpty(xs)) {
    return xs
  } else {
    return drop(n - 1, rest(xs))
  }
})

A large portion of this code is dedicated to improving usability of the library rather than implementing core functionality. For example, the library is careful to perform the following:

• It overrides both toString and [util.inspect.custom] to provide custom string representations in the Node.js REPL. The two functions produce the same output in the simplest case, but the inspect implementation adds colorization and respects util.inspect’s depth limit.

  This representation looks like this in the console:

  > Stream.force(Stream.naturals())
  Stream.cons(0, Stream.delay(Stream.cons(1, [Stream.delay])))
  

• Functions are careful to check the types of their arguments. Passing something that isn’t a stream to a function that expects a stream produces an error message in terms of the called function:

  > Stream.force(Stream.map((x) => x + 1, [1, 2, 3, 4]))
  TypeError: Stream.map: expected a Stream, but was given 1,2,3,4
  

• Functions are optionally curried to make it easy to partially apply them, and most arguments are ordered to take streams last to make currying most useful:

  > const firstFive = Stream.take(5)
  > Stream.toArray(firstFive(Stream.naturals()))
  [ 0, 1, 2, 3, 4 ]
  

• There’s an effort to provide best-effort encapsulation. Streams are effectively opaque data structures that can be built using cons and delay and consumed using uncons. They don’t expose any methods or properties that hint at the internal structure:

  > Object.keys(Stream.force(Stream.fromArray([1, 2, 3])))
  []
  

Here is a very simple sample program that implements the classic FizzBuzz problem as an infinite stream:

import * as Stream from './streams'

const fizzbuzz = (n) =>
    n % 15 == 0 ? "FizzBuzz"
  : n %  5 == 0 ? "Buzz"
  : n %  3 == 0 ? "Fizz"
                : n.toString()

const fizzbuzzStream = Stream.map(fizzbuzz, Stream.naturals(1))

Stream.forEach((x) => console.log(x), fizzbuzzStream)

The sample program is only for illustrative purposes, and it isn’t up for review, only the library code is.

My primary focus for this library is creating the most pleasant interface to use and work with, so usability is my first priority. I’m also looking for ways to make the implementation easier to read, since right now parts of it feel sort of cluttered. Finally, if there are easy ways to make it more performant, I’d be interested in that, too, but not at the expense of usability or implementation clarity.

Answer 1

First, I applaud your effort - I must confess: this is definitely lacking in the js world.

A few things come to mind:

1. You're using ramda for currying and nothing else? Ditch it and implement your own currying; it's not hard.

2. Since usability is of a concern to you then making this an actual nodejs stream would be hot and also not difficult.

3. Integrating it with iterators/generators would be handy

4. A more natural usage would be Stream.naturals(1).map(fizzbuzz).filter(...).until(terminalCondition)

If you can actually integrate it with node streams then you can do nijitsu like:

const fs = require('fs');

const readStream = fs.createReadStream('myfile.txt');
Stream.from(readStream).map(...)

others:

Stream.fromFile(...)

Stream.fromSocket(...)

IO is always infinite (or should assumed to be so) and thus is natural for this case. The default node libraries for streams are really unimaginative on this. I was just asking around on JS chat yesterday about something like this.