@@ -0,0 +1,371 @@
5+++ mk-subs :: Create sub-map.
6+ |* [=(lake) paths=mold]
9+++ mk-pubs :: Create pub-map.
10+ |* [=(lake) paths=mold]
13+++ mk-mar :: Create mar.
15+ |_ =(response:poke lake *)
22+ ++ noun (response:poke lake *)
26+++ fled :: Like +sped but head is a path.
31+ =- [%wtgr [%wtts - [%& 2]~] [%$ 1]]
32+ =/ pax ~| %path-none ;;(path -.q.vax)
35+ [%bccl ~[[%leaf %ta -.pax] $(pax +.pax)]]
37+++ zoom |= =noun ~| %need-path $/sss/;;(path noun)
39+++ da :: Manage subscriptions.
40+ |* [=(lake) paths=mold]
43+ +$ from (on-rock:poke lake paths)
44+ +$ into (response:poke lake paths)
45+ +$ result (request:poke paths)
46+ +$ fail [paths ship dude]
47+ +$ flow [=aeon stale=_| fail=_| =rock:lake]
48+ +$ subs [%0 (map [ship dude paths] (unit flow))]
58+ ++ surf :: Subscribe to [ship dude path].
59+ |= which=[ship dude paths]
60+ ^- (quip card:agent:gall subs)
61+ ?+ flow=(~(get by sub) which) `0/sub
62+ ~ [~[(pine which)] 0/(~(put by sub) which ~)]
63+ [~ ~] [~[(pine which)] 0/sub]
64+ [~ ~ [* %& * *]] [~[(scry `+(aeon.u.u.flow) which)] 0/sub]
66+ ++ quit (corl (lead %0) ~(del by sub)) :: Unsub from [ship dude path].
67+ ++ read :: See current subscribed states.
68+ ^- (map [ship dude paths] [stale=? fail=? =rock:lake])
69+ %- malt %+ murn ~(tap by sub)
70+ |= [key=[ship dude paths] val=(unit flow)]
73+ :: :: Check poke-ack for errors.
74+ :: :: If an %sss-on-rock poke nacks,
75+ ++ chit :: that state is flagged as failed.
76+ |= [[aeon=term ship=term dude=term path=paths] =sign:agent:gall]
79+ ?> ?=(%poke-ack -.sign)
81+ %+ ~(jab by sub) [(slav %p ship) dude path]
84+ ?> =(aeon.flow (slav %ud aeon))
86+ :: :: Check poke-ack for errors.
87+ :: :: If a scry request nacks,
88+ ++ tell :: that state is flagged as stale.
89+ |= [[ship=term =dude aeon=term path=paths] =sign:agent:gall]
90+ ^- (quip card:agent:gall subs)
91+ ?> ?=(%poke-ack -.sign)
93+ =/ current [ship=(slav %p ship) dude=dude path=path]
94+ ?+ flow=(~(get by sub) current) `0/sub
97+ :_ 0/(~(put by sub) current u.flow)
98+ ~[(on-rock-poke current u.u.flow ~)]
101+ :_ 0/(~(del by sub) current) :_ ~
102+ :* %pass (zoom surf-fail/aeon/ship/dude/path)
103+ %agent [our dap]:bowl
104+ %poke %sss-surf-fail fail-type ^- fail
105+ [path ship dude]:current
108+ :: :: Check if we're still interested
109+ :: :: in a wave. If no, no-op.
111+ ++ behn :: (See https://gist.github.com/belisarius222/7f8452bfea9b199c0ed717ab1778f35b)
112+ |= [ship=term =dude aeon=term path=paths]
113+ ^- (list card:agent:gall)
114+ %- fall :_ ~ %- mole |.
115+ =/ ship (slav %p ship)
116+ =/ aeon (slav %ud aeon)
117+ ?: (lte aeon aeon:(fall (~(got by sub) ship dude path) *flow)) ~
118+ ~[(scry `aeon ship dude path)]
120+ ++ apply :: Handle response from publisher.
121+ |= res=(response:poke lake paths)
122+ ^- (quip card:agent:gall subs)
123+ %- fall :_ `0/sub %- mole |.
124+ =* current [src.bowl dude.res path.res]
125+ =/ old=flow (fall (~(got by sub) current) *flow)
128+ =/ =flow old(stale &)
129+ :_ 0/(~(put by sub) current `flow) :_ ~
130+ (on-rock-poke current flow ~)
134+ (pine src.bowl dude.res path.res)
138+ (behn-s25 [dude aeon path]:res)
141+ =/ [wave=(unit wave:lake) =flow]
143+ %rock ?> (gte aeon.res aeon.old)
144+ [~ [aeon.res | | rock.res]]
145+ %wave ?> =(aeon.res +(aeon.old))
146+ [`wave.res [aeon.res | | (wash:lake rock.old wave.res)]]
148+ :_ 0/(~(put by sub) current `flow)
149+ :~ (on-rock-poke current flow wave)
150+ (scry `+(aeon.res) src.bowl dude.res path.res)
154+ :: Non-public facing arms below
157+ |= [=dude =aeon path=noun]
159+ :* %pass (zoom behn/(scot %p src.bowl)^dude^(scot %ud aeon)^path)
160+ %arvo %b %wait (add ~s25 now.bowl)
162+ ++ pine |= [ship dude paths] (scry ~ +<)
164+ |= [when=(unit aeon) who=ship which=dude where=paths]
166+ =/ when ?~ when ~ (scot %ud u.when)
167+ :* %pass (zoom scry-request/(scot %p who)^which^when^where)
169+ %poke %sss-to-pub :- result-type ^- result
170+ [where dap.bowl ^when]
173+ |= [[=ship =dude path=paths] flow wave=(unit wave:lake)]
175+ :* %pass (zoom on-rock/(scot %ud aeon)^(scot %p ship)^dude^path)
176+ %agent [our dap]:bowl
177+ %poke %sss-on-rock on-rock-type ^- from
178+ [path ship dude stale fail rock wave]
181+++ du :: Manage publications.
182+ |* [=(lake) paths=mold]
185+ +$ into (request:poke paths)
186+ +$ result (response:poke lake paths)
187+ +$ rule [rocks=_1 waves=_5] :: Retention policy
189+ $: rok=((mop aeon rock:lake) gte)
190+ wav=((mop aeon wave:lake) lte)
192+ mem=(mip ship dude @da)
195+ $: tid=$~(*tide $@(aeon tide))
196+ alo=(unit (set ship))
198+ +$ pubs [%0 (map paths buoy)]
200+ |= [pub=pubs =bowl:gall result-type=type]
202+ =* rok ((on aeon rock:lake) gte)
203+ =* wav ((on aeon wave:lake) lte)
205+ ++ rule :: Set new retention policy.
206+ |= [path=paths =^rule]
209+ %+ ~(jab by pub) path
212+ buoy(tid (form tid.buoy(rul rule)))
214+ ++ wipe :: Create new rock and wipe rest.
218+ %+ ~(jab by pub) path
221+ %* . buoy(tid (form tid.buoy(rul [0 1])))
222+ rul.tid rul.tid.buoy
225+ ++ give :: Give a wave on a path.
226+ |= [path=paths =wave:lake]
227+ ^- (quip card:agent:gall pubs)
228+ ?~ ((soft ^path) path) ~| %need-path !!
229+ =/ buoy (~(gut by pub) path *buoy)
230+ ?@ tide=tid.buoy ~| %dead-path !! ::TODO is this good behavior?
231+ =/ next=aeon +((latest tide))
232+ :- %+ murn ~(tap bi mem.tide)
233+ |= [=ship =dude =@da]
234+ ?: (lth da now.bowl) ~
235+ `(send scry/wave/wave ship dude next path)
237+ %+ ~(put by pub) path
238+ =/ last=[=aeon =rock:lake] (fall (pry:rok rok.tide) *[key val]:rok)
239+ =. wav.tide (put:wav wav.tide next wave)
240+ =. mem.tide (~(del by mem.tide) next)
241+ ?. =(next (add aeon.last waves.rul.tide)) buoy
242+ buoy(tid (form tide))
244+ ++ perm :: Change permissions with gate.
245+ |= [where=(list paths) diff=$-((unit (set ship)) (unit (set ship)))]
249+ =/ new=_alo.buoy (diff alo.buoy)
250+ ?@ tid.buoy buoy(alo new)
253+ mem.tid ?~ new mem.tid.buoy
255+ ~(int by (malt (turn ~(tap in u.new) (late *(map @ @)))))
257+ ++ public (curr perm _~) :: Make list of paths public.
258+ ++ secret (curr perm _`~) :: Make list of paths secret.
259+ :: :: Block ships from paths.
260+ ++ block :: No-ops on public paths.
261+ |= [who=(list ship) whence=(list paths)]
264+ |= old=(unit (set ship))
265+ ?~ old ~ `(~(dif in u.old) (sy who))
266+ :: :: Allow ships to paths.
267+ ++ allow :: Any public paths will no-op.
268+ |= [who=(list ship) where=(list paths)]
271+ |= old=(unit (set ship))
272+ ?~ old ~ `(~(gas in u.old) who)
273+ :: :: Kill a list of paths, i.e. tell
274+ ++ kill :: subs to not expect updates.
275+ (curr edit |=(=buoy buoy(tid (latest tid.buoy))))
276+ :: :: Reopen list of killed paths.
277+ ++ live :: No-ops on live paths.
281+ %*(. buoy(tid *tide) rok.tid (put:rok ~ +(tid.buoy) *rock:lake))
283+ ++ read :: See current published states.
284+ ^- (map paths [allowed=(unit (set ship)) =rock:lake])
285+ %- malt %+ murn ~(tap by pub)
286+ |= [path=paths =buoy]
287+ ^- (unit [paths (unit (set ship)) rock:lake])
289+ :^ ~ path alo.buoy =< rock
290+ =/ snap=[=aeon =rock:lake] (fall (pry:rok rok.tide) *[key val]:rok)
291+ %+ roll (tap:wav (lot:wav wav.tide `aeon.snap ~))
292+ |= [[=aeon =wave:lake] =_snap]
293+ ?. =(aeon +(aeon.snap)) snap
294+ [aeon (wash:lake rock.snap wave)]
296+ ++ apply :: Handle request from subscriber.
297+ |= req=(request:poke paths)
298+ ^- (quip card:agent:gall pubs)
299+ =/ =buoy (~(gut by pub) path.req *buoy)
300+ ?< &(?=(^ alo.buoy) !(~(has in u.alo.buoy) src.bowl))
303+ (send tomb/~ src.bowl dude.req tid.buoy path.req)
305+ =/ last (fall (pry:rok rok.tid.buoy) *[=key =val]:rok)
307+ (send scry/rock/val.last src.bowl dude.req key.last path.req)
308+ ?^ dat=(get:wav wav.tid.buoy u.when.req)
310+ (send scry/wave/u.dat src.bowl [dude u.when path]:req)
311+ ?: %+ lte u.when.req
312+ key::(fall (ram:wav wav.tid.buoy) (pry:rok rok.tid.buoy) [=key val]:wav)
314+ (send yore/~ src.bowl [dude u.when path]:req)
315+ ?> =(u.when.req +((latest tid.buoy)))
316+ :- ~[(send nigh/~ src.bowl [dude u.when path]:req)]
318+ %+ ~(put by pub) path.req
320+ mem.tid (~(put bi mem.tid.buoy) src.bowl dude.req (add ~s25 now.bowl))
323+ :: Non-public facing arms below
326+ |= [payload=_|3:*(response:poke lake paths) =ship =dude =aeon path=paths]
328+ =* mark (cat 3 %sss- name:lake)
329+ :* %pass (zoom scry-response/(scot %p ship)^dude^(scot %ud aeon)^path)
331+ %poke mark result-type ^- (response:poke lake paths)
332+ [path dap.bowl aeon payload]
338+ %+ max (fall (bind (pry:rok rok.tide) head) 0)
339+ (fall (bind (ram:wav wav.tide) head) 0)
342+ |= [ps=(list paths) edit=$-(buoy buoy)]
345+ %- ~(rep in (sy ps))
346+ |= [path=paths =_pub]
347+ %- fall :_ pub %- mole |.
348+ (~(jab by pub) path edit)
353+ =/ max-rock=[=aeon =rock:lake] (fall (pry:rok rok.tide) *[key val]:rok)
354+ =/ max-wave (fall (bind (ram:wav wav.tide) head) 0)
356+ %+ gas:rok +<-:gas:rok
357+ %- tab:rok :_ [~ +(rocks.rul.tide)]
358+ ?: ?| =(waves.rul.tide 0)
359+ (lth max-wave (add aeon.max-rock waves.rul.tide))
363+ %+ roll (tab:wav wav.tide `aeon.max-rock max-wave)
364+ |: [*[now=aeon =wave:lake] `[prev=aeon =rock:lake]`max-rock]
367+ [now (wash:lake rock wave)]
369+ tide(wav (lot:wav wav.tide (bind (ram:rok rok.tide) |=([r=@ *] (dec r))) ~))