vndb/mithril-vndb
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fantasy-land methods no longer absorb streams

Browse source
This commit is contained in:
Leo Horie 2016-06-30 01:28:50 -04:00
parent 7fa3fee9ce
commit 015a812610
4 changed files with 524 additions and 290 deletions
Download patch file Download diff file Expand all files Collapse all files

62
util/stream.js
View file

@ -16,8 +16,8 @@ function initStream(stream, args) {
stream.constructor = createStream
stream._state = {id: guid++, value: undefined, error: undefined, state: 0, derive: undefined, recover: undefined, deps: {}, parents: [], errorStream: undefined, endStream: undefined}
stream.map = map, stream.ap = ap, stream.of = createStream
stream.valueOf = valueOf
stream.catch = doCatch
stream.valueOf = valueOf, stream.toJSON = toJSON
stream.run = run, stream.catch = doCatch
Object.defineProperties(stream, {
error: {get: function() {
@ -46,15 +46,18 @@ function initStream(stream, args) {
})
}
function updateStream(stream, value, error) {
if (!absorbStream(stream, value, false) && !absorbStream(stream, error, true)) {
updateState(stream, value, error)
for (var id in stream._state.deps) updateDependency(stream._state.deps[id], false)
finalize(stream)
}
updateState(stream, value, error)
for (var id in stream._state.deps) updateDependency(stream._state.deps[id], false)
finalize(stream)
}
function updateState(stream, value, error) {
error = unwrapError(value, error)
if (error !== undefined && typeof stream._state.recover === "function") {
try {updateValues(stream, stream._state.recover(), undefined)}
try {
var recovered = stream._state.recover()
if (recovered === HALT) return
updateValues(stream, recovered, undefined)
}
catch (e) {updateValues(stream, undefined, e)}
}
else updateValues(stream, value, error)
@ -73,7 +76,8 @@ function updateDependency(stream, mustSync) {
else {
try {
var value = state.derive()
if (!absorbStream(stream, value)) updateState(stream, value, undefined)
if (value === HALT) return
updateState(stream, value, undefined)
}
catch (e) {
updateState(stream, undefined, e)
@ -81,30 +85,29 @@ function updateDependency(stream, mustSync) {
}
}
}
function absorbStream(stream, value, isError) {
function unwrapError(value, error) {
if (value != null && value.constructor === createStream) {
if (value._state.state === 2) {
stream.end(true)
stream(value())
}
else if (value._state.error) stream.error(value.error())
else if (value._state.state === 0) return true
else if (!isError) stream(value())
else stream.error(value())
return true
if (value._state.error !== undefined) error = value._state.error
else error = unwrapError(value._state.value, value._state.error)
}
return false
return error
}
function finalize(stream) {
stream._state.changed = false
for (var id in stream._state.deps) stream._state.deps[id]._state.changed = false
}
function run(fn) {
var self = createStream(), stream = this
return initDependency(self, [stream], function() {
return absorb(self, fn(stream()))
}, undefined)
}
function doCatch(fn) {
var stream = this
var self = createStream(), stream = this
var derive = function() {return stream._state.value}
var recover = function() {return fn(stream._state.error)}
return initDependency(createStream(), [stream], derive, recover)
var recover = function() {return absorb(self, fn(stream._state.error))}
return initDependency(self, [stream], derive, recover)
}
function combine(fn, streams) {
return initDependency(createStream(), streams, function() {
@ -113,6 +116,16 @@ function combine(fn, streams) {
return fn.apply(this, streams.concat([streams.filter(changed)]))
}, undefined)
}
function absorb(stream, value) {
if (value != null && value.constructor === createStream) {
value.error.map(stream.error)
value.map(stream)
if (value._state.state === 0) return HALT
if (value._state.error) throw value._state.error
value = value._state.value
}
return value
}
function initDependency(dep, streams, derive, recover) {
var state = dep._state
@ -148,6 +161,7 @@ function unregisterStream(stream) {
function map(fn) {return combine(function(stream) {return fn(stream())}, [this])}
function ap(stream) {return combine(function(s1, s2) {return s1()(s2())}, [this, stream])}
function valueOf() {return this._state.value}
function toJSON() {return JSON.stringify(this._state.value)}
function active(stream) {return stream._state.state === 1}
function changed(stream) {return stream._state.changed}
@ -160,4 +174,4 @@ function reject(e) {
return stream
}
module.exports = {stream: createStream, combine: combine, reject: reject}
module.exports = {stream: createStream, combine: combine, reject: reject, HALT: HALT}