a50a246 · purescript-stream

git.iko.soy/purescript-stream

on master help sign in

feat(stream): more combinators

a50a246Tylor Steinberger10 years ago

bower.json+2−1

compare with master

unifiedsplit

@@ -9,7 +9,8 @@

9 "dependencies": {

10 "purescript-eff": "^2.0.0",

11 "purescript-prelude": "^2.5.0",

12− "purescript-refs": "^2.0.0"

12+ "purescript-refs": "^2.0.0",

13+ "purescript-maybe": "^2.1.1"

13 },

14 },

14 "devDependencies": {

15 "devDependencies": {

15 "purescript-console": "^2.0.0",

16 "purescript-console": "^2.0.0",

ffi/Control/Stream/combinators/delay.jsnew file+41

@@ -0,0 +1,41 @@

1+import { endTask, eventTask } from '../scheduler'

3+import { Stream } from '../Stream'

4+import { disposeAll } from '../disposables'

6+export const delay = delayTime => stream =>

7+ delayTime <= 0 ? stream : new Stream(new Delay(delayTime, stream.source))

9+class Delay {

10+ constructor (dt, source) {

11+ this.dt = dt

12+ this.source = source

13+ }

14+

15+ run (sink) {

16+ return scheduler => {

17+ const delaySink = new DelaySink(this.dt, sink, scheduler)

18+ return disposeAll([delaySink, this.source.run(delaySink)(scheduler)])

19+ }

20+ }

21+}

22+

23+class DelaySink {

24+ constructor (dt, sink, scheduler) {

25+ this.dt = dt

26+ this.sink = sink

27+ this.scheduler = scheduler

28+ }

29+

30+ dispose () {

31+ this.scheduler.cancelAll(task => task.sink === this.sink)

32+ }

33+

34+ event (t) {

35+ return x => this.scheduler.delay(this.dt)(eventTask(x)(this.sink))

36+ }

37+

38+ end (t) {

39+ this.scheduler.delay(this.dt)(endTask(this.sink))

40+ }

41+}

ffi/Control/Stream/combinators/index.js+2

@@ -4,3 +4,5 @@ export * from './combine'

4export * from './merge'

5export * from './concat'

6export * from './switch'

7+export * from './skipRepeatsWith'

8+export * from './delay'

ffi/Control/Stream/combinators/skipRepeatsWith.jsnew file+40

@@ -0,0 +1,40 @@

1+import { Stream } from '../Stream'

3+export const skipRepeatsWith = equals => stream =>

4+ new Stream(new SkipRepeats(equals, stream))

6+class SkipRepeats {

7+ constructor (equals, source) {

8+ this.equals = equals

9+ this.source = source

10+ }

11+

12+ run (sink) {

13+ return scheduler => this.source.run(new SkipRepeatsSink(this.equals, sink), scheduler)

14+ }

15+}

16+

17+class SkipRepeatsSink {

18+ constructor (equals, sink) {

19+ this.equals = equals

20+ this.value = void 0

21+ this.init = true

22+ }

23+

24+ event (t) {

25+ return x => {

26+ if (this.init) {

27+ this.init = false

28+ this.value = x

29+ this.sink.event(t, x)

30+ } else if (!this.equals(this.value)(x)) {

31+ this.value = x

32+ this.sink.event(t)(x)

33+ }

34+ }

35+ }

36+

37+ end (t) {

38+ this.sink.end(t)

39+ }

40+}

ffi/Control/Stream/index.js+1

@@ -2,3 +2,4 @@

2export * from './disposables'

3export * from './scheduler'

4export * from './combinators'

5+export * from './multicast'

ffi/Control/Stream/multicast/HoldSource.jsnew file+45

@@ -0,0 +1,45 @@

1+import { append, drop } from '@most/prelude'

3+import { MulticastSource } from './MulticastSource'

5+export class HoldSource extends MulticastSource {

6+ constructor (source, bufferSize) {

7+ super(source)

8+ this.bufferSize = bufferSize

9+ this.has = false

10+ this.buffer = []

11+ }

12+

13+ add (sink, scheduler) {

14+ if (this.has) {

15+ pushEvents(this.buffer, sink, scheduler)

16+ }

17+

18+ return super.add(sink, scheduler)

19+ }

20+

21+ event (time) {

22+ return value => {

23+ this.has = true

24+ this.buffer = dropAndAppend(value, this.buffer, this.bufferSize)

25+

26+ return super.event(time)(value)

27+ }

28+ }

29+}

30+

31+function pushEvents (buffer, sink, scheduler) {

32+ const length = buffer.length

33+

34+ for (let i = 0; i < length; ++i) {

35+ sink.event(scheduler.now(), buffer[i])

36+ }

37+}

38+

39+export function dropAndAppend (value, buffer, bufferSize) {

40+ if (buffer.length === bufferSize) {

41+ return append(value, drop(1, buffer))

42+ }

43+

44+ return append(value, buffer)

45+}

ffi/Control/Stream/multicast/MulticastDisposable.jsnew file+16

@@ -0,0 +1,16 @@

1+export class MulticastDisposable {

2+ constructor (source, sink) {

3+ this.source = source

4+ this.sink = sink

5+ this.disposed = false

6+ }

8+ dispose () {

9+ if (this.disposed) {

10+ return

11+ }

12+ this.disposed = true

13+ const remaining = this.source.remove(this.sink)

14+ return remaining === 0 && this.source._dispose()

15+ }

16+}

ffi/Control/Stream/multicast/MulticastSource.jsnew file+62

@@ -0,0 +1,62 @@

1+import { append, findIndex, remove } from '@most/prelude'

2+import { dispose, emptyDisposable } from './dispose'

4+import { MulticastDisposable } from './MulticastDisposable'

6+export class MulticastSource {

7+ constructor (source) {

8+ this.source = source

9+ this.sinks = []

10+ this._disposable = emptyDisposable

11+ }

12+

13+ run (sink) {

14+ return scheduler => {

15+ const n = this.add(sink, scheduler)

16+ if (n === 1) {

17+ this._disposable = this.source.run(this)(scheduler)

18+ }

19+ return new MulticastDisposable(this, sink)

20+ }

21+ }

22+

23+ _dispose () {

24+ const disposable = this._disposable

25+ this._disposable = emptyDisposable

26+ return Promise.resolve(disposable).then(dispose)

27+ }

28+

29+ add (sink) {

30+ this.sinks = append(sink, this.sinks)

31+ return this.sinks.length

32+ }

33+

34+ remove (sink) {

35+ const i = findIndex(sink, this.sinks)

36+ // istanbul ignore next

37+ if (i >= 0) {

38+ this.sinks = remove(i, this.sinks)

39+ }

40+

41+ return this.sinks.length

42+ }

43+

44+ event (time) {

45+ return value => {

46+ const s = this.sinks

47+ if (s.length === 1) {

48+ return s[0].event(time)(value)

49+ }

50+ for (let i = 0; i < s.length; ++i) {

51+ s[i].event(time)(value)

52+ }

53+ }

54+ }

55+

56+ end (time) {

57+ const s = this.sinks

58+ for (let i = 0; i < s.length; ++i) {

59+ s[i].end(time)

60+ }

61+ }

62+}

ffi/Control/Stream/multicast/dispose.jsnew file+5

@@ -0,0 +1,5 @@

1+export const dispose = disposable => disposable.dispose()

3+export const emptyDisposable = {

4+ dispose () {}

5+}

ffi/Control/Stream/multicast/index.jsnew file+14

@@ -0,0 +1,14 @@

1+import { HoldSource } from './HoldSource'

2+import { MulticastSource } from './MulticastSource'

3+import { Stream } from '../Stream'

5+export function multicast (stream) {

6+ const source = stream.source

8+ return source instanceof MulticastSource

9+ ? stream

10+ : new Stream(new MulticastSource(source))

11+}

12+

13+export const hold = bufferSize => stream =>

14+ new Stream(new HoldSource(stream.source, bufferSize))

ffi/Control/Stream/scheduler/PropagateTask.js+1

@@ -11,6 +11,7 @@ export const endTask = sink => propagateTask(runEnd, void 0, sink)

11class PropagateTask {

12 constructor (run, value, sink) {

13 let active = true

14+ this.sink = sink

15 this.dispose = () => { active = false }

16 this.dispose = () => { active = false }

16 this.run = t => {

17 this.run = t => {

src/Control/Stream.js+261

@@ -12,7 +12,16 @@ Object.defineProperty(exports, '__esModule', { value: true });

13// append :: a -> [a] -> [a]

14// a with x appended

15+function append (x, a) {

16+ var l = a.length;

17+ var b = new Array(l + 1);

18+ for (var i = 0; i < l; ++i) {

19+ b[i] = a[i];

20+ }

22+ b[l] = x;

23+ return b

24+}

17// drop :: Int -> [a] -> [a]

26// drop :: Int -> [a] -> [a]

18// drop first n elements

27// drop first n elements

@@ -80,7 +89,37 @@ function reduce (f, z, a) {

81// remove :: Int -> [a] -> [a]

90// remove :: Int -> [a] -> [a]

82// remove element at index

91// remove element at index

92+function remove (i, a) { // eslint-disable-line complexity

93+ if (i < 0) {

94+ throw new TypeError('i must be >= 0')

95+ }

96+

97+ var l = a.length;

98+ if (l === 0 || i >= l) { // exit early if index beyond end of array

99+ return a

100+ }

101+

102+ if (l === 1) { // exit early if index in bounds and length === 1

103+ return []

104+ }

105+

106+ return unsafeRemove(i, a, l - 1)

107+}

108+

109+// unsafeRemove :: Int -> [a] -> Int -> [a]

110+// Internal helper to remove element at index

111+function unsafeRemove (i, a, l) {

112+ var b = new Array(l);

113+ var j;

114+ for (j = 0; j < i; ++j) {

115+ b[j] = a[j];

116+ }

117+ for (j = i; j < l; ++j) {

118+ b[j] = a[j + 1];

119+ }

120

121+ return b

122+}

123

85// removeAll :: (a -> boolean) -> [a] -> [a]

124// removeAll :: (a -> boolean) -> [a] -> [a]

86// remove all elements matching a predicate

125// remove all elements matching a predicate

@@ -491,6 +530,7 @@ var endTask = function (sink) { return propagateTask(runEnd, void 0, sink); };

491

530

492var PropagateTask = function PropagateTask (run, value, sink) {

531var PropagateTask = function PropagateTask (run, value, sink) {

493 var active = true;

532 var active = true;

533+ this.sink = sink;

494

534

495 this.dispose = function () { active = false; };

535 this.dispose = function () { active = false; };

496 this.run = function (t) {

536 this.run = function (t) {

@@ -1099,6 +1139,223 @@ Segment.prototype._dispose = function (t) {

1099 this.disposable.dispose();

1139 this.disposable.dispose();

1100};

1140};

1101

1141

1142+var skipRepeatsWith = function (equals) { return function (stream) { return new Stream(new SkipRepeats(equals, stream)); }; };

1143+

1144+var SkipRepeats = function SkipRepeats (equals, source) {

1145+ this.equals = equals;

1146+ this.source = source;

1147+};

1148+

1149+SkipRepeats.prototype.run = function run (sink) {

1150+ var this$1 = this;

1151+

1152+ return function (scheduler) { return this$1.source.run(new SkipRepeatsSink(this$1.equals, sink), scheduler); }

1153+};

1154+

1155+var SkipRepeatsSink = function SkipRepeatsSink (equals, sink) {

1156+ this.equals = equals;

1157+ this.value = void 0;

1158+ this.init = true;

1159+};

1160+

1161+SkipRepeatsSink.prototype.event = function event (t) {

1162+ var this$1 = this;

1163+

1164+ return function (x) {

1165+ if (this$1.init) {

1166+ this$1.init = false;

1167+ this$1.value = x;

1168+ this$1.sink.event(t, x);

1169+ } else if (!this$1.equals(this$1.value)(x)) {

1170+ this$1.value = x;

1171+ this$1.sink.event(t)(x);

1172+ }

1173+ }

1174+};

1175+

1176+SkipRepeatsSink.prototype.end = function end (t) {

1177+ this.sink.end(t);

1178+};

1179+

1180+var delay = function (delayTime) { return function (stream) { return delayTime <= 0 ? stream : new Stream(new Delay(delayTime, stream.source)); }; };

1181+

1182+var Delay = function Delay (dt, source) {

1183+ this.dt = dt;

1184+ this.source = source;

1185+};

1186+

1187+Delay.prototype.run = function run (sink) {

1188+ var this$1 = this;

1189+

1190+ return function (scheduler) {

1191+ var delaySink = new DelaySink(this$1.dt, sink, scheduler);

1192+ return disposeAll([delaySink, this$1.source.run(delaySink)(scheduler)])

1193+ }

1194+};

1195+

1196+var DelaySink = function DelaySink (dt, sink, scheduler) {

1197+ this.dt = dt;

1198+ this.sink = sink;

1199+ this.scheduler = scheduler;

1200+};

1201+

1202+DelaySink.prototype.dispose = function dispose () {

1203+ var this$1 = this;

1204+

1205+ this.scheduler.cancelAll(function (task) { return task.sink === this$1.sink; });

1206+};

1207+

1208+DelaySink.prototype.event = function event (t) {

1209+ var this$1 = this;

1210+

1211+ return function (x) { return this$1.scheduler.delay(this$1.dt)(eventTask(x)(this$1.sink)); }

1212+};

1213+

1214+DelaySink.prototype.end = function end (t) {

1215+ this.scheduler.delay(this.dt)(endTask(this.sink));

1216+};

1217+

1218+var dispose$1 = function (disposable) { return disposable.dispose(); };

1219+

1220+var emptyDisposable$1 = {

1221+ dispose: function dispose () {}

1222+};

1223+

1224+var MulticastDisposable = function MulticastDisposable (source, sink) {

1225+ this.source = source;

1226+ this.sink = sink;

1227+ this.disposed = false;

1228+};

1229+

1230+MulticastDisposable.prototype.dispose = function dispose () {

1231+ if (this.disposed) {

1232+ return

1233+ }

1234+ this.disposed = true;

1235+ var remaining = this.source.remove(this.sink);

1236+ return remaining === 0 && this.source._dispose()

1237+};

1238+

1239+var MulticastSource = function MulticastSource (source) {

1240+ this.source = source;

1241+ this.sinks = [];

1242+ this._disposable = emptyDisposable$1;

1243+};

1244+

1245+MulticastSource.prototype.run = function run (sink) {

1246+ var this$1 = this;

1247+

1248+ return function (scheduler) {

1249+ var n = this$1.add(sink, scheduler);

1250+ if (n === 1) {

1251+ this$1._disposable = this$1.source.run(this$1)(scheduler);

1252+ }

1253+ return new MulticastDisposable(this$1, sink)

1254+ }

1255+};

1256+

1257+MulticastSource.prototype._dispose = function _dispose () {

1258+ var disposable = this._disposable;

1259+ this._disposable = emptyDisposable$1;

1260+ return Promise.resolve(disposable).then(dispose$1)

1261+};

1262+

1263+MulticastSource.prototype.add = function add (sink) {

1264+ this.sinks = append(sink, this.sinks);

1265+ return this.sinks.length

1266+};

1267+

1268+MulticastSource.prototype.remove = function remove$1 (sink) {

1269+ var i = findIndex(sink, this.sinks);

1270+ // istanbul ignore next

1271+ if (i >= 0) {

1272+ this.sinks = remove(i, this.sinks);

1273+ }

1274+

1275+ return this.sinks.length

1276+};

1277+

1278+MulticastSource.prototype.event = function event (time) {

1279+ var this$1 = this;

1280+

1281+ return function (value) {

1282+ var s = this$1.sinks;

1283+ if (s.length === 1) {

1284+ return s[0].event(time)(value)

1285+ }

1286+ for (var i = 0; i < s.length; ++i) {

1287+ s[i].event(time)(value);

1288+ }

1289+ }

1290+};

1291+

1292+MulticastSource.prototype.end = function end (time) {

1293+ var s = this.sinks;

1294+ for (var i = 0; i < s.length; ++i) {

1295+ s[i].end(time);

1296+ }

1297+};

1298+

1299+var HoldSource = (function (MulticastSource$$1) {

1300+ function HoldSource (source, bufferSize) {

1301+ MulticastSource$$1.call(this, source);

1302+ this.bufferSize = bufferSize;

1303+ this.has = false;

1304+ this.buffer = [];

1305+ }

1306+

1307+ if ( MulticastSource$$1 ) HoldSource.__proto__ = MulticastSource$$1;

1308+ HoldSource.prototype = Object.create( MulticastSource$$1 && MulticastSource$$1.prototype );

1309+ HoldSource.prototype.constructor = HoldSource;

1310+

1311+ HoldSource.prototype.add = function add (sink, scheduler) {

1312+ if (this.has) {

1313+ pushEvents(this.buffer, sink, scheduler);

1314+ }

1315+

1316+ return MulticastSource$$1.prototype.add.call(this, sink, scheduler)

1317+ };

1318+

1319+ HoldSource.prototype.event = function event (time) {

1320+ var this$1 = this;

1321+

1322+ return function (value) {

1323+ this$1.has = true;

1324+ this$1.buffer = dropAndAppend(value, this$1.buffer, this$1.bufferSize);

1325+

1326+ return MulticastSource$$1.prototype.event.call(this$1, time)(value)

1327+ }

1328+ };

1329+

1330+ return HoldSource;

1331+}(MulticastSource));

1332+

1333+function pushEvents (buffer, sink, scheduler) {

1334+ var length = buffer.length;

1335+

1336+ for (var i = 0; i < length; ++i) {

1337+ sink.event(scheduler.now(), buffer[i]);

1338+ }

1339+}

1340+

1341+function dropAndAppend (value, buffer, bufferSize) {

1342+ if (buffer.length === bufferSize) {

1343+ return append(value, drop(1, buffer))

1344+ }

1345+

1346+ return append(value, buffer)

1347+}

1348+

1349+function multicast (stream) {

1350+ var source = stream.source;

1351+

1352+ return source instanceof MulticastSource

1353+ ? stream

1354+ : new Stream(new MulticastSource(source))

1355+}

1356+

1357+var hold = function (bufferSize) { return function (stream) { return new Stream(new HoldSource(stream.source, bufferSize)); }; };

1358+

1102/* */

1359/* */

1103

1360

1104exports.createDisposable = createDisposable;

1361exports.createDisposable = createDisposable;

@@ -1121,3 +1378,7 @@ exports.merge = merge;

1121exports.concat = concat;

1378exports.concat = concat;

1122exports.continueWith = continueWith;

1379exports.continueWith = continueWith;

1123exports.switch = switchLatest;

1380exports.switch = switchLatest;

1381+exports.skipRepeatsWith = skipRepeatsWith;

1382+exports.delay = delay;

1383+exports.multicast = multicast;

1384+exports.hold = hold;

src/Control/Stream.purs+83−38

@@ -3,14 +3,15 @@ module Control.Stream

3 , EffStream

4 , Stream

5 , Source

6+ , EventFn

7+ , EndFn

8+ , RunFn

6 , Sink

9 , Sink

7 , Scheduler

10 , Scheduler

8 , Task

11 , Task

9 , ScheduledTask

12 , ScheduledTask

10 , Disposable

13 , Disposable

11 , Time

14 , Time

12− , EventFn

13− , EndFn

14 -- Scheduler-related functions and values

15 -- Scheduler-related functions and values

15 , defaultScheduler

16 , defaultScheduler

16 , eventTask

17 , eventTask

@@ -46,29 +47,39 @@ module Control.Stream

46 , constant

47 , constant

47 , scan

48 , scan

48 , startWith

49 , startWith

50+ , multicast

51+ , hold

52+ , skipRepeats

53+ , skipRepeatsWith

54+ , delay

55+ , sample

49 -- Stream factories

56 -- Stream factories

50 , just

57 , just

51 , fromArray

58 , fromArray

52 , empty

59 , empty

60+ , never

53 , periodic

61 , periodic

54 )

62 )

55 where

63 where

65+import Control.Alt ((<|>))

57import Control.Applicative (class Applicative)

66import Control.Applicative (class Applicative)

58import Control.Apply (class Apply)

67import Control.Apply (class Apply)

59import Control.Bind (class Bind)

68import Control.Bind (class Bind)

60import Control.Category (id)

69import Control.Category (id)

61import Control.Monad (class Monad)

70import Control.Monad (class Monad)

62−import Control.Monad.Eff (Eff, Pure)

71+import Control.Monad.Eff (Eff, Pure, runPure)

63import Control.Monad.Eff.Ref (Ref, modifyRef', newRef, readRef)

72import Control.Monad.Eff.Ref (Ref, modifyRef', newRef, readRef)

64import Control.Monad.Eff.Unsafe (unsafeCoerceEff, unsafePerformEff)

73import Control.Monad.Eff.Unsafe (unsafeCoerceEff, unsafePerformEff)

65import Control.MonadPlus (class Alt, class Plus)

74import Control.MonadPlus (class Alt, class Plus)

66−import Data.Array (snoc)

75+import Data.Eq (class Eq, eq)

67import Data.Function (flip, ($))

76import Data.Function (flip, ($))

68import Data.Functor (class Functor, map)

77import Data.Functor (class Functor, map)

78+import Data.Maybe (Maybe(Just, Nothing), fromJust, isJust)

69import Data.Monoid (class Monoid)

79import Data.Monoid (class Monoid)

70import Data.Semigroup (class Semigroup)

80import Data.Semigroup (class Semigroup)

71import Data.Unit (Unit, unit)

81import Data.Unit (Unit, unit)

82+import Partial.Unsafe (unsafePartial)

73newtype Time = Time Int

84newtype Time = Time Int

@@ -78,9 +89,13 @@ type EffStream e a = Eff (stream :: STREAM | e) a

79newtype Stream a = Stream { source :: Source a }

90newtype Stream a = Stream { source :: Source a }

81−type Source a = { run :: Sink a -> Scheduler -> Disposable }

92+newtype Source a = Source { run :: RunFn a }

83−type Sink a =

94+type RunFn a = Sink a -> Scheduler -> Disposable

95+type EventFn a = Time -> a -> Unit

96+type EndFn = Time -> Unit

97+

98+newtype Sink a = Sink

84 { event :: Time -> a -> Unit

99 { event :: Time -> a -> Unit

85 , end :: Time -> Unit

100 , end :: Time -> Unit

86 }

101 }

@@ -107,9 +122,6 @@ type ScheduledTask =

107

122

108type Disposable = { dispose :: Pure Unit }

123type Disposable = { dispose :: Pure Unit }

109

124

110−type EventFn a b = (Sink b -> Scheduler -> Time -> a -> Unit)

111−type EndFn a = (Sink a -> Scheduler -> Time -> Unit)

112−

113instance functorStream :: Functor Stream where

125instance functorStream :: Functor Stream where

114 map = _map

126 map = _map

115

127

@@ -138,36 +150,37 @@ instance plusStream :: Plus Stream where

138

150

139-- Stream-related helpers

151-- Stream-related helpers

140createStream :: forall a. (Sink a -> Scheduler -> Disposable) -> Stream a

152createStream :: forall a. (Sink a -> Scheduler -> Disposable) -> Stream a

141−createStream run = Stream { source: { run } }

153+createStream run = Stream { source: Source { run } }

142

154

143getSource :: forall a. Stream a -> Source a

155getSource :: forall a. Stream a -> Source a

144−getSource stream = case stream of Stream a -> a.source

156+getSource (Stream stream) = stream.source

145

157

158+runSource :: forall a. Source a -> Sink a -> Scheduler -> Disposable

159+runSource (Source source) sink scheduler = source.run sink scheduler

160+

146−runSource :: forall a. Stream a -> Sink a -> Scheduler -> Disposable

161+runStream :: forall a. Stream a -> Sink a -> Scheduler -> Disposable

147−runSource stream sink scheduler = (getSource stream).run sink scheduler

162+runStream stream sink scheduler = runSource (getSource stream) sink scheduler

148

163

149createSink :: forall a. (Time -> a -> Unit) -> (Time -> Unit) -> Sink a

164createSink :: forall a. (Time -> a -> Unit) -> (Time -> Unit) -> Sink a

150−createSink event end = { event, end }

165+createSink event end = Sink { event, end }

151

166

152−createCombinator :: forall a b. EventFn a b -> EndFn b -> Stream a -> Stream b

167+createCombinator :: forall a b. (Sink b -> Scheduler -> EventFn a) -> (Sink b -> Scheduler -> EndFn) -> Stream a -> Stream b

153createCombinator event end stream = createStream runCombinator

168createCombinator event end stream = createStream runCombinator

154 where

169 where

155 runCombinator :: Sink b -> Scheduler -> Disposable

170 runCombinator :: Sink b -> Scheduler -> Disposable

156 runCombinator sink scheduler =

171 runCombinator sink scheduler =

157− runSource stream (createSink (event sink scheduler) (end sink scheduler)) scheduler

172+ runStream stream (createSink (event sink scheduler) (end sink scheduler)) scheduler

158

173

159−createEventCombinator :: forall a b. EventFn a b -> Stream a -> Stream b

174+createEventCombinator :: forall a b. (Sink b -> Scheduler -> EventFn a) -> Stream a -> Stream b

160createEventCombinator event stream = createCombinator event end stream

175createEventCombinator event stream = createCombinator event end stream

161 where

176 where

162− end :: EndFn b

163− end sink scheduler time =

177+ end (Sink sink) (Scheduler scheduler) time = sink.end time

164− sink.end time

165

178

166−createEndCombinator :: forall a. EndFn a -> Stream a -> Stream a

179+createEndCombinator :: forall a. (Sink a -> Scheduler -> EndFn) -> Stream a -> Stream a

167createEndCombinator end stream = createCombinator event end stream

180createEndCombinator end stream = createCombinator event end stream

168 where

181 where

169− event :: EventFn a a

182+ event :: Sink a -> Scheduler -> EventFn a

170− event sink scheduler time value = sink.event time value

183+ event (Sink sink) scheduler time value = sink.event time value

171

184

172-- Stream factories

185-- Stream factories

173just :: forall a. a -> Stream a

186just :: forall a. a -> Stream a

@@ -192,7 +205,7 @@ runFromArray :: forall a. Array a -> Sink a -> Scheduler -> Disposable

192runFromArray arr sink scheduler = scheduleTasks tasks scheduler

205runFromArray arr sink scheduler = scheduleTasks tasks scheduler

193 where

206 where

194 tasks :: Array Task

207 tasks :: Array Task

195− tasks = snoc eventTasks (endTask sink)

208+ tasks = eventTasks <|> [ (endTask sink) ]

196

209

197 eventTasks :: Array Task

210 eventTasks :: Array Task

198 eventTasks = map (flip eventTask sink) arr

211 eventTasks = map (flip eventTask sink) arr

@@ -201,39 +214,34 @@ periodic :: Number -> Stream Unit

201periodic period = createStream (runPeriodic period)

214periodic period = createStream (runPeriodic period)

202

215

203runPeriodic :: Number -> Sink Unit -> Scheduler -> Disposable

216runPeriodic :: Number -> Sink Unit -> Scheduler -> Disposable

204−runPeriodic period sink scheduler = { dispose: scheduledTask.dispose }

217+runPeriodic period sink (Scheduler scheduler) = { dispose: scheduledTask.dispose }

205 where

218 where

206− p = case scheduler of Scheduler a -> a.periodic

207− scheduledTask = p period (eventTask unit sink)

219+ scheduledTask = scheduler.periodic period (eventTask unit sink)

208

220

209-- combinators

221-- combinators

210_map :: forall a b. (a -> b) -> Stream a -> Stream b

222_map :: forall a b. (a -> b) -> Stream a -> Stream b

211_map f stream = createEventCombinator mapEvent stream

223_map f stream = createEventCombinator mapEvent stream

212 where

224 where

213− mapEvent :: EventFn a b

214− mapEvent sink scheduler time value = sink.event time (f value)

225+ mapEvent (Sink sink) (Scheduler scheduler) time value = sink.event time (f value)

215

226

216filter :: forall a. (a -> Boolean) -> Stream a -> Stream a

227filter :: forall a. (a -> Boolean) -> Stream a -> Stream a

217filter predicate stream = createEventCombinator event stream

228filter predicate stream = createEventCombinator event stream

218 where

229 where

219− event :: EventFn a a

220− event sink scheduler time value = if predicate value

230+ event (Sink sink) (Scheduler scheduler) time value = if predicate value

221 then sink.event time value

231 then sink.event time value

222 else unit

232 else unit

223

233

224tapEvent :: forall e a. (a -> EffStream e Unit) -> Stream a -> Stream a

234tapEvent :: forall e a. (a -> EffStream e Unit) -> Stream a -> Stream a

225tapEvent f stream = createEventCombinator event stream

235tapEvent f stream = createEventCombinator event stream

226 where

236 where

227− event :: EventFn a a

228− event sink scheduler time value = sink.event time value

237+ event (Sink sink) (Scheduler scheduler) time value = sink.event time value

229 where -- find a better way to perform these side effects

238 where -- find a better way to perform these side effects

230 x = unsafePerformEff (f value)

239 x = unsafePerformEff (f value)

231

240

232tapEnd :: forall e a. EffStream e Unit -> Stream a -> Stream a

241tapEnd :: forall e a. EffStream e Unit -> Stream a -> Stream a

233tapEnd f stream = createEndCombinator end stream

242tapEnd f stream = createEndCombinator end stream

234 where

243 where

235− end :: EndFn a

236− end sink scheduler time = sink.end time

244+ end (Sink sink) (Scheduler scheduler) time = sink.end time

237 where -- find a better way to perform these side effects

245 where -- find a better way to perform these side effects

238 x = unsafePerformEff f

246 x = unsafePerformEff f

239

247

@@ -243,18 +251,51 @@ startWith value stream = concat (just value) stream

243constant :: forall a b. b -> Stream a -> Stream b

251constant :: forall a b. b -> Stream a -> Stream b

244constant value stream = createEventCombinator constantEvent stream

252constant value stream = createEventCombinator constantEvent stream

245 where

253 where

246− constantEvent sink scheduler time x = sink.event time value

254+ constantEvent (Sink sink) (Scheduler scheduler) time x = sink.event time value

247

255

248scan :: forall a b. (b -> a -> b) -> b -> Stream a -> Stream b

256scan :: forall a b. (b -> a -> b) -> b -> Stream a -> Stream b

249scan f seed stream = startWith seed $ createEventCombinator scanEvent stream

257scan f seed stream = startWith seed $ createEventCombinator scanEvent stream

250 where

258 where

259+ state :: State b

251 state = createState seed

260 state = createState seed

252

261

253 scanEvent :: Sink b -> Scheduler -> Time -> a -> Unit

262 scanEvent :: Sink b -> Scheduler -> Time -> a -> Unit

254− scanEvent sink scheduler time value = sink.event time (state.set \acc -> f acc value)

263+ scanEvent (Sink sink) (Scheduler scheduler) time value = sink.event time (state.set \acc -> f acc value)

264+

265+skipRepeats :: forall a. (Eq a) => Stream a -> Stream a

266+skipRepeats = skipRepeatsWith eq

267+

268+sample :: forall a b c. (a -> b -> c) -> Stream a -> Stream b -> Stream c

269+sample f sampler stream = createStream $ runSample (createState Nothing)

270+ where

271+ runSample :: State (Maybe b) -> Sink c -> Scheduler -> Disposable

272+ runSample state sink scheduler = disposeAll

273+ [ runStream stream (createHoldSink state) scheduler

274+ , runStream sampler (createSampleSink state sink f) scheduler

275+ ]

276+

277+createHoldSink :: forall a. State (Maybe a) -> Sink a

278+createHoldSink state = createSink event end

279+ where

280+ event time value = always unit (state.set \_ -> Just value)

281+ end time = unit

282+

283+createSampleSink :: forall a b c. State (Maybe b) -> Sink c -> (a -> b -> c) -> Sink a

284+createSampleSink state (Sink sink) f = createSink event end

285+ where

286+ end time = sink.end time

287+ event time value =

288+ if isJust (runPure state.get)

289+ then sink.event time $ f value (unsafePartial $ fromJust (runPure state.get))

290+ else unit

291+

292+always :: forall a b. a -> b -> a

293+always a b = a

255

294

256-- find a better way to perform these side effects

295-- find a better way to perform these side effects

257−createState :: forall a. a -> { get :: Pure a, set :: (a -> a) -> a }

296+type State a = { get:: Pure a, set :: (a -> a) -> a }

297+

298+createState :: forall a. a -> State a

258createState seed = { set, get }

299createState seed = { set, get }

259 where

300 where

260 ref :: Ref a

301 ref :: Ref a

@@ -290,3 +331,7 @@ foreign import continueWith :: forall a. (Unit -> Stream a) -> Stream a -> Strea

290foreign import concat :: forall a. Stream a -> Stream a -> Stream a

331foreign import concat :: forall a. Stream a -> Stream a -> Stream a

291foreign import merge :: forall a. Array (Stream a) -> Stream a

332foreign import merge :: forall a. Array (Stream a) -> Stream a

292foreign import switch :: forall a. Stream (Stream a) -> Stream a

333foreign import switch :: forall a. Stream (Stream a) -> Stream a

334+foreign import multicast :: forall a. Stream a -> Stream a

335+foreign import hold :: forall a. Int -> Stream a -> Stream a

336+foreign import skipRepeatsWith :: forall a. (a -> a -> Boolean) -> Stream a -> Stream a

337+foreign import delay :: forall a. Number -> Stream a -> Stream a