git.iko.soy/purescript-stream
on master help sign in
purescript-stream/src/Control/Stream/Stream.purs
340 lines 12 kB history raw
1module Control.Stream.Stream
2 ( STREAM
3 , EffStream
4 , Stream(..)
5 , just
6 , empty
7 , never
8 , periodic
9 , createCombinator
10 , createEventCombinator
11 , createEndCombinator
12 , drain
13 , tapEvent
14 , tapEnd
15 , fromArray
16 , filter
17 , take
18 , skip
19 , constant
20 , continueWith
21 , concat
22 , startWith
23 , scan
24 , loop
25 , chain
26 , mergeConcurrently
27 , join
28 , mergeMapConcurrently
29 , concatMap
30 , switch
31 , combine2
32 , combine3
33 , combine4
34 , combine5
35 , combineArray
36 , merge
37 , skipRepeatsWith
38 , skipRepeats
39 , until
40 , since
41 , during
42 , sample
43 , debounce
44 , throttle
45 , delay
46 , multicast
47 , hold
48 , Subject
49 , toSubject
50 , toHoldSubject
51 , fromSubject
52 , subjectEvent
53 , subjectEnd
54 , subscribe
55 ) where
57import Control.Alternative (class Apply)
58import Control.Bind (class Bind)
59import Control.Category (id)
60import Control.Monad.Eff (Eff, Pure, runPure)
61import Control.Monad.Eff.Ref (Ref, modifyRef', newRef, readRef)
62import Control.Monad.Eff.Unsafe (unsafeCoerceEff, unsafePerformEff)
63import Control.MonadPlus (class Alt, class Alternative, class MonadPlus, class Plus)
64import Control.MonadZero (class Applicative, class MonadZero)
65import Control.Stream.Disposable (Disposable, emptyDisposable, disposeAll)
66import Control.Stream.Scheduler (Scheduler(..), endTask, eventTask, scheduleTasks)
67import Control.Stream.Sink (Sink(..))
68import Control.Stream.Stream (Stream(..), EffStream, empty, just)
69import Control.Stream.Time (Time)
70import Data.Eq ((==))
71import Data.EuclideanRing ((-))
72import Data.Functor (class Functor)
73import Data.Maybe (Maybe(..), fromJust, isNothing)
74import Data.Monoid (class Monoid)
75import Data.Unit (Unit, unit)
76import Partial.Unsafe (unsafePartial)
77import Prelude (class Eq, class Monad, class Semigroup, eq)
79foreign import data STREAM :: !
81type EffStream e a = Eff (stream :: STREAM | e) a
83foreign import data Subject :: * -> *
85newtype Stream a = Stream { run :: Sink a -> Scheduler -> Disposable }
87instance functorStream :: Functor Stream where
88 map = mapStream
90instance applyStream :: Apply Stream where
91 apply = combine2 id
93instance applicativeStream :: Applicative Stream where
94 pure = just
96instance bindStream :: Bind Stream where
97 bind = chain
99instance monadStream :: Monad Stream
101instance semigroupStream :: Semigroup (Stream a) where
102 append = concat
104instance altStream :: Alt Stream where
105 alt = \s1 s2 -> merge [s1, s2]
107instance monoidStream :: Monoid (Stream a) where
108 mempty = empty
110instance plusStream :: Plus Stream where
111 empty = empty
113instance alternativeStream :: Alternative Stream
115instance monadZeroStream :: MonadZero Stream
117instance monadPlusStream :: MonadPlus Stream
119just :: forall a. a -> Stream a
120just a = Stream { run } where
121 run :: Sink a -> Scheduler -> Disposable
122 run sink (Scheduler scheduler) =
123 scheduleTasks scheduler.asap [ (eventTask sink a), (endTask sink) ]
125empty :: forall a. Stream a
126empty = Stream { run } where
127 run :: Sink a -> Scheduler -> Disposable
128 run sink (Scheduler scheduler) = scheduleTasks scheduler.asap [ endTask sink ]
130never :: forall a. Stream a
131never = Stream { run: \sink scheduler -> emptyDisposable }
133periodic :: Number -> Stream Unit
134periodic period = Stream { run } where
135 run :: Sink Unit -> Scheduler -> Disposable
136 run sink (Scheduler scheduler) = do
137 let task = eventTask sink unit
138 let scheduledTask = scheduler.periodic period task
139 { dispose : scheduledTask.dispose }
141createCombinator :: forall a b. (Sink b -> Time -> a -> Unit) -> (Sink b -> Time -> Unit) -> Stream a -> Stream b
142createCombinator event end (Stream stream) = Stream { run } where
143 run :: Sink b -> Scheduler -> Disposable
144 run sink scheduler = stream.run (Sink { event: (event sink), end: (end sink) }) scheduler
146createEventCombinator :: forall a b. (Sink b -> Time -> a -> Unit) -> Stream a -> Stream b
147createEventCombinator event stream = createCombinator event end stream where
148 end (Sink sink) time = sink.end time
150createEndCombinator :: forall a. (Sink a -> Time -> Unit) -> Stream a -> Stream a
151createEndCombinator end stream = createCombinator event end stream where
152 event (Sink sink) time value = sink.event time value
154tapEvent :: forall e a. (a -> Eff e Unit) -> Stream a -> Stream a
155tapEvent f stream = createEventCombinator event stream where
156 event :: Sink a -> Time -> a -> Unit
157 event (Sink sink) time value = sink.event time (always value (unsafePerformEff (f value)))
159tapEnd :: forall e a. Eff e Unit -> Stream a -> Stream a
160tapEnd f stream = createEndCombinator end stream where
161 end :: Sink a -> Time -> Unit
162 end (Sink sink) time = sink.end (always time (unsafePerformEff f))
164take :: forall a. Int -> Stream a -> Stream a
165take 0 stream = empty
166take amount (Stream stream) = Stream { run } where
167 run (Sink sink) scheduler = disposable where
168 disposable = stream.run (Sink { event, end }) scheduler where
169 state :: State Int
170 state = createState amount
172 event ::Time -> a -> Unit
173 event time value = do
174 let amountLeft = (getState state)
175 if amountLeft == 0
176 then unit
177 else do
178 let newAmountLeft = state.set \x -> x - 1
179 let x = sink.event time value
180 if newAmountLeft == 0
181 then sink.end (always time (runPure disposable.dispose))
182 else unit
184 end time = sink.end time
186skip :: forall a. Int -> Stream a -> Stream a
187skip 0 stream = stream
188skip amount stream = createEventCombinator event stream where
189 state :: State Int
190 state = createState amount
192 event :: Sink a -> Time -> a -> Unit
193 event (Sink sink) time value = do
194 let amountLeft = runPure state.get
195 if (amountLeft == 0)
196 then sink.event time value
197 else always unit (state.set \x -> x - 1)
199mapStream :: forall a b. (a -> b) -> Stream a -> Stream b
200mapStream f stream = createEventCombinator event stream where
201 event :: Sink b -> Time -> a -> Unit
202 event (Sink sink) time value = sink.event time (f value)
204filter :: forall a. (a -> Boolean) -> Stream a -> Stream a
205filter predicate stream = createEventCombinator event stream where
206 event (Sink sink) time value = if predicate value
207 then sink.event time value
208 else unit
210skipRepeatsWith :: forall a. (a -> a -> Boolean) -> Stream a -> Stream a
211skipRepeatsWith compare stream = createEventCombinator event stream where
212 state :: State (Maybe a)
213 state = createState Nothing
215 event (Sink sink) time value = do
216 let currentState = (getState state)
217 if (isNothing currentState)
218 then sink.event time value
219 else do
220 let currentValue = unsafePartial (fromJust currentState)
221 if (compare currentValue value)
222 then unit
223 else sink.event time value
225skipRepeats :: forall a. (Eq a) => Stream a -> Stream a
226skipRepeats = skipRepeatsWith (eq)
228constant :: forall a b. b -> Stream a -> Stream b
229constant b stream = createEventCombinator event stream where
230 event :: Sink b -> Time -> a -> Unit
231 event (Sink sink) time value = sink.event time b
233concat :: forall a. Stream a -> Stream a -> Stream a
234concat s1 s2 = continueWith(\_ -> s2) s1
236startWith :: forall a. a -> Stream a -> Stream a
237startWith seed stream = concat (just seed) stream
239scan :: forall a b. (b -> a -> b) -> b -> Stream a -> Stream b
240scan f initial stream = startWith initial (createEventCombinator event stream) where
241 seed :: State b
242 seed = createState initial
244 event :: Sink b -> Time -> a -> Unit
245 event (Sink sink) time value = sink.event time (seed.set \x -> f x value)
247loop :: forall a b c. (b -> a -> { seed :: b, value :: c }) -> b -> Stream a -> Stream c
248loop f initial stream = createEventCombinator event stream where
249 seed :: State b
250 seed = createState initial
252 event :: Sink c -> Time -> a -> Unit
253 event (Sink sink) time x = do
254 let seedValue = f (getState seed) x
255 sink.event time (always seedValue.value (seed.set \_ -> seedValue.seed))
257concatMap :: forall a b. (a -> Stream b) -> Stream a -> Stream b
258concatMap f stream = mergeMapConcurrently f 1 stream
260ap :: forall a b. (Stream (a -> b)) -> Stream a -> Stream b
261ap fs xs = combine2 (\f x -> f x) fs xs
263during :: forall a b. Stream (Stream b) -> Stream a -> Stream a
264during timeWindow stream = until (join timeWindow) (since timeWindow stream)
266sample :: forall a b c. (a -> b -> c) -> Stream a -> Stream b -> Stream c
267sample f (Stream sampler) (Stream stream) = Stream { run: runSample (createState Nothing) } where
268 runSample :: State (Maybe b) -> Sink c -> Scheduler -> Disposable
269 runSample state sink scheduler = disposeAll
270 [ stream.run (createHoldSink state) scheduler
271 , sampler.run (createSampleSink state sink f) scheduler
272 ]
274createHoldSink :: forall a. State (Maybe a) -> Sink a
275createHoldSink state = Sink { event, end } where
276 event time value = always unit (state.set \_ -> Just value)
277 end time = unit
279createSampleSink :: forall a b c. State (Maybe b) -> Sink c -> (a -> b -> c) -> Sink a
280createSampleSink state (Sink sink) f = Sink { event, end } where
281 end time = sink.end time
282 event time value =
283 if isNothing (runPure state.get)
284 then unit
285 else sink.event time (f value (unsafePartial (fromJust (runPure state.get))))
287sampleWith :: forall a b. Stream b -> Stream a -> Stream a
288sampleWith signal stream = sample (\_ y -> y) signal stream
290-- find a better way to perform these side effects
291always :: forall a b. a -> b -> a
292always a b = a
294type State a = { get:: Pure a, set :: (a -> a) -> a }
296getState :: forall a. State a -> a
297getState state = runPure state.get
299createState :: forall a. a -> State a
300createState seed = { set, get }
301 where
302 ref :: Ref a
303 ref = unsafePerformEff (unsafeCoerceEff (newRef seed))
305 set :: (a -> a) -> a
306 set f = unsafePerformEff (unsafeCoerceEff (modifyRef' ref \x -> { state: (f x), value: (f x) }))
308 get :: Pure a
309 get = unsafeCoerceEff (readRef ref)
311foreign import drain :: forall e a. Stream a -> EffStream e Unit
312foreign import subscribe :: forall e a. (a -> Eff e Unit) -> (Unit -> Eff e Unit) -> Stream a -> EffStream e Unit
313foreign import fromArray :: forall a b. Array a -> Stream b -> Stream a
314foreign import chain :: forall a b. Stream a -> (a -> Stream b) -> Stream b
315foreign import mergeConcurrently :: forall a. Int -> Stream (Stream a) -> Stream a
316foreign import join :: forall a. Stream (Stream a) -> Stream a
317foreign import mergeMapConcurrently :: forall a b. (a -> Stream b) -> Int -> Stream a -> Stream b
318foreign import switch :: forall a. Stream (Stream a) -> Stream a
319foreign import continueWith :: forall a. (Unit -> Stream a) -> Stream a -> Stream a
320foreign import combine2 :: forall a b c. (a -> b -> c) -> Stream a -> Stream b -> Stream c
321foreign import combine3 :: forall a b c d. (a -> b -> c -> d) -> Stream a -> Stream b -> Stream c -> Stream d
322foreign import combine4 :: forall a b c d e. (a -> b -> c -> d -> e) -> Stream a -> Stream b -> Stream c -> Stream d -> Stream e
323foreign 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
324foreign import combineArray :: forall a. Array (Stream a) -> Stream (Array a)
325foreign import merge :: forall a. Array (Stream a) -> Stream a
326foreign import until :: forall a b. Stream b -> Stream a -> Stream a
327foreign import since :: forall a b. Stream b -> Stream a -> Stream a
328foreign import debounce :: forall a. Number -> Stream a -> Stream a
329foreign import throttle :: forall a. Number -> Stream a -> Stream a
330foreign import delay :: forall a. Number -> Stream a -> Stream a
331foreign import multicast :: forall a. Stream a -> Stream a
332foreign import hold :: forall a. Stream a -> Stream a
333foreign import proxy :: forall a. a -> { attach :: (Stream a -> Stream a), stream :: Stream a }
335-- Sad subject types :/
336foreign import toSubject :: forall a. Stream a -> Subject a
337foreign import toHoldSubject :: forall a. Int -> Stream a -> Subject a
338foreign import fromSubject :: forall a. Subject a -> Stream a
339foreign import subjectEvent :: forall a. a -> Subject a -> Subject a
340foreign import subjectEnd :: forall a. a -> Subject a -> Subject a