blob: caa1173c2af6d2b09dc1000e998f41eeca096bf6 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
|
(.module:
lux
(lux (control [functor #+ Functor]
[apply #+ Apply]
[monad #+ do Monad])
[io #+ IO io]
(data (coll [list "list/" Monoid<List>]))
(type abstract))
(// [atom #+ Atom atom]
[promise #+ Promise]))
## [Types]
(abstract: #export (Channel a)
{#.doc "An asynchronous channel to distribute values."}
(Atom (List (-> a (IO Top))))
(def: #export (channel _)
(All [a] (-> Top (Channel a)))
(@abstraction (atom (list))))
(def: #export (listen listener (^@representation channel))
(All [a] (-> (-> a (IO Top)) (Channel a) (IO Top)))
## TODO: Simplify when possible.
(do io.Monad<IO>
[_ (atom.update (|>> (#.Cons listener)) channel)]
(wrap [])))
(def: #export (publish (^@representation channel) value)
{#.doc "Publish to a channel."}
(All [a] (-> (Channel a) a (IO Top)))
(do io.Monad<IO>
[listeners (atom.read channel)]
(monad.map @ (function (_ listener) (listener value)) listeners)))
)
## [Values]
(def: #export (filter predicate input)
(All [a] (-> (-> a Bool) (Channel a) (Channel a)))
(let [output (channel [])]
(exec (io.run (listen (function (_ value)
(if (predicate value)
(publish output value)
(io [])))
input))
output)))
(def: #export (pipe output input)
{#.doc "Copy/pipe the contents of a channel on to another."}
(All [a] (-> (Channel a) (Channel a) (IO Top)))
(listen (publish output)
input))
(def: #export (merge inputs)
{#.doc "Fuse all the elements in a list of channels by piping them onto a new output channel."}
(All [a] (-> (List (Channel a)) (IO (Channel a))))
(let [output (channel [])]
(do io.Monad<IO>
[_ (monad.map @ (pipe output) inputs)]
(wrap output))))
(def: #export (from-promise promise)
(All [a] (-> (Promise a) (Channel a)))
(let [output (channel [])]
(exec (promise.await (publish output) promise)
output)))
(def: #export (poll time action)
(All [a] (-> Nat (IO a) (Channel a)))
(let [output (channel [])]
(exec (io.run
(loop [_ []]
(do io.Monad<IO>
[value action
_ (publish output value)]
(wrap (promise.await recur (promise.wait time))))))
output)))
(def: #export (periodic time)
(-> Nat (Channel Top))
(let [output (channel [])]
(exec (io.run
(loop [_ []]
(do io.Monad<IO>
[_ (publish output [])]
(wrap (promise.await recur (promise.wait time))))))
output)))
(def: #export (iterate f init)
(All [a] (-> (-> a (Promise a)) a (Channel a)))
(let [output (channel [])]
(exec (io.run
(loop [zero init]
(do io.Monad<IO>
[_ (publish output zero)]
(wrap (promise.await recur (f zero))))))
output)))
## [Structures]
(struct: #export _ (Functor Channel)
(def: (map f input)
(let [output (channel [])]
(exec (io.run (listen (|>> f (publish output))
input))
output))))
(struct: #export _ (Apply Channel)
(def: functor Functor<Channel>)
(def: (apply ff fa)
(let [output (channel [])]
(exec (io.run (listen (function (_ f)
(listen (|>> f (publish output))
fa))
ff))
output))))
(struct: #export _ (Monad Channel)
(def: functor Functor<Channel>)
(def: (wrap a)
(let [output (channel [])]
(exec (io.run (publish output a))
output)))
(def: (join mma)
(let [output (channel [])]
(exec (io.run (listen (listen (publish output))
mma))
output))))
|
