Implementing Streams in Javascript

What is a Stream?

According to Wikipedia, “In computer science, a stream is a sequence of data elements made available over time. A stream can be thought of as items on a conveyor belt being processed one at a time rather than in large batches.”

In this blog, I will be implementing our very own Stream in javascript.

Before understanding Stream, let’s take a detour and understand what lists are? Specifically, Linked list.

Lists

Lists are finite structures which have a “head” (the first value), and a “tail” (rest of the values as another list). It’s a very primitive and well known recursive datastructure.

In javascript you can describe a list as follows-

const cons = (hd, tl) => f => f(hd, tl);
const car = p => p((hd, tl) => hd);
const cdr = p => p((hd, tl) => tl);

Wait… Were you expecting something else? Node with pointers - next, prev, value, etc. No!, we don’t need those. In javascript (or any other programming language with closures and lambdas), the above description is enough for defining lists.

cons(hd, tl) is a function/ closure that is waiting for another function f to apply arbitrary logic on hd and tl. It’s wierd for some people to conceive lists as function. But, It’s a well known fact that - “data is function” & “function is data”. You can understand this by a very simple example - You can always replace a (pure) function with a table of input and it’s corresponding output. Ofcourse this will be very inefficient, but it clearly shows this fact.

const add = (x, y) => x + y;
// can be replaced with
const add = {
  [1, 2]: 3,
  [1, 1]: 2,
  [10, 100]: 110,
  // and so on ...
}

cons help us create pairs of data. e.g-

const name = cons("Ujjawal Sinha", 1998);

Then, we can use car and cdr to access the data elements. car access the first element, and cdr access the second.

console.log(car(name)); // Ujjawal Sinha
console.log(cdr(name)); // 1998

Once you have an idea of storing pair, you can compose these pairs to form arbitrary data structures. For lists, you represent it by repeated post composition. e.g- to represent list [1, 2, 3, 4], you can write-

const myItems = cons(1, cons(2, cons(3, cons(4, undefined))));

undefined is used as a special value to represent: end of list

And then you can define other useful HOFs like map, filter, foldl etc. for manipulating lists.

const map = (items, mapper) =>
  items === undefined
    ? undefined
    : cons(mapper(car(items)), map(cdr(items), mapper));

const foldl = (folder, items, x) =>
  items === undefined
    ? x
    : foldl(folder, cdr(items), folder(x, car(items)));

const filter = (items, pred) =>
  items === undefined
    ? undefined
    : pred(car(items))
    ? cons(car(items), filter(cdr(items), pred))
    : filter(cdr(items), pred);

Streams

We can use lists to represent streams, but the problem with lists is that it’s eager. We want streams to be lazy - we want to process elements one at a time. You can implement this logic using cons, where the second element is a suspended/ delegated operation that upon request produces the next element in the stream.

const cons = (hd, suspendedTail) => f => f(hd, suspendedTail);
const car = p => p((hd, suspendedTail) => hd);
const cdr = p => p((hd, suspendedTail) => suspendedTail());

We suspend the second element of the pair by wrapping it in a thunk/ lambda with no arguments. car inspects the first element in the stream, and cdr executes the delegated procedure to produce next element in the stream.

const myItems = cons(1, () => cons(2, () => cons(3, () => undefined)));
// notice it's similarity to previous definition of myItems

console.log(car(myItems)); // 1
console.log(car(cdr(myItems))); // 2

We can also define useful HOFs for manipulating streams-

const map = (items, mapper) =>
  items === undefined
    ? undefined
    : cons(mapper(car(items)), () => map(cdr(items), mapper));

const foldl = (folder, items, x) =>
  items === undefined
    ? x
    : foldl(folder, cdr(items), folder(x, car(items)));

const foldr = (folder, items, x) =>
  items === undefined
    ? x
    : folder(foldr(folder, cdr(items), x), car(items));

const reverse = (items) =>
  foldl((lst, item) => cons(item, () => lst), items, undefined);

const filter = (items, pred) =>
  items === undefined
    ? undefined
    : pred(car(items))
    ? cons(car(items), () => filter(cdr(items), pred))
    : filter(cdr(items), pred);

// and other useful HOFs

const take = (n, items) =>
  (n <= 0 || items === undefined)
    ? undefined
    : cons(car(items), () => take(n - 1, cdr(items)));

const drop = (n, items) =>
  (n <= 0 || items === undefined)
    ? items
    : drop(n - 1, cdr(items));

