a50a246 · purescript-stream

git.iko.soy/purescript-stream

feat(stream): more combinators

commita50a246e69381a068903ae48195fd934cae0dc85 authorTylor Steinberger <tlsteinberger167@gmail.com> dateThu 16 Mar 2017, 00:38 (10 years ago) parent13c18b7

13 files changed+573−39

compare with master

unifiedsplit

bower.json+2−1

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

99 "dependencies": {

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

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

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

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

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

1314 },

1415 "devDependencies": {

1516 "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'

44export * from './merge'

55export * from './concat'

66export * 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 @@

22export * from './disposables'

33export * from './scheduler'

44export * 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)

1111class PropagateTask {

1212 constructor (run, value, sink) {

1313 let active = true

14+ this.sink = sink

1415

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

1617 this.run = t => {

src/Control/Stream.js+261

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

1212

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

1414// 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+ }

1521

22+ b[l] = x;

23+ return b

24+}

1625

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

1827// drop first n elements

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

8089

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

8291// 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+ }

83120

121+ return b

122+}

84123

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

86125// remove all elements matching a predicate

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

491530

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

493532 var active = true;

533+ this.sink = sink;

494534

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

496536 this.run = function (t) {

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

10991139 this.disposable.dispose();

11001140};

11011141

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+

11021359/* */

11031360

11041361exports.createDisposable = createDisposable;

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

11211378exports.concat = concat;

11221379exports.continueWith = continueWith;

11231380exports.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

33 , EffStream

44 , Stream

55 , Source

6+ , EventFn

7+ , EndFn

8+ , RunFn

69 , Sink

710 , Scheduler

811 , Task

912 , ScheduledTask

1013 , Disposable

1114 , Time

12− , EventFn

13− , EndFn

1415 -- Scheduler-related functions and values

1516 , defaultScheduler

1617 , eventTask

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

4647 , constant

4748 , scan

4849 , startWith

50+ , multicast

51+ , hold

52+ , skipRepeats

53+ , skipRepeatsWith

54+ , delay

55+ , sample

4956 -- Stream factories

5057 , just

5158 , fromArray

5259 , empty

60+ , never

5361 , periodic

5462 )

5563 where

5664

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

5766import Control.Applicative (class Applicative)

5867import Control.Apply (class Apply)

5968import Control.Bind (class Bind)

6069import Control.Category (id)

6170import Control.Monad (class Monad)

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

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

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

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

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

66−import Data.Array (snoc)

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

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

6877import Data.Functor (class Functor, map)

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

6979import Data.Monoid (class Monoid)

7080import Data.Semigroup (class Semigroup)

7181import Data.Unit (Unit, unit)

82+import Partial.Unsafe (unsafePartial)

7283

7384newtype Time = Time Int

7485

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

7889

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

8091

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

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

8293

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

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

85100 , end :: Time -> Unit

86101 }

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

107122

108123type Disposable = { dispose :: Pure Unit }

109124

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

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

112−

113125instance functorStream :: Functor Stream where

114126 map = _map

115127

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

138150

139151-- Stream-related helpers

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

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

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

142154

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

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

156+getSource (Stream stream) = stream.source

145157

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

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

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

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

160+

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

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

148163

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

150−createSink event end = { event, end }

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

151166

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

153168createCombinator event end stream = createStream runCombinator

154169 where

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

156171 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

158173

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

160175createEventCombinator event stream = createCombinator event end stream

161176 where

162− end :: EndFn b

163− end sink scheduler time =

164− sink.end time

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

165178

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

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

167180createEndCombinator end stream = createCombinator event end stream

168181 where

169− event :: EventFn a a

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

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

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

171184

172185-- Stream factories

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

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

192205runFromArray arr sink scheduler = scheduleTasks tasks scheduler

193206 where

194207 tasks :: Array Task

195− tasks = snoc eventTasks (endTask sink)

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

196209

197210 eventTasks :: Array Task

198211 eventTasks = map (flip eventTask sink) arr

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

201214periodic period = createStream (runPeriodic period)

202215

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

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

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

205218 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)

208220

209221-- combinators

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

211223_map f stream = createEventCombinator mapEvent stream

212224 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)

215226

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

217228filter predicate stream = createEventCombinator event stream

218229 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

221231 then sink.event time value

222232 else unit

223233

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

225235tapEvent f stream = createEventCombinator event stream

226236 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

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

230239 x = unsafePerformEff (f value)

231240

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

233242tapEnd f stream = createEndCombinator end stream

234243 where

235− end :: EndFn a

236− end sink scheduler time = sink.end time

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

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

238246 x = unsafePerformEff f

239247

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

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

244252constant value stream = createEventCombinator constantEvent stream

245253 where

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

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

247255

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

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

250258 where

259+ state :: State b

251260 state = createState seed

252261

253262 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

255294

256295-- 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

258299createState seed = { set, get }

259300 where

260301 ref :: Ref a

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

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

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

292333foreign 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