stream.js

// ==ClosureCompiler==
// @output_file_name default.js
// @compilation_level SIMPLE_OPTIMIZATIONS
// ==/ClosureCompiler==

/* eslint-disable */
;(function() {
"use strict"
/* eslint-enable */
Stream.HALT = {}
Stream.scan = scan
Stream.merge = merge
Stream.combine = combine
Stream.scanMerge = scanMerge

function Stream(value) {
	var dependentStreams = []
	var dependentFns = []

	function stream(v) {
		if (arguments.length && v !== Stream.HALT && open(stream)) {
			value = v
			stream.state = "active"
			dependentStreams.forEach(function (s) { return s.state = open(s) ? "changing" : s.state })
			dependentStreams.forEach(function (s, i) { return open(s) && s(dependentFns[i](value)) })
		}
		return value
	}

	stream.state = arguments.length && value !== Stream.HALT ? "active" : "pending"
	stream.map = function (fn, ignoreInitial) {
		var target = stream.state === "active" && ignoreInitial !== Stream.HALT
			? Stream(fn(value))
			: Stream()
		dependentStreams.push(target)
		dependentFns.push(fn)
		return target
	}

	var end
	function createEnd() {
		end = Stream()
		end.map(function(v) {
			if (v === true) {
				stream.state = "ended"
				dependentStreams.length = dependentFns.length = 0
			}
			return v
		})
		return end
	}

	stream.toJSON = function () { return value != null && typeof value.toJSON === "function" ? value.toJSON() : value }
	stream.toString = function () { return value }
	stream.valueOf = function () { return value }
	stream["fantasy-land/map"] = stream.map
	stream["fantasy-land/ap"] = function (x) { return combine(function (s1, s2) { return s1()(s2()) }, [x, stream]) }
	stream["fantasy-land/of"] = Stream
	Object.defineProperty(stream, "end", {get: function () { return end || createEnd() }})

	return stream
}

function combine(fn, streams) {
	var ready = streams.every(function (s) {
		if (s["fantasy-land/of"] !== Stream)
			throw new Error("Ensure that each item passed to stream.combine/stream.merge is a stream")
		return s.state === "active"
	})
	var stream = ready
		? Stream(fn.apply(null, streams.concat([streams])))
		: Stream()
	var changed = []
	streams.forEach(function (s) {
		return s.map(function() {
			changed.push(s)
			if (streams.every(function (s) { return s.state !== "changing" })) {
				stream(fn.apply(null, streams.concat([changed.slice(0)])))
				changed = []
			}
		}, Stream.HALT)
	})
	return stream
}

function merge(streams) {
	var stream = streams.every(function (s) { return s.state === "active" })
		? Stream(streams.map(function (s) { return s() }))
		: Stream()

	streams.forEach(function (s) {
		return s.map(function () { return stream(streams.map(function (s) { return s() })) }, Stream.HALT)
	})

	return stream
}

function scan(fn, acc, origin) {
	var stream = origin.map(function (v) {
		acc = fn(acc, v)
		return acc
	})

	stream(acc)

	return stream
}

function scanMerge(tuples, seed) {
	var streams = tuples.map(function (a) {
		return a[0]
	})

	var stream = combine(function () {
		var changed = arguments[arguments.length - 1]
		streams.forEach(function (stream, i) {
			if (changed.indexOf(stream) > -1)
				seed = tuples[i][1](seed, stream())
		})
		return seed
	}, streams)

	stream(seed)

	return stream
}

function open(s) {
	return s.state === "pending" || s.state === "active" || s.state === "changing"
}

if (typeof module !== "undefined") module["exports"] = Stream
else window.stream = Stream
}());