git.iko.soy/purescript-stream
on master help sign in
feat(stream): implement more helpers and combinators
8e8b516Tylor Steinberger
.gitignore+1
compare with master
unifiedsplit
@@ -5,3 +5,4 @@
5/generated-docs/
5/generated-docs/
6/.psc*
6/.psc*
7/.psa*
7/.psa*
8+src/Main.purs
bower.json+1−2
@@ -9,8 +9,7 @@
9 "dependencies": {
9 "dependencies": {
10 "purescript-eff": "^2.0.0",
10 "purescript-eff": "^2.0.0",
11 "purescript-prelude": "^2.5.0",
11 "purescript-prelude": "^2.5.0",
12− "purescript-aff-promise": "^0.4.0",
12+ "purescript-refs": "^2.0.0"
13− "purescript-aff": "^2.0.3"
14 },
13 },
15 "devDependencies": {
14 "devDependencies": {
16 "purescript-console": "^2.0.0",
15 "purescript-console": "^2.0.0",
ffi/Control/Stream/combinators/combine.jsnew file+97
@@ -0,0 +1,97 @@
1+/** @license MIT License (c) copyright 2010-2016 original author or authors */
2+/** @author Brian Cavalier */
3+/** @author John Hann */
4+
5+import { map, tail } from '@most/prelude'
6+
7+import { IndexSink } from '../sink/IndexSink'
8+import { Stream } from '../Stream'
9+import { disposeAll } from '../disposables'
10+
11+export const combine2 = f => s1 => s2 => combineArray(f, [s1, s2])
12+export const combine3 = f => s1 => s2 => s3 => combineArray(f, [s1, s2, s3])
13+export const combine4 = f => s1 => s2 => s3 => s4 => combineArray(f, [s1, s2, s3, s4])
14+export const combine5 = f => s1 => s2 => s3 => s4 => s5 => combineArray(f, [s1, s2, s3, s4, s5])
15+
16+function combineArray (f, streams) {
17+ return new Stream(combineSources(f, streams))
18+}
19+
20+function combineSources (f, streams) {
21+ return new Combine(f, map(getSource, streams))
22+}
23+
24+function getSource (stream) {
25+ return stream.source
26+}
27+
28+function Combine (f, sources) {
29+ this.f = f
30+ this.sources = sources
31+}
32+
33+Combine.prototype.run = function (sink) {
34+ return scheduler => {
35+ var l = this.sources.length
36+ var disposables = new Array(l)
37+ var sinks = new Array(l)
38+
39+ var mergeSink = new CombineSink(disposables, sinks, sink, this.f)
40+
41+ for (var indexSink, i = 0; i < l; ++i) {
42+ indexSink = sinks[i] = new IndexSink(i, mergeSink)
43+ disposables[i] = this.sources[i].run(indexSink)(scheduler)
44+ }
45+
46+ return disposeAll(disposables)
47+ }
48+}
49+
50+function CombineSink (disposables, sinks, sink, f) {
51+ this.sink = sink
52+ this.disposables = disposables
53+ this.sinks = sinks
54+ this.f = f
55+
56+ var l = sinks.length
57+ this.awaiting = l
58+ this.values = new Array(l)
59+ this.hasValue = new Array(l)
60+
61+ for (var i = 0; i < l; ++i) {
62+ this.hasValue[i] = false
63+ }
64+
65+ this.activeCount = sinks.length
66+}
67+
68+CombineSink.prototype.event = function (t) {
69+ return indexedValue => {
70+ var i = indexedValue.index
71+ var awaiting = this._updateReady(i)
72+
73+ this.values[i] = indexedValue.value
74+ if (awaiting === 0) {
75+ const value = tail(this.values).reduce((f, x) => f(x), this.f(this.values[0]))
76+ this.sink.event(t)(value)
77+ }
78+ }
79+}
80+
81+CombineSink.prototype._updateReady = function (index) {
82+ if (this.awaiting > 0) {
83+ if (!this.hasValue[index]) {
84+ this.hasValue[index] = true
85+ this.awaiting -= 1
86+ }
87+ }
88+ return this.awaiting
89+}
90+
91+CombineSink.prototype.end = function (t, indexedValue) {
92+ this.disposables[indexedValue.index].dispose()
93+
94+ if (--this.activeCount === 0) {
95+ this.sink.end(t)
96+ }
97+}
ffi/Control/Stream/combinators/concat.jsnew file+56
@@ -0,0 +1,56 @@
1+import { MemoizedDisposable } from '../disposables/MemoizedDisposable'
2+import { Stream } from '../Stream'
3+
4+export const concat = s1 => s2 => continueWith(() => s2)(s1)
5+
6+export const continueWith = f => stream => new Stream(new ContinueWith(f, stream.source))
7+
8+function ContinueWith (f, source) {
9+ this.f = f
10+ this.source = source
11+}
12+
13+ContinueWith.prototype.run = function (sink) {
14+ return scheduler => {
15+ return new ContinueWithSink(this.f, this.source, sink, scheduler)
16+ }
17+}
18+
19+function ContinueWithSink (f, source, sink, scheduler) {
20+ this.f = f
21+ this.sink = sink
22+ this.scheduler = scheduler
23+ this.active = true
24+ this.disposable = new MemoizedDisposable(source.run(this)(scheduler))
25+}
26+
27+ContinueWithSink.prototype.event = function (t) {
28+ return x => {
29+ if (!this.active) {
30+ return
31+ }
32+ this.sink.event(t)(x)
33+ }
34+}
35+
36+ContinueWithSink.prototype.end = function (t, x) {
37+ if (!this.active) {
38+ return
39+ }
40+
41+ this.disposable.dispose()
42+ this._startNext(t, x, this.sink)
43+}
44+
45+ContinueWithSink.prototype._startNext = function (t, x, sink) {
46+ this._continue(this.f, x, sink)
47+}
48+
49+ContinueWithSink.prototype._continue = function (f, x, sink) {
50+ return f(x).source.run(sink)(this.scheduler)
51+}
52+
53+ContinueWithSink.prototype.dispose = function () {
54+ this.active = false
55+ return this.disposable.dispose()
56+}
ffi/Control/Stream/combinators/index.js+5−1
@@ -1,2 +1,6 @@
1−export * from './tap'
2export * from './drain'
1export * from './drain'
2+export * from './mergeConcurrently'
3+export * from './combine'
4+export * from './merge'
5+export * from './concat'
6+export * from './switch'
ffi/Control/Stream/combinators/merge.jsnew file+74
@@ -0,0 +1,74 @@
1+/** @license MIT License (c) copyright 2010-2016 original author or authors */
2+/** @author Brian Cavalier */
3+/** @author John Hann */
4+
5+import { IndexSink } from '../sink/IndexSink'
6+import { Stream } from '../Stream'
7+import { disposeAll } from '../disposables'
8+import { reduce } from '@most/prelude'
9+
10+/**
11+ * @param {Array} streams array of stream to merge
12+ * @returns {Stream} stream containing events from all input observables
13+ * in time order. If two events are simultaneous they will be merged in
14+ * arbitrary order.
15+ */
16+export function merge (streams) {
17+ return new Stream(mergeSources(streams))
18+}
19+
20+/**
21+ * This implements fusion/flattening for merge. It will
22+ * fuse adjacent merge operations. For example:
23+ * - a.merge(b).merge(c) effectively becomes merge(a, b, c)
24+ * - merge(a, merge(b, c)) effectively becomes merge(a, b, c)
25+ * It does this by concatenating the sources arrays of
26+ * any nested Merge sources, in effect "flattening" nested
27+ * merge operations into a single merge.
28+ */
29+function mergeSources (streams) {
30+ return new Merge(reduce(appendSources, [], streams))
31+}
32+
33+function appendSources (sources, stream) {
34+ var source = stream.source
35+ return source instanceof Merge
36+ ? sources.concat(source.sources)
37+ : sources.concat(source)
38+}
39+
40+function Merge (sources) {
41+ this.sources = sources
42+}
43+
44+Merge.prototype.run = function (sink, scheduler) {
45+ var l = this.sources.length
46+ var disposables = new Array(l)
47+ var sinks = new Array(l)
48+
49+ var mergeSink = new MergeSink(disposables, sinks, sink)
50+
51+ for (var indexSink, i = 0; i < l; ++i) {
52+ indexSink = sinks[i] = new IndexSink(i, mergeSink)
53+ disposables[i] = this.sources[i].run(indexSink, scheduler)
54+ }
55+
56+ return disposeAll(disposables)
57+}
58+
59+function MergeSink (disposables, sinks, sink) {
60+ this.sink = sink
61+ this.disposables = disposables
62+ this.activeCount = sinks.length
63+}
64+
65+MergeSink.prototype.event = function (t, indexValue) {
66+ this.sink.event(t, indexValue.value)
67+}
68+
69+MergeSink.prototype.end = function (t, indexedValue) {
70+ dispose.tryDispose(t, this.disposables[indexedValue.index], this.sink)
71+ if (--this.activeCount === 0) {
72+ this.sink.end(t, indexedValue.value)
73+ }
74+}
ffi/Control/Stream/combinators/mergeConcurrently.js+30−27
@@ -2,15 +2,23 @@
2/** @author Brian Cavalier */
2/** @author Brian Cavalier */
3/** @author John Hann */
3/** @author John Hann */
4
4
5+import { curry2, curry3, id as identity } from '@most/prelude'
6+
5import { LinkedList } from '../LinkedList'
7import { LinkedList } from '../LinkedList'
6−import { MemoizedDisosable } from '../disposables/MemoizedDisposable'
8+import { MemoizedDisposable } from '../disposables/MemoizedDisposable'
7−import { id as identity } from '@most/prelude'
9+import { Stream } from '../Stream'
10+
11+export const flatMap =
12+ curry2((stream, f) => mergeMapConcurrently(f, Infinity, stream))
13+
14+export const mergeConcurrently = curry2((concurrency, stream) =>
15+ mergeMapConcurrently(identity, concurrency, stream))
8
16
9−export const mergeConcurrently = (concurrency, stream) =>
17+export const join =
10− mergeMapConcurrently(identity, concurrency, stream)
18+ mergeConcurrently(Infinity)
11
19
12−export const mergeMapConcurrently = (f, concurrency, stream) =>
20+export const mergeMapConcurrently = curry3((f, concurrency, stream) =>
13− new MergeConcurrently(f, concurrency, stream)
21+ new Stream(new MergeConcurrently(f, concurrency, stream.source)))
14
22
15class MergeConcurrently {
23class MergeConcurrently {
16 constructor (f, concurrency, source) {
24 constructor (f, concurrency, source) {
@@ -19,8 +27,8 @@ class MergeConcurrently {
19 this.source = source
27 this.source = source
20 }
28 }
21
29
22− run (sink, scheduler) {
30+ run (sink) {
23− return new Outer(this.f, this.concurrency, this.source, sink, scheduler)
31+ return scheduler => new Outer(this.f, this.concurrency, this.source, sink, scheduler)
24 }
32 }
25}
33}
26
34
@@ -32,12 +40,12 @@ class Outer {
32 this.scheduler = scheduler
40 this.scheduler = scheduler
33 this.pending = []
41 this.pending = []
34 this.current = new LinkedList()
42 this.current = new LinkedList()
35− this.disposable = new MemoizedDisosable(source.run(this, scheduler))
43+ this.disposable = new MemoizedDisposable(source.run(this)(scheduler))
36 this.active = true
44 this.active = true
37 }
45 }
38
46
39− event (t, x) {
47+ event (t) {
40− this._addInner(t, x)
48+ return x => this._addInner(t, x)
41 }
49 }
42
50
43 _addInner (t, x) {
51 _addInner (t, x) {
@@ -62,10 +70,10 @@ class Outer {
62 this.current.add(innerSink)
70 this.current.add(innerSink)
63 }
71 }
64
72
65− end (t, x) {
73+ end (t) {
66 this.active = false
74 this.active = false
67 this.disposable.dispose()
75 this.disposable.dispose()
68− this._checkEnd(t, x)
76+ this._checkEnd(t)
69 }
77 }
70
78
71 dispose () {
79 dispose () {
@@ -75,26 +83,25 @@ class Outer {
75 this.current.dispose()
83 this.current.dispose()
76 }
84 }
77
85
78− _endInner (t, x, inner) {
86+ _endInner (t, inner) {
79 this.current.remove(inner)
87 this.current.remove(inner)
80 inner.dispose()
88 inner.dispose()
81
89
82 if (this.pending.length === 0) {
90 if (this.pending.length === 0) {
83− this._checkEnd(t, x)
91+ this._checkEnd(t)
84 } else {
92 } else {
85 this._startInner(t, this.pending.shift())
93 this._startInner(t, this.pending.shift())
86 }
94 }
87 }
95 }
88
96
89− _checkEnd (t, x) {
97+ _checkEnd (t) {
90 if (!this.active && this.current.isEmpty()) {
98 if (!this.active && this.current.isEmpty()) {
91− this.sink.end(t, x)
99+ this.sink.end(t)
92 }
100 }
93 }
101 }
94}
102}
95
103
96−const mapAndRun = (f, x, sink, scheduler) =>
104+const mapAndRun = (f, x, sink, scheduler) => f(x).source.run(sink)(scheduler)
97− f(x).run(sink, scheduler)
98
105
99class Inner {
106class Inner {
100 constructor (time, outer, sink) {
107 constructor (time, outer, sink) {
@@ -105,16 +112,12 @@ class Inner {
105 this.disposable = void 0
112 this.disposable = void 0
106 }
113 }
107
114
108− event (t, x) {
115+ event (t) {
109− this.sink.event(Math.max(t, this.time), x)
116+ return x => this.sink.event(Math.max(t, this.time))(x)
110− }
111−
112− end (t, x) {
113− this.outer._endInner(Math.max(t, this.time), x, this)
114 }
117 }
115
118
116− error (t, e) {
119+ end (t) {
117− this.outer.error(Math.max(t, this.time), e)
120+ this.outer._endInner(Math.max(t, this.time), this)
118 }
121 }
119
122
120 dispose () {
123 dispose () {
ffi/Control/Stream/combinators/switch.jsnew file+99
@@ -0,0 +1,99 @@
1+import { disposeAll, emptyDisposable } from '../disposables'
2+
3+import { Stream } from '../Stream'
4+
5+/**
6+ * Given a stream of streams, return a new stream that adopts the behavior
7+ * of the most recent inner stream.
8+ * @param {Stream} stream of streams on which to switch
9+ * @returns {Stream} switching stream
10+ */
11+function switchLatest (stream) {
12+ return new Stream(new Switch(stream.source))
13+}
14+
15+export { switchLatest as switch }
16+
17+function Switch (source) {
18+ this.source = source
19+}
20+
21+Switch.prototype.run = function (sink) {
22+ return scheduler => {
23+ const switchSink = new SwitchSink(sink, scheduler)
24+ return disposeAll([switchSink, this.source.run(switchSink)(scheduler)])
25+ }
26+}
27+
28+function SwitchSink (sink, scheduler) {
29+ this.sink = sink
30+ this.scheduler = scheduler
31+ this.current = null
32+ this.ended = false
33+}
34+
35+SwitchSink.prototype.event = function (t) {
36+ return stream => {
37+ this._disposeCurrent(t) // TODO: capture the result of this dispose
38+ this.current = new Segment(t, Infinity, this, this.sink)
39+ this.current.disposable = stream.source.run(this.current)(this.scheduler)
40+ }
41+}
42+
43+SwitchSink.prototype.end = function (t, x) {
44+ this.ended = true
45+ this._checkEnd(t, x)
46+}
47+
48+SwitchSink.prototype.dispose = function () {
49+ return this._disposeCurrent(this.scheduler.now())
50+}
51+
52+SwitchSink.prototype._disposeCurrent = function (t) {
53+ if (this.current !== null) {
54+ return this.current._dispose(t)
55+ }
56+}
57+
58+SwitchSink.prototype._disposeInner = function (t, inner) {
59+ inner._dispose(t) // TODO: capture the result of this dispose
60+ if (inner === this.current) {
61+ this.current = null
62+ }
63+}
64+
65+SwitchSink.prototype._checkEnd = function (t, x) {
66+ if (this.ended && this.current === null) {
67+ this.sink.end(t, x)
68+ }
69+}
70+
71+SwitchSink.prototype._endInner = function (t, x, inner) {
72+ this._disposeInner(t, inner)
73+ this._checkEnd(t, x)
74+}
75+
76+function Segment (min, max, outer, sink) {
77+ this.min = min
78+ this.max = max
79+ this.outer = outer
80+ this.sink = sink
81+ this.disposable = emptyDisposable
82+}
83+
84+Segment.prototype.event = function (t) {
85+ return x => {
86+ if (t < this.max) {
87+ this.sink.event(Math.max(t, this.min))(x)
88+ }
89+ }
90+}
91+
92+Segment.prototype.end = function (t, x) {
93+ this.outer._endInner(Math.max(t, this.min), x, this)
94+}
95+
96+Segment.prototype._dispose = function (t) {
97+ this.max = t
98+ this.disposable.dispose()
99+}
ffi/Control/Stream/combinators/tap.jsdeleted−50
@@ -1,50 +0,0 @@
1−import { Stream } from '../Stream'
2−import { curry2 } from '@most/prelude'
3−
4−export const tapEnd = curry2((f, stream) => new Stream(new TapEnd(f, stream.source)))
5−
6−export const tapEvent = curry2((f, stream) => new Stream(new TapEvent(f, stream.source)))
7−
8−class TapEvent {
9− constructor (f, source) {
10− this.source = source
11− this.f = f
12−
13− this.run = sink => scheduler =>
14− source.run(new TapSink(f, sink))(scheduler)
15− }
16−}
17−
18−class TapSink {
19− constructor (f, sink) {
20− this.event = t => x => {
21− f(x)()
22− sink.event(t)(x)
23− }
24−
25− this.end = t => {
26− sink.end(t)
27− }
28− }
29−}
30−
31−class TapEnd {
32− constructor (f, source) {
33− this.source = source
34− this.f = f
35−
36− this.run = sink => scheduler =>
37− source.run(new TapEndSink(f, sink))(scheduler)
38− }
39−}
40−
41−class TapEndSink {
42− constructor (f, sink) {
43− this.event = t => x => sink.event(t)(x)
44−
45− this.end = t => {
46− f()
47− sink.end(t)
48− }
49− }
50−}
ffi/Control/Stream/disposables/Disposable.js+1−3
@@ -7,9 +7,7 @@ export class Disposable {
7 constructor (dispose, data) {
7 constructor (dispose, data) {
8 this._dispose = dispose
8 this._dispose = dispose
9 this._data = data
9 this._data = data
10− }
11
10
12− dispose () {
11+ this.dispose = () => dispose(data)
13− return this._dispose(this._data)
14 }
12 }
15}
13}
ffi/Control/Stream/disposables/MemoizedDisposable.js+1−1
@@ -8,7 +8,7 @@ export class MemoizedDisposable {
8 dispose () {
8 dispose () {
9 if (!this.disposed) {
9 if (!this.disposed) {
10 this.disposed = true
10 this.disposed = true
11− this.value = disposeSafely(this.disposable)
11+ this.value = this.disposable.dispose()
12 this.disposable = undefined
12 this.disposable = undefined
13 }
13 }
14
14
ffi/Control/Stream/disposables/index.js+1−1
@@ -9,4 +9,4 @@ export const createDisposable = curry2((dispose, data) => new MemoizedDisposable
9export const emptyDisposable = new Disposable(id, undefined)
9export const emptyDisposable = new Disposable(id, undefined)
10
10
11export const disposeAll = disposables =>
11export const disposeAll = disposables =>
12− createDisposable(map(disposeSafely, disposables), disposables)
12+ createDisposable(curry2(map)(disposeSafely), disposables)
ffi/Control/Stream/runEffects.js+2
@@ -11,6 +11,8 @@ function runSource (source, scheduler) {
11 const observer = new RunEffectsSink(disposable)
11 const observer = new RunEffectsSink(disposable)
12
12
13 disposable.setDisposable(source.run(observer)(scheduler))
13 disposable.setDisposable(source.run(observer)(scheduler))
14+
15+ return {}
14}
16}
15
17
16var RunEffectsSink = function RunEffectsSink (disposable) {
18var RunEffectsSink = function RunEffectsSink (disposable) {
ffi/Control/Stream/scheduler/ScheduledTask.js+15−10
@@ -1,14 +1,19 @@
1/* @flow */
1/* @flow */
2+export class ScheduledTask {
2−export function ScheduledTask (delay, period, task, scheduler) {
3+ constructor (delay, period, task, scheduler) {
3− this.time = delay
4+ this.time = delay
4− this.period = period
5+ this.period = period
5− this.task = task
6+ this.task = task
6− this.scheduler = scheduler
7+ this.scheduler = scheduler
7− this.active = true
8+ this.active = true
9+ }
10+
11+ run () {
12+ this.task.run(this.time)
13+ }
8
14
9− this.run = () => task.run(this.time)
10− this.dispose = () => {
15+ dispose () {
11− scheduler.cancel(this)
16+ this.scheduler.cancel(this)
12− task.dispose()
17+ this.task.dispose()
13 }
18 }
14}
19}
ffi/Control/Stream/scheduler/Scheduler.js+13−11
@@ -25,24 +25,26 @@ export class Scheduler {
25 }
25 }
26
26
27 asap (task) {
27 asap (task) {
28− return this.schedule(0, -1, task)
28+ return this.schedule(0)(-1)(task)
29 }
29 }
30
30
31− delay (delay, task) {
31+ delay (delay) {
32− return this.schedule(delay, -1, task)
32+ return task => this.schedule(delay)(-1)(task)
33 }
33 }
34
34
35− periodic (period, task) {
35+ periodic (period) {
36− return this.schedule(0, period, task)
36+ return task => this.schedule(0)(period)(task)
37 }
37 }
38
38
39− schedule (delay, period, task) {
39+ schedule (delay) {
40+ return period => task => {
40− const now = this.now()
41+ const now = this.now()
41− const st = new ScheduledTask(now + Math.max(0, delay), period, task, this)
42+ const st = new ScheduledTask(now + Math.max(0, delay), period, task, this)
42
43
43− this.timeline.add(st)
44+ this.timeline.add(st)
44− this._scheduleNextRun(now)
45+ this._scheduleNextRun(now)
45− return st
46+ return st
47+ }
46 }
48 }
47
49
48 cancel (task) {
50 cancel (task) {
ffi/Control/Stream/sink/IndexSink.js+2−2
@@ -15,10 +15,10 @@ export class IndexSink {
15 }
15 }
16 }
16 }
17
17
18− end (t, x) {
18+ end (t) {
19 if (!this.active) return
19 if (!this.active) return
20
20
21 this.active = false
21 this.active = false
22− this.sink.end(t, { index: this.index, value: x })
22+ this.sink.end(t, { index: this.index })
23 }
23 }
24}
24}
package.json+1
@@ -4,6 +4,7 @@
4 "description": "",
4 "description": "",
5 "main": "index.js",
5 "main": "index.js",
6 "scripts": {
6 "scripts": {
7+ "start": "yarn build && pulp run",
7 "build": "rollup -c && pulp build",
8 "build": "rollup -c && pulp build",
8 "changelog": "conventional-changelog -i CHANGELOG.md -s -r 0 -p angular",
9 "changelog": "conventional-changelog -i CHANGELOG.md -s -r 0 -p angular",
9 "clean": "rimraf lib lib.es2015",
10 "clean": "rimraf lib lib.es2015",
src/Control/Stream.js+552−117
@@ -16,11 +16,38 @@ Object.defineProperty(exports, '__esModule', { value: true });
16
16
17// drop :: Int -> [a] -> [a]
17// drop :: Int -> [a] -> [a]
18// drop first n elements
18// drop first n elements
19+function drop (n, a) { // eslint-disable-line complexity
20+ if (n < 0) {
21+ throw new TypeError('n must be >= 0')
22+ }
23+
24+ var l = a.length;
25+ if (n === 0 || l === 0) {
26+ return a
27+ }
28+
29+ if (n >= l) {
30+ return []
31+ }
19
32
33+ return unsafeDrop(n, a, l - n)
34+}
35+
36+// unsafeDrop :: Int -> [a] -> Int -> [a]
37+// Internal helper for drop
38+function unsafeDrop (n, a, l) {
39+ var b = new Array(l);
40+ for (var i = 0; i < l; ++i) {
41+ b[i] = a[n + i];
42+ }
43+ return b
44+}
20
45
21// tail :: [a] -> [a]
46// tail :: [a] -> [a]
22// drop head element
47// drop head element
23−
48+function tail (a) {
49+ return drop(1, a)
50+}
24
51
25// copy :: [a] -> [a]
52// copy :: [a] -> [a]
26// duplicate a (shallow duplication)
53// duplicate a (shallow duplication)
@@ -39,7 +66,13 @@ function map (f, a) {
39
66
40// reduce :: (a -> b -> a) -> a -> [b] -> a
67// reduce :: (a -> b -> a) -> a -> [b] -> a
41// accumulate via left-fold
68// accumulate via left-fold
42−
69+function reduce (f, z, a) {
70+ var r = z;
71+ for (var i = 0, l = a.length; i < l; ++i) {
72+ r = f(r, a[i], i);
73+ }
74+ return r
75+}
43
76
44// replace :: a -> Int -> [a]
77// replace :: a -> Int -> [a]
45// replace element at index
78// replace element at index
@@ -124,10 +157,8 @@ function curry3 (f) {
124var Disposable = function Disposable (dispose, data) {
157var Disposable = function Disposable (dispose, data) {
125 this._dispose = dispose;
158 this._dispose = dispose;
126 this._data = data;
159 this._data = data;
127−};
128
160
129−Disposable.prototype.dispose = function dispose () {
161+ this.dispose = function () { return dispose(data); };
130− return this._dispose(this._data)
131};
162};
132
163
133var MemoizedDisposable = function MemoizedDisposable (disposable) {
164var MemoizedDisposable = function MemoizedDisposable (disposable) {
@@ -139,14 +170,14 @@ var MemoizedDisposable = function MemoizedDisposable (disposable) {
139MemoizedDisposable.prototype.dispose = function dispose () {
170MemoizedDisposable.prototype.dispose = function dispose () {
140 if (!this.disposed) {
171 if (!this.disposed) {
141 this.disposed = true;
172 this.disposed = true;
142− this.value = disposeSafely(this.disposable);
173+ this.value = this.disposable.dispose();
143 this.disposable = undefined;
174 this.disposable = undefined;
144 }
175 }
145
176
146 return this.value
177 return this.value
147};
178};
148
179
149−function disposeSafely$1 (disposable) {
180+function disposeSafely (disposable) {
150 disposable.dispose();
181 disposable.dispose();
151}
182}
152
183
@@ -154,7 +185,7 @@ var createDisposable = curry2(function (dispose, data) { return new MemoizedDisp
154
185
155var emptyDisposable = new Disposable(id, undefined);
186var emptyDisposable = new Disposable(id, undefined);
156
187
157−var disposeAll = function (disposables) { return createDisposable(map(disposeSafely$1, disposables), disposables); };
188+var disposeAll = function (disposables) { return createDisposable(curry2(map)(disposeSafely), disposables); };
158
189
159/** @license MIT License (c) copyright 2010-2016 original author or authors */
190/** @license MIT License (c) copyright 2010-2016 original author or authors */
160/** @author Brian Cavalier */
191/** @author Brian Cavalier */
@@ -202,21 +233,22 @@ function runAsap (f) {
202}
233}
203
234
204/* */
235/* */
205−function ScheduledTask (delay, period, task, scheduler) {
236+var ScheduledTask = function ScheduledTask (delay, period, task, scheduler) {
206− var this$1 = this;
207−
208 this.time = delay;
237 this.time = delay;
209 this.period = period;
238 this.period = period;
210 this.task = task;
239 this.task = task;
211 this.scheduler = scheduler;
240 this.scheduler = scheduler;
212 this.active = true;
241 this.active = true;
242+};
213
243
244+ScheduledTask.prototype.run = function run () {
214− this.run = function () { return task.run(this$1.time); };
245+ this.task.run(this.time);
246+};
247+
215− this.dispose = function () {
248+ScheduledTask.prototype.dispose = function dispose () {
216− scheduler.cancel(this$1);
249+ this.scheduler.cancel(this);
217− task.dispose();
250+ this.task.dispose();
218− };
251+};
219−}
220
252
221/* */
253/* */
222function runTask$1 (task) {
254function runTask$1 (task) {
@@ -244,24 +276,32 @@ Scheduler.prototype.now = function now () {
244};
276};
245
277
246Scheduler.prototype.asap = function asap (task) {
278Scheduler.prototype.asap = function asap (task) {
247− return this.schedule(0, -1, task)
279+ return this.schedule(0)(-1)(task)
248};
280};
249
281
250−Scheduler.prototype.delay = function delay (delay$1, task) {
282+Scheduler.prototype.delay = function delay (delay$1) {
283+ var this$1 = this;
284+
251− return this.schedule(delay$1, -1, task)
285+ return function (task) { return this$1.schedule(delay$1)(-1)(task); }
252};
286};
253
287
254−Scheduler.prototype.periodic = function periodic (period, task) {
288+Scheduler.prototype.periodic = function periodic (period) {
289+ var this$1 = this;
290+
255− return this.schedule(0, period, task)
291+ return function (task) { return this$1.schedule(0)(period)(task); }
256};
292};
257
293
258−Scheduler.prototype.schedule = function schedule (delay, period, task) {
294+Scheduler.prototype.schedule = function schedule (delay) {
259− var now = this.now();
295+ var this$1 = this;
260− var st = new ScheduledTask(now + Math.max(0, delay), period, task, this);
261
296
297+ return function (period) { return function (task) {
298+ var now = this$1.now();
299+ var st = new ScheduledTask(now + Math.max(0, delay), period, task, this$1);
300+
262− this.timeline.add(st);
301+ this$1.timeline.add(st);
263− this._scheduleNextRun(now);
302+ this$1._scheduleNextRun(now);
264− return st
303+ return st
304+ }; }
265};
305};
266
306
267Scheduler.prototype.cancel = function cancel (task) {
307Scheduler.prototype.cancel = function cancel (task) {
@@ -478,48 +518,6 @@ var scheduleTasks = curry2(function (tasks, scheduler) {
478 }
518 }
479});
519});
480
520
481−var Stream = function Stream (source) {
482− this.source = source;
483−};
484−
485−var tapEnd = curry2(function (f, stream) { return new Stream(new TapEnd(f, stream.source)); });
486−
487−var tapEvent = curry2(function (f, stream) { return new Stream(new TapEvent(f, stream.source)); });
488−
489−var TapEvent = function TapEvent (f, source) {
490− this.source = source;
491− this.f = f;
492−
493− this.run = function (sink) { return function (scheduler) { return source.run(new TapSink(f, sink))(scheduler); }; };
494−};
495−
496−var TapSink = function TapSink (f, sink) {
497− this.event = function (t) { return function (x) {
498− f(x)();
499− sink.event(t)(x);
500− }; };
501−
502− this.end = function (t) {
503− sink.end(t);
504− };
505−};
506−
507−var TapEnd = function TapEnd (f, source) {
508− this.source = source;
509− this.f = f;
510−
511− this.run = function (sink) { return function (scheduler) { return source.run(new TapEndSink(f, sink))(scheduler); }; };
512−};
513−
514−var TapEndSink = function TapEndSink (f, sink) {
515− this.event = function (t) { return function (x) { return sink.event(t)(x); }; };
516−
517− this.end = function (t) {
518− f();
519− sink.end(t);
520− };
521−};
522−
523var SettableDisposable = function SettableDisposable () {
521var SettableDisposable = function SettableDisposable () {
524 this.disposable = void 0;
522 this.disposable = void 0;
525 this.disposed = false;
523 this.disposed = false;
@@ -559,6 +557,8 @@ function runSource (source, scheduler) {
559 var observer = new RunEffectsSink(disposable);
557 var observer = new RunEffectsSink(disposable);
560
558
561 disposable.setDisposable(source.run(observer)(scheduler));
559 disposable.setDisposable(source.run(observer)(scheduler));
560+
561+ return {}
562}
562}
563
563
564var RunEffectsSink = function RunEffectsSink (disposable) {
564var RunEffectsSink = function RunEffectsSink (disposable) {
@@ -577,100 +577,526 @@ RunEffectsSink.prototype.end = function end (t) {
577
577
578var drain = function (stream) { return function () { return runEffects(stream, defaultScheduler); }; };
578var drain = function (stream) { return function () { return runEffects(stream, defaultScheduler); }; };
579
579
580−var combine =
580+var LinkedList = function LinkedList () {
581− curry3(function (f, stream1, stream2) { return new Stream(new Combine(f, stream1.source, stream2.source)); });
581+ this.head = null;
582+ this.length = 0;
583+};
584+
585+LinkedList.prototype.add = function add (x) {
586+ if (this.head !== null) {
587+ this.head.prev = x;
588+ x.next = this.head;
589+ }
590+ this.head = x;
591+ ++this.length;
592+};
593+
594+LinkedList.prototype.remove = function remove (x) { // eslint-disable-linecomplexity
595+ --this.length;
596+ if (x === this.head) {
597+ this.head = this.head.next;
598+ }
599+ if (x.next !== null) {
600+ x.next.prev = x.prev;
601+ x.next = null;
602+ }
603+ if (x.prev !== null) {
604+ x.prev.next = x.next;
605+ x.prev = null;
606+ }
607+};
608+
609+LinkedList.prototype.isEmpty = function isEmpty () {
610+ return this.length === 0
611+};
612+
613+LinkedList.prototype.dispose = function dispose () {
614+ if (this.isEmpty()) { return }
615+
616+ var x = this.head;
617+ this.head = null;
618+ this.length = 0;
619+
620+ while (x !== null) {
621+ x.dispose();
622+ x = x.next;
623+ }
624+};
625+
626+var Stream = function Stream (source) {
627+ this.source = source;
628+};
629+
630+/** @license MIT License (c) copyright 2010-2016 original author or authors */
631+/** @author Brian Cavalier */
632+/** @author John Hann */
582
633
583−var Combine = function Combine (f, source1, source2) {
634+var flatMap =
635+ curry2(function (stream, f) { return mergeMapConcurrently(f, Infinity, stream); });
636+
637+var mergeConcurrently = curry2(function (concurrency, stream) { return mergeMapConcurrently(id, concurrency, stream); });
638+
639+var join =
640+ mergeConcurrently(Infinity);
641+
642+var mergeMapConcurrently = curry3(function (f, concurrency, stream) { return new Stream(new MergeConcurrently(f, concurrency, stream.source)); });
643+
644+var MergeConcurrently = function MergeConcurrently (f, concurrency, source) {
584 this.f = f;
645 this.f = f;
585− this.source1 = source1;
646+ this.concurrency = concurrency;
586− this.source2 = source2;
647+ this.source = source;
587};
648};
588
649
589−Combine.prototype.run = function run (sink) {
650+MergeConcurrently.prototype.run = function run (sink) {
590 var this$1 = this;
651 var this$1 = this;
591
652
592− return function (scheduler) {
653+ return function (scheduler) { return new Outer(this$1.f, this$1.concurrency, this$1.source, sink, scheduler); }
593− var disposables = Array(2);
654+};
594− var sinks = Array(2);
595
655
596− var mergeSink = new CombineSink(disposables, sinks, sink, this$1.f);
656+var Outer = function Outer (f, concurrency, source, sink, scheduler) {
657+ this.f = f;
658+ this.concurrency = concurrency;
659+ this.sink = sink;
660+ this.scheduler = scheduler;
661+ this.pending = [];
662+ this.current = new LinkedList();
663+ this.disposable = new MemoizedDisposable(source.run(this)(scheduler));
664+ this.active = true;
665+};
597
666
598− sinks[0] = new IndexSink(0, mergeSink);
667+Outer.prototype.event = function event (t) {
599− sinks[1] = new IndexSink(1, mergeSink);
668+ var this$1 = this;
600− disposables[0] = this$1.source1.run(sinks[0])(scheduler);
601− disposables[1] = this$1.source2.run(sinks[1])(scheduler);
602
669
603− return disposeAll(disposables)
670+ return function (x) { return this$1._addInner(t, x); }
671+};
672+
673+Outer.prototype._addInner = function _addInner (t, x) {
674+ if (this.current.length < this.concurrency) {
675+ this._startInner(t, x);
676+ } else {
677+ this.pending.push(x);
604 }
678 }
605};
679};
606
680
607−var IndexSink = function IndexSink () {};
681+Outer.prototype._startInner = function _startInner (t, x) {
682+ try {
683+ this._initInner(t, x);
684+ } catch (e) {
685+ this.error(t, e);
686+ }
687+};
688+
689+Outer.prototype._initInner = function _initInner (t, x) {
690+ var innerSink = new Inner(t, this, this.sink);
691+ innerSink.disposable = mapAndRun(this.f, x, innerSink, this.scheduler);
692+ this.current.add(innerSink);
693+};
694+
695+Outer.prototype.end = function end (t) {
696+ this.active = false;
697+ this.disposable.dispose();
698+ this._checkEnd(t);
699+};
608
700
609−IndexSink.prototype.cosntructor = function cosntructor (i, sink) {
701+Outer.prototype.dispose = function dispose () {
610− this.i = i;
702+ this.active = false;
703+ this.pending.length = 0;
704+ this.disposable.dispose();
705+ this.current.dispose();
706+};
707+
708+Outer.prototype._endInner = function _endInner (t, inner) {
709+ this.current.remove(inner);
710+ inner.dispose();
711+
712+ if (this.pending.length === 0) {
713+ this._checkEnd(t);
714+ } else {
715+ this._startInner(t, this.pending.shift());
716+ }
717+};
718+
719+Outer.prototype._checkEnd = function _checkEnd (t) {
720+ if (!this.active && this.current.isEmpty()) {
721+ this.sink.end(t);
722+ }
723+};
724+
725+var mapAndRun = function (f, x, sink, scheduler) { return f(x).source.run(sink)(scheduler); };
726+
727+var Inner = function Inner (time, outer, sink) {
728+ this.prev = this.next = null;
729+ this.time = time;
730+ this.outer = outer;
731+ this.sink = sink;
732+ this.disposable = void 0;
733+};
734+
735+Inner.prototype.event = function event (t) {
736+ var this$1 = this;
737+
738+ return function (x) { return this$1.sink.event(Math.max(t, this$1.time))(x); }
739+};
740+
741+Inner.prototype.end = function end (t) {
742+ this.outer._endInner(Math.max(t, this.time), this);
743+};
744+
745+Inner.prototype.dispose = function dispose () {
746+ return this.disposable.dispose()
747+};
748+
749+var IndexSink = function IndexSink (i, sink) {
611 this.sink = sink;
750 this.sink = sink;
751+ this.index = i;
752+ this.active = true;
753+ this.value = void 0;
612};
754};
613
755
614IndexSink.prototype.event = function event (t) {
756IndexSink.prototype.event = function event (t) {
615 var this$1 = this;
757 var this$1 = this;
616
758
617 return function (x) {
759 return function (x) {
760+ if (!this$1.active) { return }
761+
762+ this$1.value = x;
618− this$1.sink.event(t)({ index: this$1.i, value: x });
763+ this$1.sink.event(t)(this$1);
619 }
764 }
620};
765};
621
766
622IndexSink.prototype.end = function end (t) {
767IndexSink.prototype.end = function end (t) {
768+ if (!this.active) { return }
769+
770+ this.active = false;
623− this.sink.end(t, this.i);
771+ this.sink.end(t, { index: this.index });
624};
772};
625
773
774+/** @license MIT License (c) copyright 2010-2016 original author or authors */
775+/** @author Brian Cavalier */
776+/** @author John Hann */
777+
778+var combine2 = function (f) { return function (s1) { return function (s2) { return combineArray(f, [s1, s2]); }; }; };
779+var combine3 = function (f) { return function (s1) { return function (s2) { return function (s3) { return combineArray(f, [s1, s2, s3]); }; }; }; };
780+var combine4 = function (f) { return function (s1) { return function (s2) { return function (s3) { return function (s4) { return combineArray(f, [s1, s2, s3, s4]); }; }; }; }; };
781+var combine5 = function (f) { return function (s1) { return function (s2) { return function (s3) { return function (s4) { return function (s5) { return combineArray(f, [s1, s2, s3, s4, s5]); }; }; }; }; }; };
782+
783+function combineArray (f, streams) {
784+ return new Stream(combineSources(f, streams))
785+}
786+
787+function combineSources (f, streams) {
788+ return new Combine(f, map(getSource, streams))
789+}
790+
791+function getSource (stream) {
792+ return stream.source
793+}
794+
795+function Combine (f, sources) {
796+ this.f = f;
797+ this.sources = sources;
798+}
799+
800+Combine.prototype.run = function (sink) {
801+ var this$1 = this;
802+
803+ return function (scheduler) {
804+ var l = this$1.sources.length;
805+ var disposables = new Array(l);
806+ var sinks = new Array(l);
807+
808+ var mergeSink = new CombineSink(disposables, sinks, sink, this$1.f);
809+
810+ for (var indexSink, i = 0; i < l; ++i) {
811+ indexSink = sinks[i] = new IndexSink(i, mergeSink);
812+ disposables[i] = this$1.sources[i].run(indexSink)(scheduler);
813+ }
814+
815+ return disposeAll(disposables)
816+ }
817+};
818+
626−var CombineSink = function CombineSink (disposables, sinks, sink, f) {
819+function CombineSink (disposables, sinks, sink, f) {
820+ var this$1 = this;
821+
822+ this.sink = sink;
627 this.disposables = disposables;
823 this.disposables = disposables;
628 this.sinks = sinks;
824 this.sinks = sinks;
629− this.sink = sink;
630 this.f = f;
825 this.f = f;
631
826
827+ var l = sinks.length;
632− this.awaiting = 2;
828+ this.awaiting = l;
633− this.values = Array(2);
829+ this.values = new Array(l);
634− this.hasValue = [false, false];
830+ this.hasValue = new Array(l);
635− this.activeCount = 2;
636−};
637
831
638−CombineSink.prototype.event = function event (t) {
832+ for (var i = 0; i < l; ++i) {
639− var this$1 = this;
833+ this$1.hasValue[i] = false;
834+ }
835+
836+ this.activeCount = sinks.length;
837+}
640
838
641− return function (ref) {
839+CombineSink.prototype.event = function (t) {
642− var index = ref.index;
840+ var this$1 = this;
643− var value = ref.value;
644
841
645− var f = this$1.f;
842+ return function (indexedValue) {
843+ var i = indexedValue.index;
646− var awaiting = this$1._updateReady(index);
844+ var awaiting = this$1._updateReady(i);
647− this$1.values[index] = value;
648
845
846+ this$1.values[i] = indexedValue.value;
649 if (awaiting === 0) {
847 if (awaiting === 0) {
650− var result = f(this$1.values[0])(this$1.values[1]);
848+ var value = tail(this$1.values).reduce(function (f, x) { return f(x); }, this$1.f(this$1.values[0]));
651− this$1.sink.event(t)(result);
849+ this$1.sink.event(t)(value);
652 }
850 }
653 }
851 }
654};
852};
655
853
656−CombineSink.prototype._updateReady = function _updateReady (index) {
854+CombineSink.prototype._updateReady = function (index) {
657 if (this.awaiting > 0) {
855 if (this.awaiting > 0) {
658 if (!this.hasValue[index]) {
856 if (!this.hasValue[index]) {
659 this.hasValue[index] = true;
857 this.hasValue[index] = true;
660 this.awaiting -= 1;
858 this.awaiting -= 1;
661 }
859 }
662 }
860 }
663−
664 return this.awaiting
861 return this.awaiting
665};
862};
666
863
667−CombineSink.prototype.end = function end (t, ref) {
864+CombineSink.prototype.end = function (t, indexedValue) {
668− var index = ref.index;
865+ this.disposables[indexedValue.index].dispose();
669− var value = ref.value;
866+
867+ if (--this.activeCount === 0) {
868+ this.sink.end(t);
869+ }
870+};
871+
872+/** @license MIT License (c) copyright 2010-2016 original author or authors */
873+/** @author Brian Cavalier */
874+/** @author John Hann */
670
875
876+/**
877+ * @param {Array} streams array of stream to merge
878+ * @returns {Stream} stream containing events from all input observables
879+ * in time order. If two events are simultaneous they will be merged in
880+ * arbitrary order.
881+ */
882+function merge (streams) {
883+ return new Stream(mergeSources(streams))
884+}
885+
886+/**
887+ * This implements fusion/flattening for merge. It will
888+ * fuse adjacent merge operations. For example:
889+ * - a.merge(b).merge(c) effectively becomes merge(a, b, c)
890+ * - merge(a, merge(b, c)) effectively becomes merge(a, b, c)
891+ * It does this by concatenating the sources arrays of
892+ * any nested Merge sources, in effect "flattening" nested
893+ * merge operations into a single merge.
894+ */
895+function mergeSources (streams) {
896+ return new Merge(reduce(appendSources, [], streams))
897+}
898+
899+function appendSources (sources, stream) {
900+ var source = stream.source;
901+ return source instanceof Merge
902+ ? sources.concat(source.sources)
903+ : sources.concat(source)
904+}
905+
906+function Merge (sources) {
907+ this.sources = sources;
908+}
909+
910+Merge.prototype.run = function (sink, scheduler) {
911+ var this$1 = this;
912+
913+ var l = this.sources.length;
671− this.disposables[index].dispose();
914+ var disposables = new Array(l);
915+ var sinks = new Array(l);
916+
917+ var mergeSink = new MergeSink(disposables, sinks, sink);
918+
919+ for (var indexSink, i = 0; i < l; ++i) {
920+ indexSink = sinks[i] = new IndexSink(i, mergeSink);
921+ disposables[i] = this$1.sources[i].run(indexSink, scheduler);
922+ }
923+
924+ return disposeAll(disposables)
925+};
926+
927+function MergeSink (disposables, sinks, sink) {
928+ this.sink = sink;
929+ this.disposables = disposables;
930+ this.activeCount = sinks.length;
931+}
932+
933+MergeSink.prototype.event = function (t, indexValue) {
934+ this.sink.event(t, indexValue.value);
935+};
936+
937+MergeSink.prototype.end = function (t, indexedValue) {
938+ dispose.tryDispose(t, this.disposables[indexedValue.index], this.sink);
939+ if (--this.activeCount === 0) {
940+ this.sink.end(t, indexedValue.value);
941+ }
942+};
943+
944+var concat = function (s1) { return function (s2) { return continueWith(function () { return s2; })(s1); }; };
945+
946+var continueWith = function (f) { return function (stream) { return new Stream(new ContinueWith(f, stream.source)); }; };
947+
948+function ContinueWith (f, source) {
949+ this.f = f;
950+ this.source = source;
951+}
952+
953+ContinueWith.prototype.run = function (sink) {
954+ var this$1 = this;
955+
956+ return function (scheduler) {
957+ return new ContinueWithSink(this$1.f, this$1.source, sink, scheduler)
958+ }
959+};
960+
961+function ContinueWithSink (f, source, sink, scheduler) {
962+ this.f = f;
963+ this.sink = sink;
964+ this.scheduler = scheduler;
965+ this.active = true;
966+ this.disposable = new MemoizedDisposable(source.run(this)(scheduler));
967+}
968+
969+ContinueWithSink.prototype.event = function (t) {
970+ var this$1 = this;
971+
972+ return function (x) {
973+ if (!this$1.active) {
974+ return
975+ }
976+ this$1.sink.event(t)(x);
977+ }
978+};
979+
980+ContinueWithSink.prototype.end = function (t, x) {
981+ if (!this.active) {
982+ return
983+ }
984+
985+ this.disposable.dispose();
986+ this._startNext(t, x, this.sink);
987+};
988+
989+ContinueWithSink.prototype._startNext = function (t, x, sink) {
990+ this._continue(this.f, x, sink);
991+};
992+
993+ContinueWithSink.prototype._continue = function (f, x, sink) {
994+ return f(x).source.run(sink)(this.scheduler)
995+};
996+
997+ContinueWithSink.prototype.dispose = function () {
998+ this.active = false;
999+ return this.disposable.dispose()
1000+};
1001+
1002+/**
1003+ * Given a stream of streams, return a new stream that adopts the behavior
1004+ * of the most recent inner stream.
1005+ * @param {Stream} stream of streams on which to switch
1006+ * @returns {Stream} switching stream
1007+ */
1008+function switchLatest (stream) {
1009+ return new Stream(new Switch(stream.source))
1010+}
1011+
1012+function Switch (source) {
1013+ this.source = source;
1014+}
1015+
1016+Switch.prototype.run = function (sink) {
1017+ var this$1 = this;
1018+
1019+ return function (scheduler) {
1020+ var switchSink = new SwitchSink(sink, scheduler);
1021+ return disposeAll([switchSink, this$1.source.run(switchSink)(scheduler)])
1022+ }
1023+};
1024+
1025+function SwitchSink (sink, scheduler) {
1026+ this.sink = sink;
1027+ this.scheduler = scheduler;
1028+ this.current = null;
1029+ this.ended = false;
1030+}
1031+
1032+SwitchSink.prototype.event = function (t) {
1033+ var this$1 = this;
1034+
1035+ return function (stream) {
1036+ this$1._disposeCurrent(t); // TODO: capture the result of this dispose
1037+ this$1.current = new Segment(t, Infinity, this$1, this$1.sink);
1038+ this$1.current.disposable = stream.source.run(this$1.current)(this$1.scheduler);
1039+ }
1040+};
1041+
1042+SwitchSink.prototype.end = function (t, x) {
1043+ this.ended = true;
1044+ this._checkEnd(t, x);
1045+};
1046+
1047+SwitchSink.prototype.dispose = function () {
1048+ return this._disposeCurrent(this.scheduler.now())
1049+};
1050+
1051+SwitchSink.prototype._disposeCurrent = function (t) {
1052+ if (this.current !== null) {
1053+ return this.current._dispose(t)
1054+ }
1055+};
1056+
1057+SwitchSink.prototype._disposeInner = function (t, inner) {
1058+ inner._dispose(t); // TODO: capture the result of this dispose
1059+ if (inner === this.current) {
1060+ this.current = null;
1061+ }
1062+};
1063+
1064+SwitchSink.prototype._checkEnd = function (t, x) {
1065+ if (this.ended && this.current === null) {
1066+ this.sink.end(t, x);
1067+ }
1068+};
1069+
1070+SwitchSink.prototype._endInner = function (t, x, inner) {
1071+ this._disposeInner(t, inner);
1072+ this._checkEnd(t, x);
1073+};
1074+
1075+function Segment (min, max, outer, sink) {
1076+ this.min = min;
1077+ this.max = max;
1078+ this.outer = outer;
1079+ this.sink = sink;
1080+ this.disposable = emptyDisposable;
1081+}
1082+
1083+Segment.prototype.event = function (t) {
1084+ var this$1 = this;
1085+
1086+ return function (x) {
1087+ if (t < this$1.max) {
1088+ this$1.sink.event(Math.max(t, this$1.min))(x);
1089+ }
1090+ }
1091+};
1092+
1093+Segment.prototype.end = function (t, x) {
1094+ this.outer._endInner(Math.max(t, this.min), x, this);
1095+};
672
1096
673− if (--this.activeCount === 0) { this.sink.end(t); }
1097+Segment.prototype._dispose = function (t) {
1098+ this.max = t;
1099+ this.disposable.dispose();
674};
1100};
675
1101
676/* */
1102/* */
@@ -682,7 +1108,16 @@ exports.defaultScheduler = defaultScheduler;
682exports.scheduleTasks = scheduleTasks;
1108exports.scheduleTasks = scheduleTasks;
683exports.endTask = endTask;
1109exports.endTask = endTask;
684exports.eventTask = eventTask;
1110exports.eventTask = eventTask;
685−exports.tapEnd = tapEnd;
686−exports.tapEvent = tapEvent;
687exports.drain = drain;
1111exports.drain = drain;
688−exports.combine = combine;
1112+exports.flatMap = flatMap;
1113+exports.mergeConcurrently = mergeConcurrently;
1114+exports.join = join;
1115+exports.mergeMapConcurrently = mergeMapConcurrently;
1116+exports.combine2 = combine2;
1117+exports.combine3 = combine3;
1118+exports.combine4 = combine4;
1119+exports.combine5 = combine5;
1120+exports.merge = merge;
1121+exports.concat = concat;
1122+exports.continueWith = continueWith;
1123+exports.switch = switchLatest;
src/Control/Stream.purs+179−17
@@ -10,9 +10,9 @@ module Control.Stream
10 , ScheduledTask
10 , ScheduledTask
11 , Disposable
11 , Disposable
12 , Time
12 , Time
13− , createStream
13+ , EventFn
14− , getSource
14+ , EndFn
15− , runSource
15+ -- Scheduler-related functions and values
16 , defaultScheduler
16 , defaultScheduler
17 , eventTask
17 , eventTask
18 , endTask
18 , endTask
@@ -20,19 +20,54 @@ module Control.Stream
20 , emptyDisposable
20 , emptyDisposable
21 , disposeAll
21 , disposeAll
22 , scheduleTasks
22 , scheduleTasks
23+ -- Stream-related helpers
24+ , createStream
25+ , createSink
26+ , createCombinator
27+ , createEventCombinator
28+ , getSource
29+ , runSource
30+ -- Combinators
23 , drain
31 , drain
24 , tapEvent
32 , tapEvent
25 , tapEnd
33 , tapEnd
34+ , filter
35+ , mergeConcurrently
36+ , mergeMapConcurrently
37+ , join
38+ , combine2
39+ , combine3
40+ , combine4
41+ , combine5
42+ , continueWith
43+ , concat
44+ , merge
45+ , switch
46+ , constant
47+ , scan
48+ -- Stream factories
26 , just
49 , just
27 , fromArray
50 , fromArray
51+ , empty
52+ , periodic
28 )
53 )
29 where
54 where
30
55
56+import Control.Applicative (class Applicative)
57+import Control.Apply (class Apply)
58+import Control.Bind (class Bind)
59+import Control.Category (id)
60+import Control.Monad (class Monad)
31import Control.Monad.Eff (Eff, Pure)
61import Control.Monad.Eff (Eff, Pure)
62+import Control.Monad.Eff.Ref (Ref, modifyRef', newRef, readRef)
63+import Control.Monad.Eff.Unsafe (unsafeCoerceEff, unsafePerformEff)
64+import Control.MonadPlus (class Alt, class Plus)
32import Data.Array (snoc)
65import Data.Array (snoc)
66+import Data.Function (flip, ($))
33import Data.Functor (class Functor, map)
67import Data.Functor (class Functor, map)
34−import Data.Unit (Unit)
68+import Data.Monoid (class Monoid)
35−import Prelude (flip)
69+import Data.Semigroup (class Semigroup)
70+import Data.Unit (Unit, unit)
36
71
37newtype Time = Time Int
72newtype Time = Time Int
38
73
@@ -50,11 +85,11 @@ type Sink a =
50 , end :: Time -> Unit
85 , end :: Time -> Unit
51 }
86 }
52
87
53−type Scheduler =
88+newtype Scheduler = Scheduler
54 { now :: Pure Time
89 { now :: Pure Time
55 , asap :: Task -> ScheduledTask
90 , asap :: Task -> ScheduledTask
56 , delay :: Number -> Task -> ScheduledTask
91 , delay :: Number -> Task -> ScheduledTask
57− , period :: Number -> Task -> ScheduledTask
92+ , periodic :: Number -> Task -> ScheduledTask
58 , schedule :: Number -> Number -> Task -> ScheduledTask
93 , schedule :: Number -> Number -> Task -> ScheduledTask
59 , cancel :: ScheduledTask -> Unit
94 , cancel :: ScheduledTask -> Unit
60 , cancelAll :: (ScheduledTask -> Boolean) -> Unit
95 , cancelAll :: (ScheduledTask -> Boolean) -> Unit
@@ -72,9 +107,36 @@ type ScheduledTask =
72
107
73type Disposable = { dispose :: Pure Unit }
108type Disposable = { dispose :: Pure Unit }
74
109
110+type EventFn a b = (Sink b -> Scheduler -> Time -> a -> Unit)
111+type EndFn a = (Sink a -> Scheduler -> Time -> Unit)
112+
75instance functorStream :: Functor Stream where
113instance functorStream :: Functor Stream where
76 map = _map
114 map = _map
77
115
116+instance applyStream :: Apply Stream where
117+ apply = combine2 id
118+
119+instance applicativeStream :: Applicative Stream where
120+ pure = just
121+
122+instance bindStream :: Bind Stream where
123+ bind = flatMap
124+
125+instance monadStream :: Monad Stream
126+
127+instance semigroupStream :: Semigroup (Stream a) where
128+ append = concat
129+
130+instance altStream :: Alt Stream where
131+ alt = \s1 s2 -> merge [s1, s2]
132+
133+instance monoidStream :: Monoid (Stream a) where
134+ mempty = empty
135+
136+instance plusStream :: Plus Stream where
137+ empty = empty
138+
139+-- Stream-related helpers
78createStream :: forall a. (Sink a -> Scheduler -> Disposable) -> Stream a
140createStream :: forall a. (Sink a -> Scheduler -> Disposable) -> Stream a
79createStream run = Stream { source: { run } }
141createStream run = Stream { source: { run } }
80
142
@@ -84,6 +146,30 @@ getSource stream = case stream of Stream a -> a.source
84runSource :: forall a. Stream a -> Sink a -> Scheduler -> Disposable
146runSource :: forall a. Stream a -> Sink a -> Scheduler -> Disposable
85runSource stream sink scheduler = (getSource stream).run sink scheduler
147runSource stream sink scheduler = (getSource stream).run sink scheduler
86
148
149+createSink :: forall a. (Time -> a -> Unit) -> (Time -> Unit) -> Sink a
150+createSink event end = { event, end }
151+
152+createCombinator :: forall a b. EventFn a b -> EndFn b -> Stream a -> Stream b
153+createCombinator event end stream = createStream runCombinator
154+ where
155+ runCombinator :: Sink b -> Scheduler -> Disposable
156+ runCombinator sink scheduler =
157+ runSource stream (createSink (event sink scheduler) (end sink scheduler)) scheduler
158+
159+createEventCombinator :: forall a b. EventFn a b -> Stream a -> Stream b
160+createEventCombinator event stream = createCombinator event end stream
161+ where
162+ end :: EndFn b
163+ end sink scheduler time =
164+ sink.end time
165+
166+createEndCombinator :: forall a. EndFn a -> Stream a -> Stream a
167+createEndCombinator end stream = createCombinator event end stream
168+ where
169+ event :: EventFn a a
170+ event sink scheduler time value = sink.event time value
171+
172+-- Stream factories
87just :: forall a. a -> Stream a
173just :: forall a. a -> Stream a
88just a = createStream (runJust a)
174just a = createStream (runJust a)
89
175
@@ -91,6 +177,12 @@ runJust :: forall a. a -> Sink a -> Scheduler -> Disposable
91runJust a sink scheduler =
177runJust a sink scheduler =
92 scheduleTasks [ eventTask a sink, endTask sink ] scheduler
178 scheduleTasks [ eventTask a sink, endTask sink ] scheduler
93
179
180+empty :: forall a. Stream a
181+empty = createStream \sink scheduler -> scheduleTasks [endTask sink] scheduler
182+
183+never :: forall a. Stream a
184+never = createStream \sink scheduler -> emptyDisposable
185+
94-- TODO: remove fromArray because it's not a real event stream
186-- TODO: remove fromArray because it's not a real event stream
95-- replace with a fromArray combinator
187-- replace with a fromArray combinator
96fromArray :: forall a. Array a -> Stream a
188fromArray :: forall a. Array a -> Stream a
@@ -105,19 +197,78 @@ runFromArray arr sink scheduler = scheduleTasks tasks scheduler
105 eventTasks :: Array Task
197 eventTasks :: Array Task
106 eventTasks = map (flip eventTask sink) arr
198 eventTasks = map (flip eventTask sink) arr
107
199
200+periodic :: Number -> Stream Unit
201+periodic period = createStream (runPeriodic period)
202+
203+runPeriodic :: Number -> Sink Unit -> Scheduler -> Disposable
204+runPeriodic period sink scheduler = { dispose: scheduledTask.dispose }
205+ where
206+ p = case scheduler of Scheduler a -> a.periodic
207+ scheduledTask = p period (eventTask unit sink)
208+
209+-- combinators
108_map :: forall a b. (a -> b) -> Stream a -> Stream b
210_map :: forall a b. (a -> b) -> Stream a -> Stream b
109−_map f stream = createStream runMap
211+_map f stream = createEventCombinator mapEvent stream
212+ where
213+ mapEvent :: EventFn a b
214+ mapEvent sink scheduler time value = sink.event time (f value)
215+
216+filter :: forall a. (a -> Boolean) -> Stream a -> Stream a
217+filter predicate stream = createEventCombinator event stream
218+ where
219+ event :: EventFn a a
220+ event sink scheduler time value = if predicate value
221+ then sink.event time value
222+ else unit
223+
224+tapEvent :: forall e a. (a -> EffStream e Unit) -> Stream a -> Stream a
225+tapEvent f stream = createEventCombinator event stream
110 where
226 where
111− runMap :: Sink b -> Scheduler -> Disposable
227+ event :: EventFn a a
112− runMap sink scheduler =
228+ event sink scheduler time value = sink.event time value
113− runSource stream (mapSink f sink) scheduler
229+ where -- find a better way to perform these side effects
230+ x = unsafePerformEff (f value)
114
231
115−mapSink :: forall a b. (a -> b) -> Sink b -> Sink a
232+tapEnd :: forall e a. EffStream e Unit -> Stream a -> Stream a
116−mapSink f sink = { event: mapEvent, end: sink.end }
233+tapEnd f stream = createEndCombinator end stream
117 where
234 where
118− mapEvent :: Time -> a -> Unit
235+ end :: EndFn a
236+ end sink scheduler time = sink.end time
237+ where -- find a better way to perform these side effects
238+ x = unsafePerformEff f
239+
240+startWith :: forall a. a -> Stream a -> Stream a
241+startWith value stream = concat (just value) stream
242+
243+constant :: forall a b. b -> Stream a -> Stream b
244+constant value stream = createEventCombinator constantEvent stream
245+ where
246+ constantEvent sink scheduler time x = sink.event time value
247+
248+scan :: forall a b. (b -> a -> b) -> b -> Stream a -> Stream b
249+scan f seed stream = startWith seed $ createEventCombinator scanEvent stream
250+ where
251+ state = createState seed
252+
253+ scanEvent :: Sink b -> Scheduler -> Time -> a -> Unit
119− mapEvent time value = sink.event time (f value)
254+ scanEvent sink scheduler time value = sink.event time (state.set \acc -> f acc value)
255+
256+-- find a better way to perform these side effects
257+createState :: forall a. a -> { get :: Pure a, set :: (a -> a) -> a }
258+createState seed = { set, get }
259+ where
260+ ref :: Ref a
261+ ref = unsafePerformEff $ unsafeCoerceEff $ newRef seed
262+
263+ set :: (a -> a) -> a
264+ set f = unsafePerformEff $ unsafeCoerceEff $
265+ modifyRef' ref \x -> { state: (f x), value: (f x) }
266+
267+ get :: Pure a
268+ get = unsafeCoerceEff $ readRef ref
120
269
270+-- Foreign imports
271+-- Scheduler-related
121foreign import defaultScheduler :: Scheduler
272foreign import defaultScheduler :: Scheduler
122foreign import eventTask :: forall a. a -> Sink a -> Task
273foreign import eventTask :: forall a. a -> Sink a -> Task
123foreign import endTask :: forall a. Sink a -> Task
274foreign import endTask :: forall a. Sink a -> Task
@@ -125,6 +276,17 @@ foreign import createDisposable :: forall a. (a -> Unit) -> a -> Disposable
125foreign import emptyDisposable :: Disposable
276foreign import emptyDisposable :: Disposable
126foreign import disposeAll :: Array Disposable -> Disposable
277foreign import disposeAll :: Array Disposable -> Disposable
127foreign import scheduleTasks :: Array Task -> Scheduler -> Disposable
278foreign import scheduleTasks :: Array Task -> Scheduler -> Disposable
279+-- combinators
128foreign import drain :: forall e a. Stream a -> EffStream e Unit
280foreign import drain :: forall e a. Stream a -> EffStream e Unit
129−foreign import tapEvent :: forall e a. (a -> EffStream e Unit) -> Stream a -> Stream a
281+foreign import mergeConcurrently :: forall a. Int -> Stream (Stream a) -> Stream a
130−foreign import tapEnd :: forall e a. EffStream e Unit -> Stream a -> Stream a
282+foreign import mergeMapConcurrently :: forall a b. (a -> Stream b) -> Stream a -> Stream b
283+foreign import combine2 :: forall a b c. (a -> b -> c) -> Stream a -> Stream b -> Stream c
284+foreign import combine3 :: forall a b c d. (a -> b -> c -> d) -> Stream a -> Stream b -> Stream c -> Stream d
285+foreign import combine4 :: forall a b c d e. (a -> b -> c -> d -> e) -> Stream a -> Stream b -> Stream c -> Stream d -> Stream e
286+foreign import combine5 :: forall a b c d e f. (a -> b -> c -> d -> e -> f) -> Stream a -> Stream b -> Stream c -> Stream d -> Stream e -> Stream f
287+foreign import join :: forall a. Stream (Stream a) -> Stream a
288+foreign import flatMap :: forall a b. Stream a -> (a -> Stream b) -> Stream b
289+foreign import continueWith :: forall a. (Unit -> Stream a) -> Stream a -> Stream a
290+foreign import concat :: forall a. Stream a -> Stream a -> Stream a
291+foreign import merge :: forall a. Array (Stream a) -> Stream a
292+foreign import switch :: forall a. Stream (Stream a) -> Stream a