syndicate-lang
/
hop-2012
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Clean up Subscription API a little

This commit is contained in:
Tony Garnock-Jones 2012-03-05 16:56:28 -05:00
parent dbe0124314
commit 3868235d92
3 changed files with 13 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
directnode.ml
View File

@ -14,7 +14,7 @@ let classname = "direct"
let unsubscribe info uuid = let unsubscribe info uuid =
Util.with_mutex0 info.mtx Util.with_mutex0 info.mtx
(fun () -> (fun () ->
match Subscription.delete info.name info.subscriptions uuid with match Subscription.delete info.subscriptions uuid with
| Some sub -> | Some sub ->
(match sub.Subscription.filter with (match sub.Subscription.filter with
| Str binding_key -> | Str binding_key ->
@ -38,17 +38,15 @@ let route_message info n sexp =
(fun (uuid) -> (fun (uuid) ->
match Subscription.lookup info.subscriptions uuid with match Subscription.lookup info.subscriptions uuid with
| Some sub -> | Some sub ->
ignore (Subscription.send_to_subscription' info.name info.subscriptions sub body ignore (Subscription.send_to_subscription' sub body (unsubscribe info))
(unsubscribe info)) | None ->
| None -> ()) ())
matching matching
| Message.Subscribe (Str binding_key as filter, Str sink, name, Str reply_sink, reply_name) -> | Message.Subscribe (Str binding_key as filter, Str sink, name, Str reply_sink, reply_name) ->
Util.with_mutex0 info.mtx Util.with_mutex0 info.mtx
(fun () -> (fun () ->
let sub = let sub =
Subscription.create info.name info.subscriptions Subscription.create info.subscriptions filter sink name reply_sink reply_name in
filter sink name reply_sink reply_name
in
let old_set = let old_set =
(try StringMap.find binding_key info.routing_table with Not_found -> UuidSet.empty) in (try StringMap.find binding_key info.routing_table with Not_found -> UuidSet.empty) in
let new_set = UuidSet.add sub.Subscription.uuid old_set in let new_set = UuidSet.add sub.Subscription.uuid old_set in

8
queuenode.ml
View File

@ -29,7 +29,7 @@ let rec do_burst info n =
let (body, new_backlog) = Fqueue.really_pop_front info.backlog in let (body, new_backlog) = Fqueue.really_pop_front info.backlog in
let (sub, new_waiters) = Fqueue.really_pop_front info.waiters in let (sub, new_waiters) = Fqueue.really_pop_front info.waiters in
info.waiters <- new_waiters; info.waiters <- new_waiters;
if Subscription.send_to_subscription info.name info.subscriptions sub body if Subscription.send_to_subscription info.subscriptions sub body
then then
(info.waiters <- Fqueue.push_back info.waiters sub; (info.waiters <- Fqueue.push_back info.waiters sub;
info.backlog <- new_backlog; info.backlog <- new_backlog;
@ -53,13 +53,11 @@ let shoveller info =
loop () loop ()
| Message.Subscribe (filter, Str sink, name, Str reply_sink, reply_name) -> | Message.Subscribe (filter, Str sink, name, Str reply_sink, reply_name) ->
let sub = let sub =
Subscription.create info.name info.subscriptions Subscription.create info.subscriptions filter sink name reply_sink reply_name in
filter sink name reply_sink reply_name
in
info.waiters <- Fqueue.push_back info.waiters sub; info.waiters <- Fqueue.push_back info.waiters sub;
loop () loop ()
| Message.Unsubscribe (Str token) -> | Message.Unsubscribe (Str token) ->
ignore (Subscription.delete info.name info.subscriptions token); ignore (Subscription.delete info.subscriptions token);
loop () loop ()
| m -> | m ->
Util.message_not_understood "queue" m; Util.message_not_understood "queue" m;

10
subscription.ml
View File

@ -12,7 +12,7 @@ type set_t = t StringMap.t ref
let new_set () = ref StringMap.empty let new_set () = ref StringMap.empty
let create source subs filter sink name reply_sink reply_name = let create subs filter sink name reply_sink reply_name =
let uuid = Uuid.create () in let uuid = Uuid.create () in
let sub = { let sub = {
live = true; live = true;
@ -25,7 +25,7 @@ let create source subs filter sink name reply_sink reply_name =
Node.post_ignore reply_sink reply_name (Message.subscribe_ok (Sexp.Str uuid)) (Sexp.Str ""); Node.post_ignore reply_sink reply_name (Message.subscribe_ok (Sexp.Str uuid)) (Sexp.Str "");
sub sub
let delete source subs uuid = let delete subs uuid =
try try
let sub = StringMap.find uuid !subs in let sub = StringMap.find uuid !subs in
sub.live <- false; sub.live <- false;
@ -38,7 +38,7 @@ let lookup subs uuid =
try Some (StringMap.find uuid !subs) try Some (StringMap.find uuid !subs)
with Not_found -> None with Not_found -> None
let send_to_subscription' source subs sub body delete_action = let send_to_subscription' sub body delete_action =
if not sub.live if not sub.live
then false then false
else else
@ -46,5 +46,5 @@ let send_to_subscription' source subs sub body delete_action =
then true then true
else (delete_action sub.uuid; false) else (delete_action sub.uuid; false)
let send_to_subscription source subs sub body = let send_to_subscription subs sub body =
send_to_subscription' source subs sub body (fun (uuid) -> delete source subs uuid) send_to_subscription' sub body (fun (uuid) -> delete subs uuid)