git.iko.soy/purescript-stream
on 3773a5e help sign in
purescript-stream/src/Control/Stream/Stream.purs
339 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 , take
17 , skip
18 , constant
19 , continueWith
20 , concat
21 , startWith
22 , scan
23 , loop
24 , chain
25 , mergeConcurrently
26 , join
27 , mergeMapConcurrently
28 , concatMap
29 , switch
30 , combine2
31 , combine3
32 , combine4
33 , combine5
34 , combineArray
35 , merge
36 , skipRepeatsWith
37 , skipRepeats
38 , until
39 , since
40 , during
41 , sample
42 , debounce
43 , throttle
44 , delay
45 , multicast
46 , hold
47 , Subject
48 , toSubject
49 , toHoldSubject
50 , fromSubject
51 , subjectEvent
52 , subjectEnd
53 , subscribe
54 ) where
56import Control.Alternative (class Apply)
57import Control.Bind (class Bind)
58import Control.Category (id)
59import Control.Monad.Eff (Eff, Pure, runPure)
60import Control.Monad.Eff.Ref (Ref, modifyRef', newRef, readRef)
61import Control.Monad.Eff.Unsafe (unsafeCoerceEff, unsafePerformEff)
62import Control.MonadPlus (class Alt, class Alternative, class MonadPlus, class Plus)
63import Control.MonadZero (class Applicative, class MonadZero)
64import Control.Stream.Disposable (Disposable, emptyDisposable, disposeAll)
65import Control.Stream.Scheduler (Scheduler(..), endTask, eventTask, scheduleTasks)
66import Control.Stream.Sink (Sink(..))
67import Control.Stream.Stream (Stream(..), EffStream, empty, just)
68import Control.Stream.Time (Time)
69import Data.Eq ((==))
70import Data.EuclideanRing ((-))
71import Data.Functor (class Functor)
72import Data.Maybe (Maybe(..), fromJust, isNothing)
73import Data.Monoid (class Monoid)
74import Data.Unit (Unit, unit)
75import Partial.Unsafe (unsafePartial)
76import Prelude (class Eq, class Monad, class Semigroup, eq)
78foreign import data STREAM :: !
80type EffStream e a = Eff (stream :: STREAM | e) a
82foreign import data Subject :: * -> *
84newtype Stream a = Stream { run :: Sink a -> Scheduler -> Disposable }
86instance functorStream :: Functor Stream where
87 map = mapStream
89instance applyStream :: Apply Stream where
90 apply = combine2 id
92instance applicativeStream :: Applicative Stream where
93 pure = just
95instance bindStream :: Bind Stream where
96 bind = chain
98instance monadStream :: Monad Stream
100instance semigroupStream :: Semigroup (Stream a) where
101 append = concat
103instance altStream :: Alt Stream where
104 alt = \s1 s2 -> merge [s1, s2]
106instance monoidStream :: Monoid (Stream a) where
107 mempty = empty
109instance plusStream :: Plus Stream where
110 empty = empty
112instance alternativeStream :: Alternative Stream
114instance monadZeroStream :: MonadZero Stream
116instance monadPlusStream :: MonadPlus Stream
118just :: forall a. a -> Stream a
119just a = Stream { run } where
120 run :: Sink a -> Scheduler -> Disposable
121 run sink (Scheduler scheduler) =
122 scheduleTasks scheduler.asap [ (eventTask sink a), (endTask sink) ]
124empty :: forall a. Stream a
125empty = Stream { run } where
126 run :: Sink a -> Scheduler -> Disposable
127 run sink (Scheduler scheduler) = scheduleTasks scheduler.asap [ endTask sink ]
129never :: forall a. Stream a
130never = Stream { run: \sink scheduler -> emptyDisposable }
132periodic :: Number -> Stream Unit
133periodic period = Stream { run } where
134 run :: Sink Unit -> Scheduler -> Disposable
135 run sink (Scheduler scheduler) = do
136 let task = eventTask sink unit
137 let scheduledTask = scheduler.periodic period task
138 { dispose : scheduledTask.dispose }
140createCombinator :: forall a b. (Sink b -> Time -> a -> Unit) -> (Sink b -> Time -> Unit) -> Stream a -> Stream b
141createCombinator event end (Stream stream) = Stream { run } where
142 run :: Sink b -> Scheduler -> Disposable
143 run sink scheduler = stream.run (Sink { event: (event sink), end: (end sink) }) scheduler
145createEventCombinator :: forall a b. (Sink b -> Time -> a -> Unit) -> Stream a -> Stream b
146createEventCombinator event stream = createCombinator event end stream where
147 end (Sink sink) time = sink.end time
149createEndCombinator :: forall a. (Sink a -> Time -> Unit) -> Stream a -> Stream a
150createEndCombinator end stream = createCombinator event end stream where
151 event (Sink sink) time value = sink.event time value
153tapEvent :: forall e a. (a -> Eff e Unit) -> Stream a -> Stream a
154tapEvent f stream = createEventCombinator event stream where
155 event :: Sink a -> Time -> a -> Unit
156 event (Sink sink) time value = sink.event time (always value (unsafePerformEff (f value)))
158tapEnd :: forall e a. Eff e Unit -> Stream a -> Stream a
159tapEnd f stream = createEndCombinator end stream where
160 end :: Sink a -> Time -> Unit
161 end (Sink sink) time = sink.end (always time (unsafePerformEff f))
163take :: forall a. Int -> Stream a -> Stream a
164take 0 stream = empty
165take amount (Stream stream) = Stream { run } where
166 run (Sink sink) scheduler = disposable where
167 disposable = stream.run (Sink { event, end }) scheduler where
168 state :: State Int
169 state = createState amount
171 event ::Time -> a -> Unit
172 event time value = do
173 let amountLeft = (getState state)
174 if amountLeft == 0
175 then unit
176 else do
177 let newAmountLeft = state.set \x -> x - 1
178 let x = sink.event time value
179 if newAmountLeft == 0
180 then sink.end (always time (runPure disposable.dispose))
181 else unit
183 end time = sink.end time
185skip :: forall a. Int -> Stream a -> Stream a
186skip 0 stream = stream
187skip amount stream = createEventCombinator event stream where
188 state :: State Int
189 state = createState amount
191 event :: Sink a -> Time -> a -> Unit
192 event (Sink sink) time value = do
193 let amountLeft = runPure state.get
194 if (amountLeft == 0)
195 then sink.event time value
196 else always unit (state.set \x -> x - 1)
198mapStream :: forall a b. (a -> b) -> Stream a -> Stream b
199mapStream f stream = createEventCombinator event stream where
200 event :: Sink b -> Time -> a -> Unit
201 event (Sink sink) time value = sink.event time (f value)
203filter :: forall a. (a -> Boolean) -> Stream a -> Stream a
204filter predicate stream = createEventCombinator event stream where
205 event (Sink sink) time value = if predicate value
206 then sink.event time value
207 else unit
209skipRepeatsWith :: forall a. (a -> a -> Boolean) -> Stream a -> Stream a
210skipRepeatsWith compare stream = createEventCombinator event stream where
211 state :: State (Maybe a)
212 state = createState Nothing
214 event (Sink sink) time value = do
215 let currentState = (getState state)
216 if (isNothing currentState)
217 then sink.event time value
218 else do
219 let currentValue = unsafePartial (fromJust currentState)
220 if (compare currentValue value)
221 then unit
222 else sink.event time value
224skipRepeats :: forall a. (Eq a) => Stream a -> Stream a
225skipRepeats = skipRepeatsWith (eq)
227constant :: forall a b. b -> Stream a -> Stream b
228constant b stream = createEventCombinator event stream where
229 event :: Sink b -> Time -> a -> Unit
230 event (Sink sink) time value = sink.event time b
232concat :: forall a. Stream a -> Stream a -> Stream a
233concat s1 s2 = continueWith(\_ -> s2) s1
235startWith :: forall a. a -> Stream a -> Stream a
236startWith seed stream = concat (just seed) stream
238scan :: forall a b. (b -> a -> b) -> b -> Stream a -> Stream b
239scan f initial stream = startWith initial (createEventCombinator event stream) where
240 seed :: State b
241 seed = createState initial
243 event :: Sink b -> Time -> a -> Unit
244 event (Sink sink) time value = sink.event time (seed.set \x -> f x value)
246loop :: forall a b c. (b -> a -> { seed :: b, value :: c }) -> b -> Stream a -> Stream c
247loop f initial stream = createEventCombinator event stream where
248 seed :: State b
249 seed = createState initial
251 event :: Sink c -> Time -> a -> Unit
252 event (Sink sink) time x = do
253 let seedValue = f (getState seed) x
254 sink.event time (always seedValue.value (seed.set \_ -> seedValue.seed))
256concatMap :: forall a b. (a -> Stream b) -> Stream a -> Stream b
257concatMap f stream = mergeMapConcurrently f 1 stream
259ap :: forall a b. (Stream (a -> b)) -> Stream a -> Stream b
260ap fs xs = combine2 (\f x -> f x) fs xs
262during :: forall a b. Stream (Stream b) -> Stream a -> Stream a
263during timeWindow stream = until (join timeWindow) (since timeWindow stream)
265sample :: forall a b c. (a -> b -> c) -> Stream a -> Stream b -> Stream c
266sample f (Stream sampler) (Stream stream) = Stream { run: runSample (createState Nothing) } where
267 runSample :: State (Maybe b) -> Sink c -> Scheduler -> Disposable
268 runSample state sink scheduler = disposeAll
269 [ stream.run (createHoldSink state) scheduler
270 , sampler.run (createSampleSink state sink f) scheduler
271 ]
273createHoldSink :: forall a. State (Maybe a) -> Sink a
274createHoldSink state = Sink { event, end } where
275 event time value = always unit (state.set \_ -> Just value)
276 end time = unit
278createSampleSink :: forall a b c. State (Maybe b) -> Sink c -> (a -> b -> c) -> Sink a
279createSampleSink state (Sink sink) f = Sink { event, end } where
280 end time = sink.end time
281 event time value =
282 if isNothing (runPure state.get)
283 then unit
284 else sink.event time (f value (unsafePartial (fromJust (runPure state.get))))
286sampleWith :: forall a b. Stream b -> Stream a -> Stream a
287sampleWith signal stream = sample (\_ y -> y) signal stream
289-- find a better way to perform these side effects
290always :: forall a b. a -> b -> a
291always a b = a
293type State a = { get:: Pure a, set :: (a -> a) -> a }
295getState :: forall a. State a -> a
296getState state = runPure state.get
298createState :: forall a. a -> State a
299createState seed = { set, get }
300 where
301 ref :: Ref a
302 ref = unsafePerformEff (unsafeCoerceEff (newRef seed))
304 set :: (a -> a) -> a
305 set f = unsafePerformEff (unsafeCoerceEff (modifyRef' ref \x -> { state: (f x), value: (f x) }))
307 get :: Pure a
308 get = unsafeCoerceEff (readRef ref)
310foreign import drain :: forall e a. Stream a -> EffStream e Unit
311foreign import subscribe :: forall e a. (a -> Eff e Unit) -> (Unit -> Eff e Unit) -> Stream a -> EffStream e Unit
312foreign import fromArray :: forall a b. Array a -> Stream b -> Stream a
313foreign import chain :: forall a b. Stream a -> (a -> Stream b) -> Stream b
314foreign import mergeConcurrently :: forall a. Int -> Stream (Stream a) -> Stream a
315foreign import join :: forall a. Stream (Stream a) -> Stream a
316foreign import mergeMapConcurrently :: forall a b. (a -> Stream b) -> Int -> Stream a -> Stream b
317foreign import switch :: forall a. Stream (Stream a) -> Stream a
318foreign import continueWith :: forall a. (Unit -> Stream a) -> Stream a -> Stream a
319foreign import combine2 :: forall a b c. (a -> b -> c) -> Stream a -> Stream b -> Stream c
320foreign import combine3 :: forall a b c d. (a -> b -> c -> d) -> Stream a -> Stream b -> Stream c -> Stream d
321foreign import combine4 :: forall a b c d e. (a -> b -> c -> d -> e) -> Stream a -> Stream b -> Stream c -> Stream d -> Stream e
322foreign 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
323foreign import combineArray :: forall a. Array (Stream a) -> Stream (Array a)
324foreign import merge :: forall a. Array (Stream a) -> Stream a
325foreign import until :: forall a b. Stream b -> Stream a -> Stream a
326foreign import since :: forall a b. Stream b -> Stream a -> Stream a
327foreign import debounce :: forall a. Number -> Stream a -> Stream a
328foreign import throttle :: forall a. Number -> Stream a -> Stream a
329foreign import delay :: forall a. Number -> Stream a -> Stream a
330foreign import multicast :: forall a. Stream a -> Stream a
331foreign import hold :: forall a. Stream a -> Stream a
332foreign import proxy :: forall a. a -> { attach :: (Stream a -> Stream a), stream :: Stream a }
334-- Sad subject types :/
335foreign import toSubject :: forall a. Stream a -> Subject a
336foreign import toHoldSubject :: forall a. Int -> Stream a -> Subject a
337foreign import fromSubject :: forall a. Subject a -> Stream a
338foreign import subjectEvent :: forall a. a -> Subject a -> Subject a
339foreign import subjectEnd :: forall a. a -> Subject a -> Subject a