const zip2 = (s1, s2, zipf) =>
  (s1 === undefined || s2 === undefined)
    ? undefined
    : cons(zipf(car(s1), car(s2)), () => zip2(cdr(s1), cdr(s2), zipf))

Converting back and forth from javascript’s arrays to streams is a good idea-

const fromArray = (items) =>
  reverse(items.reduce((lst, item) => cons(item, () => lst), undefined));

const myItems = fromArray([1, 2, 3, 4]);
console.log(foldl((x, y) => x + y, myItems, 0)); // 1 + 2 + 3 + 4 = 10
console.log(foldl((x, y) => x * y, myItems, 1)); // 1 * 2 * 3 * 4 = 24

const toArray = (items) =>
  foldl(
    (arr, item) => {
      arr.push(item);
      return arr;
    },
    items,
    []
  );

console.log(toArray(myItems)); // [1, 2, 3, 4]
console.log(toArray(take(3, myItems))); // [1, 2, 3]
console.log(toArray(drop(1, myItems))); // [2, 3, 4]
console.log(toArray(filter(myItems, (n) => n % 2 === 0))); // [2, 4]
console.log(toArray(zip2(myItems, drop(1, myItems), (x, y) => [x, y]))); // [[1, 2], [2, 3], [3, 4]]

One thing to note in the above implementation of stream is that cdr executes the suspended procedure every single time it’s called. It might be a good idea to cache/ memoize the result of first execution, so that the next time it’s called we dont have to execute the procedure again-

So cdr changes a little.

const cdr = (p) =>
  p(
    (() => {
      let cache = undefined;
      return (hd, suspendedTail) => {
        if (cache === undefined) {
          cache = suspendedTail();
        }
        return cache;
      };
    })()
  );

Now we are pretty much done, we can now define streams in javascript using the above constructs-

Representing stream of natural numbers

const countFrom = (i) => cons(i, () => countFrom(i + 1));
const nats = countFrom(0);
// nats is a cons whose first element is 0, and second element is a suspended procedure to generate rest of nats (1, 2, 3, ...)
// we can call car or cdr on nats to inspect it's elements
console.log(car(nats)); // 0
console.log(car(cdr(nats))); // 1
console.log(car(cdr(cdr(nats)))); // 2
// ...

// we can also use our HOFs to access first 5 elements
console.log(toArray(take(5, nats))); // [0, 1, 2, 3, 4]

Writing car & cdr is rather unwieldy, let’s define a function at that takes an index and return value stored at that index-

const at = (i, items) =>
  (i < 0 || items === undefined)
    ? undefined
    : i === 0
    ? car(items)
    : at(i - 1, cdr(items));

console.log(at(0, nats)); // 0
console.log(at(1, nats)); // 1
console.log(at(2, nats)); // 2
console.log(at(3, nats)); // 3
console.log(at(4, nats)); // 4
// ...

Representing stream of Fibonacci numbers

const fibs = (f0, f1) =>
  cons(f0, () => fibs(f1, f0 + f1));

console.log(toArray(take(5, fibs(1, 1)))); // [1, 1, 2, 3, 5]

Representing Collatz sequence

const collatz = (n) =>
  cons(n, () => collatz(n % 2 === 0 ? n >> 1 : 3 * n + 1));

console.log(toArray(take(10, collatz(10)))); // [10, 5, 16, 8, 4, 2, 1, 4, 2, 1]

Reading file data in chunks

Asuming I have a way of creating a file descriptor and read some bytes from it, I can create a stream of those bytes using cons

const readFileInChunks = (maxBytes, filePath) => {
  const readChunkFromFileDescriptor = (fp) => {
    const data = fp.readBytes(maxBytes)
    if (data === undefined) {
      fp.close();
      return undefined;
    }
    return cons(data, () => readChunkFromFileDescriptor(fp));
  }
  return readChunkFromFileDescriptor(new File(filePath));
}

Conclusion

Streams are a very powerful way of representing finite/ infinite lists. And defining in javascript is trivial. Hope you found some insights on how streams are implemented in javascript.

There is one problem with the above implementation though. Functions like, at, foldl, foldr, filter, drop are recursive, so it is bounded by the max stack size of the environment and hence are susceptible to Stack-Overflow errors.

Some programming languages like Scala, Haskell do tail call optimization to avoid Stack Overflow errors in recursive programs. Unfortunately javascript does not have this built-in.

In order to fix this, we use a pattern called: Trampoline. Planning to write on it next.

And with that I will wrap this up. Thanks for reading. Keep Learning :)