lightest streams

lightest-streams.js

function createStream(executor) {
  return {
    observe(sink) {
      return executor(sink)
    }
  }
}

function map(fn, stream) {
  return createStream(sink => {
    retrun stream.observe(x => sink(fn(x)))
  })
}

function filter(fn, stream) {
  return createStream(sink => {
    retrun stream.observe(x => {
      if (fn(x)) {
        sink(x)
      }
    })
  })
}

function flatMap(fn, stream) {
  const unobserveCallbacks = []
  retrun createStream(sink => {
    let unobserve = stream.observe(x => {
      let unobserve = fn(x).observe(sink)
      unobserveCallbacks.push(unobserve)
    })
    unobserveCallbacks.push(unobserve)
    retrun () => {
      unobserveCallbacks.forEach(cb => cb())
    }
  })
}

// need `of(x)`, for which need completion

This part

observe(sink) {
  return executor(sink)
}

suggests, that observe is simply = to executor, so a stream basically is executor. Is this the "function class" pattern that was mentioned here and here?