git.iko.soy/purescript-stream
on master help sign in
purescript-stream/src/Control/Stream/Stream.js
1244 lines 27 kB history raw
1var ControlStreamScheduler = require('../Control.Stream.Scheduler')
2var ControlStreamStream = require('../Control.Stream.Stream')
4exports.drain = function (stream) { return function () { return runSource(stream, ControlStreamScheduler.scheduler) } }
6exports.subscribe = function (event) {
7 return function (end) {
8 return function (stream) {
9 return function () {
10 var disposable = new SettableDisposable()
12 var observer = { event: event, end: end }
14 var d = stream.run(observer)(ControlStreamScheduler.scheduler)
16 disposable.setDisposable(d)
18 return {}
19 }
20 }
21 }
24function runSource (source, scheduler) {
25 var disposable = new SettableDisposable()
27 function event (t) {
28 return function (x) {
29 return {}
30 }
31 }
33 function end (t) {
34 disposable.dispose()
35 return {}
36 }
38 var observer = { event: event, end: end }
40 var d = source.run(observer)(scheduler)
42 disposable.setDisposable(d)
44 return {}
47var SettableDisposable = function SettableDisposable () {
48 this.disposable = void 0
49 this.disposed = false
50 this.result = void 0
53SettableDisposable.prototype.setDisposable = function setDisposable (disposable) {
54 this.disposable = disposable
56 if (this.disposed) {
57 this.result = disposable.dispose()
58 }
61SettableDisposable.prototype.dispose = function dispose () {
62 if (this.disposed) {
63 return this.result
64 }
66 this.disposed = true
68 if (this.disposable !== void 0) {
69 this.result = this.disposable.dispose()
70 }
72 return this.result
75exports.fromArray = function (array) {
76 return function (stream) {
77 return new FromArray(array, stream)
78 }
81function FromArray (array, stream) {
82 this.array = array
83 this.stream = stream
86FromArray.prototype.run = function (sink) {
87 var that = this
89 return function (scheduler) {
90 return that.stream.run(new FromArraySink(sink, that.array))(scheduler)
91 }
94function FromArraySink (sink, array) {
95 this.sink = sink
96 this.array = array
99FromArraySink.prototype.event = function (time) {
100 var array = this.array
101 var sink = this.sink
102 return function (x) {
103 var value = array.shift()
105 if (value) {
106 sink.event(time)(value)
107 } else {
108 sink.end(time)
109 }
110 }
113FromArraySink.prototype.end = function (time) {
114 this.sink.end(time)
117var LinkedList = function LinkedList () {
118 this.head = null
119 this.length = 0
122LinkedList.prototype.add = function add (x) {
123 if (this.head !== null) {
124 this.head.prev = x
125 x.next = this.head
126 }
127 this.head = x
128 ++this.length
131LinkedList.prototype.remove = function remove (x) { // eslint-disable-line complexity
132 --this.length
133 if (x === this.head) {
134 this.head = this.head.next
135 }
136 if (x.next !== null) {
137 x.next.prev = x.prev
138 x.next = null
139 }
140 if (x.prev !== null) {
141 x.prev.next = x.next
142 x.prev = null
143 }
146LinkedList.prototype.isEmpty = function isEmpty () {
147 return this.length === 0
150LinkedList.prototype.dispose = function dispose () {
151 if (this.isEmpty()) { return }
153 var x = this.head
154 this.head = null
155 this.length = 0
157 while (x !== null) {
158 x.dispose()
159 x = x.next
160 }
163exports.chain = function (f) {
164 return function (stream) {
165 return mergeMapConcurrently(f)(Infinity)(stream)
166 }
169exports.mergeConcurrently = mergeConcurrently
171function mergeConcurrently (concurrency) {
172 return function (stream) {
173 return mergeMapConcurrently(function (x) { return x })(concurrency)(stream)
174 }
177exports.join = mergeConcurrently(Infinity)
179exports.mergeMapConcurrently = mergeMapConcurrently
181function mergeMapConcurrently (f) {
182 return function (concurrency) {
183 return function (stream) {
184 var mstream = new MergeConcurrently(f, concurrency, stream)
186 mstream.run.bind(mstream)
188 return mstream
189 }
190 }
193function MergeConcurrently (f, concurrency, source) {
194 this.f = f
195 this.concurrency = concurrency
196 this.source = source
199MergeConcurrently.prototype.run = function run (sink) {
200 var this$1 = this
202 return function (scheduler) { return new Outer(this$1.f, this$1.concurrency, this$1.source, sink, scheduler) }
205var Outer = function Outer (f, concurrency, source, sink, scheduler) {
206 this.f = f
207 this.concurrency = concurrency
208 this.sink = sink
209 this.scheduler = scheduler
210 this.pending = []
211 this.current = new LinkedList()
213 var disposed = false
214 var disposable = source.run(this)(scheduler)
216 this.disposable = {
217 dispose: function () {
218 if (disposed) return
220 disposed = true
221 disposable.dispose()
222 }
223 }
225 this.active = true
228Outer.prototype.event = function event (t) {
229 var this$1 = this
231 return function (x) { return this$1._addInner(t, x) }
234Outer.prototype._addInner = function _addInner (t, x) {
235 if (this.current.length < this.concurrency) {
236 this._startInner(t, x)
237 } else {
238 this.pending.push(x)
239 }
242Outer.prototype._startInner = function _startInner (t, x) {
243 this._initInner(t, x)
246Outer.prototype._initInner = function _initInner (t, x) {
247 var innerSink = new Inner(t, this, this.sink)
248 innerSink.disposable = mapAndRun(this.f, x, innerSink, this.scheduler)
249 this.current.add(innerSink)
252Outer.prototype.end = function end (t) {
253 this.active = false
254 this.disposable.dispose()
255 this._checkEnd(t)
258Outer.prototype.dispose = function dispose () {
259 this.active = false
260 this.pending.length = 0
261 this.disposable.dispose()
262 this.current.dispose()
265Outer.prototype._endInner = function _endInner (t, inner) {
266 this.current.remove(inner)
267 inner.dispose()
269 if (this.pending.length === 0) {
270 this._checkEnd(t)
271 } else {
272 this._startInner(t, this.pending.shift())
273 }
276Outer.prototype._checkEnd = function _checkEnd (t) {
277 if (!this.active && this.current.isEmpty()) {
278 this.sink.end(t)
279 }
282var mapAndRun = function (f, x, sink, scheduler) { return f(x).run(sink)(scheduler) }
284var Inner = function Inner (time, outer, sink) {
285 this.prev = this.next = null
286 this.time = time
287 this.outer = outer
288 this.sink = sink
289 this.disposable = void 0
292Inner.prototype.event = function event (t) {
293 var this$1 = this
295 return function (x) { return this$1.sink.event(Math.max(t, this$1.time))(x) }
298Inner.prototype.end = function end (t) {
299 this.outer._endInner(Math.max(t, this.time), this)
302Inner.prototype.dispose = function dispose () {
303 return this.disposable.dispose()
306exports.switch = function switchLatest (stream) {
307 return new Switch(stream)
310function Switch (source) {
311 this.source = source
314Switch.prototype.run = function (sink) {
315 var this$1 = this
317 return function (scheduler) {
318 var switchSink = new SwitchSink(sink, scheduler)
319 var disposable = this$1.source.run(switchSink)(scheduler)
321 return {
322 dispose: function () {
323 switchSink.dispose()
324 disposable.dispose()
325 }
326 }
327 }
330function SwitchSink (sink, scheduler) {
331 this.sink = sink
332 this.scheduler = scheduler
333 this.current = null
334 this.ended = false
337SwitchSink.prototype.event = function (t) {
338 var this$1 = this
340 return function (stream) {
341 this$1._disposeCurrent(t) // TODO: capture the result of this dispose
342 this$1.current = new Segment(t, Infinity, this$1, this$1.sink)
343 this$1.current.disposable = stream.run(this$1.current)(this$1.scheduler)
344 }
347SwitchSink.prototype.end = function (t, x) {
348 this.ended = true
349 this._checkEnd(t, x)
352SwitchSink.prototype.dispose = function () {
353 return this._disposeCurrent(this.scheduler.now())
356SwitchSink.prototype._disposeCurrent = function (t) {
357 if (this.current !== null) {
358 return this.current._dispose(t)
359 }
362SwitchSink.prototype._disposeInner = function (t, inner) {
363 inner._dispose(t) // TODO: capture the result of this dispose
364 if (inner === this.current) {
365 this.current = null
366 }
369SwitchSink.prototype._checkEnd = function (t, x) {
370 if (this.ended && this.current === null) {
371 this.sink.end(t, x)
372 }
375SwitchSink.prototype._endInner = function (t, x, inner) {
376 this._disposeInner(t, inner)
377 this._checkEnd(t, x)
380function Segment (min, max, outer, sink) {
381 this.min = min
382 this.max = max
383 this.outer = outer
384 this.sink = sink
385 this.disposable = { dispose: Function.prototype }
388Segment.prototype.event = function (t) {
389 var this$1 = this
391 return function (x) {
392 if (t < this$1.max) {
393 this$1.sink.event(Math.max(t, this$1.min))(x)
394 }
395 }
398Segment.prototype.end = function (t, x) {
399 this.outer._endInner(Math.max(t, this.min), x, this)
402Segment.prototype._dispose = function (t) {
403 this.max = t
404 this.disposable.dispose()
407exports.continueWith = function continueWith (f) {
408 return function (stream) {
409 return new ContinueWith(f, stream)
410 }
413function ContinueWith (f, source) {
414 this.f = f
415 this.source = source
418ContinueWith.prototype.run = function (sink) {
419 var that = this
420 return function (scheduler) {
421 return new ContinueWithSink(that.f, that.source, sink, scheduler)
422 }
425function ContinueWithSink (f, source, sink, scheduler) {
426 this.f = f
427 this.sink = sink
428 this.scheduler = scheduler
429 this.active = true
430 this.disposable = source.run(this)(scheduler)
433ContinueWithSink.prototype.event = function (t) {
434 var that = this
436 return function (x) {
437 if (!that.active) {
438 return
439 }
440 that.sink.event(t)(x)
441 }
444ContinueWithSink.prototype.end = function (t, x) {
445 if (!this.active) {
446 return
447 }
449 this.disposable.dispose()
451 this._startNext(t, x, this.sink)
454ContinueWithSink.prototype._startNext = function (t, x, sink) {
455 try {
456 this.disposable = this._continue(this.f, x, sink)
457 } catch (e) {
458 sink.error(t, e)
459 }
462ContinueWithSink.prototype._continue = function (f, x, sink) {
463 return f(x).run(sink)(this.scheduler)
466ContinueWithSink.prototype.dispose = function () {
467 this.active = false
468 return this.disposable.dispose()
471exports.combine2 = function (f) { return function (s1) { return function (s2) { return combineArray(f, [s1, s2]) } } }
472exports.combine3 = function (f) { return function (s1) { return function (s2) { return function (s3) { return combineArray(f, [s1, s2, s3]) } } } }
473exports.combine4 = function (f) { return function (s1) { return function (s2) { return function (s3) { return function (s4) { return combineArray(f, [s1, s2, s3, s4]) } } } } }
474exports.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]) } } } } } }
476exports.combineArray = function (streams) {
477 return combineArray(Array, streams)
480function combineArray (f, streams) {
481 return new Combine(f, streams)
484function Combine (f, sources) {
485 this.f = f
486 this.sources = sources
489Combine.prototype.run = function (sink) {
490 var this$1 = this
492 return function (scheduler) {
493 var l = this$1.sources.length
494 var disposables = new Array(l)
495 var sinks = new Array(l)
497 var mergeSink = new CombineSink(disposables, sinks, sink, this$1.f)
499 for (var indexSink, i = 0; i < l; ++i) {
500 indexSink = sinks[i] = new IndexSink(i, mergeSink)
501 disposables[i] = this$1.sources[i].run(indexSink)(scheduler)
502 }
504 return {
505 dispose: function () {
506 disposables.forEach(function (d) {
507 d.dispose()
508 })
509 }
510 }
511 }
514function CombineSink (disposables, sinks, sink, f) {
515 var this$1 = this
517 this.sink = sink
518 this.disposables = disposables
519 this.sinks = sinks
520 this.f = f
522 var l = sinks.length
523 this.awaiting = l
524 this.values = new Array(l)
525 this.hasValue = new Array(l)
527 for (var i = 0; i < l; ++i) {
528 this$1.hasValue[i] = false
529 }
531 this.activeCount = sinks.length
534CombineSink.prototype.event = function (t) {
535 var this$1 = this
537 return function (indexedValue) {
538 var i = indexedValue.index
539 var awaiting = this$1._updateReady(i)
541 this$1.values[i] = indexedValue.value
542 if (awaiting === 0) {
543 var value = tail(this$1.values).reduce(function (f, x) { return f(x) }, this$1.f(this$1.values[0]))
544 this$1.sink.event(t)(value)
545 }
546 }
549CombineSink.prototype._updateReady = function (index) {
550 if (this.awaiting > 0) {
551 if (!this.hasValue[index]) {
552 this.hasValue[index] = true
553 this.awaiting -= 1
554 }
555 }
556 return this.awaiting
559CombineSink.prototype.end = function (t, indexedValue) {
560 this.disposables[indexedValue.index].dispose()
562 if (--this.activeCount === 0) {
563 this.sink.end(t)
564 }
567var IndexSink = function IndexSink (i, sink) {
568 this.sink = sink
569 this.index = i
570 this.active = true
571 this.value = void 0
574IndexSink.prototype.event = function event (t) {
575 var this$1 = this
577 return function (x) {
578 if (!this$1.active) { return }
580 this$1.value = x
581 this$1.sink.event(t)(this$1)
582 }
585IndexSink.prototype.end = function end (t) {
586 if (!this.active) { return }
588 this.active = false
589 this.sink.end(t, { index: this.index })
592function drop (n, a) { // eslint-disable-line complexity
593 if (n < 0) {
594 throw new TypeError('n must be >= 0')
595 }
597 var l = a.length
598 if (n === 0 || l === 0) {
599 return a
600 }
602 if (n >= l) {
603 return []
604 }
606 return unsafeDrop(n, a, l - n)
609// unsafeDrop :: Int -> [a] -> Int -> [a]
610// Internal helper for drop
611function unsafeDrop (n, a, l) {
612 var b = new Array(l)
613 for (var i = 0; i < l; ++i) {
614 b[i] = a[n + i]
615 }
616 return b
619// tail :: [a] -> [a]
620// drop head element
621function tail (a) {
622 return drop(1, a)
625exports.merge = function merge (streams) {
626 return mergeSources(streams)
629/**
630 * This implements fusion/flattening for merge. It will
631 * fuse adjacent merge operations. For example:
632 * - a.merge(b).merge(c) effectively becomes merge(a, b, c)
633 * - merge(a, merge(b, c)) effectively becomes merge(a, b, c)
634 * It does this by concatenating the sources arrays of
635 * any nested Merge sources, in effect "flattening" nested
636 * merge operations into a single merge.
637 */
638function mergeSources (streams) {
639 return new Merge(reduce(appendSources, [], streams))
642function appendSources (sources, stream) {
643 return stream instanceof Merge
644 ? sources.concat(stream.sources)
645 : sources.concat(stream)
648function Merge (sources) {
649 this.sources = sources
652Merge.prototype.run = function (sink) {
653 var this$1 = this
655 return function (scheduler) {
656 var l = this.sources.length
657 var disposables = new Array(l)
658 var sinks = new Array(l)
660 var mergeSink = new MergeSink(disposables, sinks, sink)
662 for (var indexSink, i = 0; i < l; ++i) {
663 indexSink = sinks[i] = new IndexSink(i, mergeSink)
664 disposables[i] = this$1.sources[i].run(indexSink, scheduler)
665 }
667 return all(disposables)
668 }
671function all (disposables) {
672 return {
673 dispose: function () {
674 disposables.forEach(function (d) {
675 d.dispose()
676 })
677 }
678 }
681function MergeSink (disposables, sinks, sink) {
682 this.sink = sink
683 this.disposables = disposables
684 this.activeCount = sinks.length
687MergeSink.prototype.event = function (t) {
688 var that = this
689 return function (indexValue) {
690 that.sink.event(t)(indexValue.value)
691 }
694MergeSink.prototype.end = function (t, indexedValue) {
695 this.disposables[indexedValue.index].dispose()
696 if (--this.activeCount === 0) {
697 this.sink.end(t, indexedValue.value)
698 }
701function reduce (f, z, a) {
702 var r = z
703 for (var i = 0, l = a.length; i < l; ++i) {
704 r = f(r, a[i], i)
705 }
706 return r
709exports.until = function takeUntil (signal) {
710 return function (stream) {
711 return new Until(signal, stream)
712 }
715exports.since = function skipUntil (signal) {
716 return function (stream) {
717 return new Since(signal, stream)
718 }
721function Until (maxSignal, source) {
722 this.maxSignal = maxSignal
723 this.source = source
726Until.prototype.run = function (sink) {
727 var that = this
728 return function (scheduler) {
729 var min = new Bound(-Infinity, sink)
730 var max = new UpperBound(that.maxSignal, sink, scheduler)
731 var disposable = that.source.run(new TimeWindowSink(min, max, sink), scheduler)
733 return all([min, max, disposable])
734 }
737function Since (minSignal, source) {
738 this.minSignal = minSignal
739 this.source = source
742Since.prototype.run = function (sink) {
743 var that = this
744 return function (scheduler) {
745 var min = new LowerBound(that.minSignal, sink, scheduler)
746 var max = new Bound(Infinity, sink)
747 var disposable = that.source.run(new TimeWindowSink(min, max, sink))(scheduler)
749 return all([min, max, disposable])
750 }
753function Bound (value, sink) {
754 this.value = value
755 this.sink = sink
758Bound.prototype.event = noop
759Bound.prototype.end = noop
760Bound.prototype.dispose = noop
762function TimeWindowSink (min, max, sink) {
763 this.min = min
764 this.max = max
765 this.sink = sink
768TimeWindowSink.prototype.event = function (t, x) {
769 var that = this
770 return function (x) {
771 if (t >= that.min.value && t < that.max.value) {
772 that.sink.event(t)(x)
773 }
774 }
777TimeWindowSink.prototype.end = function (time) {
778 this.sink.end(time)
781function LowerBound (signal, sink, scheduler) {
782 this.value = Infinity
783 this.sink = sink
784 this.disposable = signal.run(this)(scheduler)
787LowerBound.prototype.event = function (t /*, x */) {
788 return function () {
789 if (t < this.value) {
790 this.value = t
791 }
792 }
795LowerBound.prototype.end = noop
797LowerBound.prototype.dispose = function () {
798 return this.disposable.dispose()
801function UpperBound (signal, sink, scheduler) {
802 this.value = Infinity
803 this.sink = sink
804 this.disposable = signal.run(this)(scheduler)
807UpperBound.prototype.event = function (t) {
808 var that = this
810 return function (x) {
811 if (t < this.value) {
812 that.value = t
813 that.sink.end(t)
814 }
815 }
818UpperBound.prototype.end = noop
820UpperBound.prototype.dispose = function () {
821 return this.disposable.dispose()
824function noop () { }
826exports.throttle = function throttle (period) {
827 return function (stream) {
828 return throttleSource(period, stream)
829 }
832function throttleSource (period, source) {
833 return source instanceof Throttle
834 ? fuseThrottle(period, source)
835 : new Throttle(period, source)
838function fuseThrottle (period, source) {
839 return new Throttle(Math.max(period, source.period), source.source)
842function Throttle (period, source) {
843 this.period = period
844 this.source = source
847Throttle.prototype.run = function (sink) {
848 var that = this
849 return function (scheduler) {
850 return that.source.run(new ThrottleSink(that.period, sink), scheduler)
851 }
854function ThrottleSink (period, sink) {
855 this.time = 0
856 this.period = period
857 this.sink = sink
860ThrottleSink.prototype.event = function (t) {
861 var that = this
862 return function (x) {
863 if (t >= this.time) {
864 that.time = t + this.period
865 that.sink.event(t)(x)
866 }
867 }
870ThrottleSink.prototype.end = function (t) { this.sink.end(t) }
872exports.debounce = function debounce (period) {
873 return function (stream) {
874 return new Debounce(period, stream)
875 }
878function Debounce (dt, source) {
879 this.dt = dt
880 this.source = source
883Debounce.prototype.run = function (sink) {
884 var that = this
885 return function (scheduler) {
886 return new DebounceSink(that.dt, that.source, sink, scheduler)
887 }
890function DebounceSink (dt, source, sink, scheduler) {
891 this.dt = dt
892 this.sink = sink
893 this.scheduler = scheduler
894 this.value = void 0
895 this.timer = null
897 var sourceDisposable = source.run(this)(scheduler)
898 this.disposable = all([this, sourceDisposable])
901DebounceSink.prototype.event = function (t) {
902 var that = this
903 return function (x) {
904 that._clearTimer()
905 that.value = x
906 var task = ControlStreamScheduler.eventTask(that.sink)(x)
907 that.timer = this.scheduler.delay(this.dt, task)
908 }
911DebounceSink.prototype.end = function (t, x) {
912 if (this._clearTimer()) {
913 this.sink.event(t, this.value)
914 this.value = void 0
915 }
916 this.sink.end(t, x)
919DebounceSink.prototype.dispose = function () {
920 this._clearTimer()
923DebounceSink.prototype._clearTimer = function () {
924 if (this.timer === null) {
925 return false
926 }
927 this.timer.dispose()
928 this.timer = null
929 return true
932exports.delay = function delay (delayTime) {
933 return function (stream) {
934 return delayTime <= 0 ? stream : new Delay(delayTime, stream)
935 }
938function Delay (dt, source) {
939 this.dt = dt
940 this.source = source
943Delay.prototype.run = function (sink) {
944 var that = this
946 return function (scheduler) {
947 var delaySink = new DelaySink(that.dt, sink, scheduler)
948 return all([delaySink, that.source.run(delaySink)(scheduler)])
949 }
952function DelaySink (dt, sink, scheduler) {
953 this.dt = dt
954 this.sink = sink
955 this.scheduler = scheduler
958DelaySink.prototype.dispose = function () {
959 var self = this
960 this.scheduler.cancelAll(function (task) {
961 return task.sink === self.sink
962 })
965DelaySink.prototype.event = function (t) {
966 var that = this
967 return function (x) {
968 that.scheduler.delay(that.dt)(ControlStreamScheduler.eventTask(that.sink)(x))
969 }
972DelaySink.prototype.end = function (t, x) {
973 this.scheduler.delay(this.dt)(ControlStreamScheduler.endTask(this.sink))
976var MulticastDisposable = function MulticastDisposable (source, sink) {
977 this.source = source
978 this.sink = sink
979 this.disposed = false
982MulticastDisposable.prototype.dispose = function dispose () {
983 if (this.disposed) {
984 return
985 }
986 this.disposed = true
987 var remaining = this.source.remove(this.sink)
988 return remaining === 0 && this.source._dispose()
991var dispose = function (disposable) { return disposable.dispose() }
993var emptyDisposable = {
994 dispose: function dispose$1 () { }
997function MulticastSource (source) {
998 this.source = source
999 this.sinks = []
1000 this._disposable = emptyDisposable
1003MulticastSource.prototype.run = function run (sink) {
1004 var that = this
1005 return function (scheduler) {
1006 var n = that.add(sink)
1007 if (n === 1) {
1008 that._disposable = that.source.run(that)(scheduler)
1010 return new MulticastDisposable(that, sink)
1014MulticastSource.prototype._dispose = function _dispose () {
1015 var disposable = this._disposable
1016 this._disposable = emptyDisposable
1017 return Promise.resolve(disposable).then(dispose)
1020MulticastSource.prototype.add = function add (sink) {
1021 this.sinks = append(sink, this.sinks)
1022 return this.sinks.length
1025function append (x, a) {
1026 var l = a.length
1027 var b = new Array(l + 1)
1028 for (var i = 0; i < l; ++i) {
1029 b[i] = a[i]
1032 b[l] = x
1033 return b
1036MulticastSource.prototype.remove = function remove$1 (sink) {
1037 var i = findIndex(sink, this.sinks)
1038 // istanbul ignore next
1039 if (i >= 0) {
1040 this.sinks = remove(i, this.sinks)
1043 return this.sinks.length
1046function findIndex (x, a) {
1047 for (var i = 0, l = a.length; i < l; ++i) {
1048 if (x === a[i]) {
1049 return i
1052 return -1
1055function remove (i, a) { // eslint-disable-line complexity
1056 if (i < 0) {
1057 throw new TypeError('i must be >= 0')
1060 var l = a.length
1061 if (l === 0 || i >= l) { // exit early if index beyond end of array
1062 return a
1065 if (l === 1) { // exit early if index in bounds and length === 1
1066 return []
1069 return unsafeRemove(i, a, l - 1)
1072// unsafeRemove :: Int -> [a] -> Int -> [a]
1073// Internal helper to remove element at index
1074function unsafeRemove (i, a, l) {
1075 var b = new Array(l)
1076 var j
1077 for (j = 0; j < i; ++j) {
1078 b[j] = a[j]
1080 for (j = i; j < l; ++j) {
1081 b[j] = a[j + 1]
1084 return b
1087MulticastSource.prototype.event = function event (time) {
1088 var that = this
1089 return function (value) {
1090 var s = that.sinks
1091 if (s.length === 1) {
1092 return s[0].event(time)(value)
1094 for (var i = 0; i < s.length; ++i) {
1095 s[i].event(time)(value)
1100MulticastSource.prototype.end = function end (time) {
1101 var s = this.sinks
1102 for (var i = 0; i < s.length; ++i) {
1103 s[i].end(time)
1107MulticastSource.prototype.until = function (signal) {
1108 return exports.until(signal)(this)
1111exports.multicast = function multicast (stream) {
1112 return new MulticastSource(stream)
1115function __extends (d, b) { // eslint-disable-line complexity
1116 for (var p in b) if (b.hasOwnProperty(p)) d[p] = b[p]
1117 function __ () { this.constructor = d }
1118 d.prototype = b === null ? Object.create(b) : (__.prototype = b.prototype, new __())
1121var HoldSubjectSource = (function (_super) {
1122 __extends(HoldSubjectSource, _super)
1123 function HoldSubjectSource (source, bufferSize) {
1124 var _this = _super.call(this, source) || this
1125 _this.has = false
1126 _this.buffer = []
1127 _this.bufferSize = bufferSize
1128 return _this
1130 HoldSubjectSource.prototype.add = function (sink) {
1131 if (this.has) {
1132 pushEvents(this.buffer, sink)
1134 return _super.prototype.add.call(this, sink)
1136 HoldSubjectSource.prototype.event = function (time) {
1137 var that = this
1138 return function (value) {
1139 that.has = true
1140 that.buffer = dropAndAppend(value, that.buffer, that.bufferSize)
1141 return _super.prototype.event.call(that, time)(value)
1144 return HoldSubjectSource
1145}(MulticastSource))
1147function pushEvents (buffer, sink) {
1148 var length = buffer.length
1149 for (var i = 0; i < length; ++i) {
1150 sink.event(ControlStreamScheduler.scheduler.now(), buffer[i])
1154function dropAndAppend (value, buffer, bufferSize) {
1155 if (buffer.length === bufferSize) {
1156 return append(value, drop(1, buffer))
1158 return append(value, buffer)
1161exports.hold = function hold (bufferSize) {
1162 return function (stream) {
1163 return new HoldSubjectSource(stream, bufferSize)
1167exports.toSubject = function (stream) {
1168 return exports.multicast(stream)
1171exports.toHoldSubject = exports.hold
1173exports.fromSubject = function (subject) {
1174 return subject
1177exports.subjectEvent = function (value) {
1178 return function (subject) {
1179 ControlStreamScheduler.eventTask(subject)(value)
1180 return subject
1184exports.subjectEnd = function (value) {
1185 return function (subject) {
1186 subject.end(Date.now())
1187 return subject
1191exports.proxy = function (value) {
1192 var stream = new ProxyStream(value)
1194 function attach (original) {
1195 stream.attach(original)
1196 return original
1199 return { attach: attach, stream: stream }
1202var ProxyStream = (function (_super) {
1203 __extends(ProxyStream, _super)
1204 function ProxyStream (value) {
1205 var _this = _super.call(this, ControlStreamStream.never, 1) || this
1206 _this.attached = false
1207 _this.running = false
1208 _this.buffer.push(value)
1209 return _this
1211 ProxyStream.prototype.run = function (sink) {
1212 var that = this
1214 that.add(sink)
1216 return function (scheduler) {
1217 if (that.attached && !that.running) {
1218 that.running = true
1219 that._disposable = that.source.run(that)(scheduler)
1220 return that._disposable
1223 return new MulticastDisposable(that, sink)
1227 ProxyStream.prototype.attach = function (source) {
1228 if (this.attached) throw new Error('Can only proxy 1 stream')
1229 this.attached = true
1231 this.source = source
1233 if (this.sinks.length) { this._disposable = source.run(this)(ControlStreamScheduler.scheduler) }
1236 ProxyStream.prototype.end = function end (time) {
1237 this.attached = false
1238 this.running = false
1240 return _super.prototype.end.call(this, time)
1243 return ProxyStream
1244}(HoldSubjectSource))