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

Rearrange message dispatch procedures slightly

This commit is contained in:
Tony Garnock-Jones 2011-01-02 22:08:30 -05:00
parent 0cdf9f6e68
commit d09902fb06
4 changed files with 49 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
direct.c
View File

@ -87,20 +87,27 @@ static void direct_handle_message(node_t *n, sexp_t *m) {
} else {
warn("Non-string routing key in direct\n");
}
} else if ((msglen == 6) && !cmsg_bytes_cmp(selector, cmsg_cstring_bytes("subscribe"))) {
return;
}
if ((msglen == 6) && !cmsg_bytes_cmp(selector, cmsg_cstring_bytes("subscribe"))) {
subscription_t *sub = handle_subscribe_message(&d->subscriptions, args);
if (sub != NULL) {
sexp_t *filter = sexp_listref(args, 0);
hashtable_get(&d->routing_table, sexp_data(filter), (void **) &sub->link);
hashtable_put(&d->routing_table, sexp_data(filter), sub);
}
} else if ((msglen == 2) && !cmsg_bytes_cmp(selector, cmsg_cstring_bytes("unsubscribe"))) {
handle_unsubscribe_message(&d->subscriptions, args);
} else {
warn("Message not understood in direct; selector <<%.*s>>, length %u\n",
selector.len, selector.bytes,
msglen);
return;
}
if ((msglen == 2) && !cmsg_bytes_cmp(selector, cmsg_cstring_bytes("unsubscribe"))) {
handle_unsubscribe_message(&d->subscriptions, args);
return;
}
warn("Message not understood in direct; selector <<%.*s>>, length %u\n",
selector.len, selector.bytes,
msglen);
}
static node_class_t direct_class = {

23
fanout.c
View File

@ -78,15 +78,22 @@ static void fanout_handle_message(node_t *n, sexp_t *m) {
context.f = f;
context.body = sexp_listref(args, 1);
hashtable_foreach(&f->subscriptions, send_to_sub, &context);
} else if ((msglen == 6) && !cmsg_bytes_cmp(selector, cmsg_cstring_bytes("subscribe"))) {
handle_subscribe_message(&f->subscriptions, args);
} else if ((msglen == 2) && !cmsg_bytes_cmp(selector, cmsg_cstring_bytes("unsubscribe"))) {
handle_unsubscribe_message(&f->subscriptions, args);
} else {
warn("Message not understood in fanout; selector <<%.*s>>, length %u\n",
selector.len, selector.bytes,
msglen);
return;
}
if ((msglen == 6) && !cmsg_bytes_cmp(selector, cmsg_cstring_bytes("subscribe"))) {
handle_subscribe_message(&f->subscriptions, args);
return;
}
if ((msglen == 2) && !cmsg_bytes_cmp(selector, cmsg_cstring_bytes("unsubscribe"))) {
handle_unsubscribe_message(&f->subscriptions, args);
return;
}
warn("Message not understood in fanout; selector <<%.*s>>, length %u\n",
selector.len, selector.bytes,
msglen);
}
static node_class_t fanout_class = {

12
main.c
View File

@ -39,8 +39,7 @@ static void factory_handle_message(node_t *n, sexp_t *m) {
selector = sexp_data(sexp_head(m));
args = sexp_tail(m);
if ((msglen == 5)
&& !cmsg_bytes_cmp(selector, cmsg_cstring_bytes("create"))) {
if ((msglen == 5) && !cmsg_bytes_cmp(selector, cmsg_cstring_bytes("create"))) {
sexp_t *classname = sexp_listref(args, 0);
sexp_t *ctor_arg = sexp_listref(args, 1);
sexp_t *reply_sink = sexp_listref(args, 2);
@ -63,11 +62,12 @@ static void factory_handle_message(node_t *n, sexp_t *m) {
DECREF(reply, sexp_destructor);
}
}
} else {
warn("Message not understood in factory; selector <<%.*s>>, length %u\n",
selector.len, selector.bytes,
msglen);
return;
}
warn("Message not understood in factory; selector <<%.*s>>, length %u\n",
selector.len, selector.bytes,
msglen);
}
static node_class_t factory_class = {

21
queue.c
View File

@ -165,19 +165,26 @@ static void queue_handle_message(node_t *n, sexp_t *m) {
if ((msglen == 4) && !cmsg_bytes_cmp(selector, cmsg_cstring_bytes("post"))) {
sexp_enqueue(q->backlog_q, sexp_listref(args, 1));
throck_shovel(q);
} else if ((msglen == 6) && !cmsg_bytes_cmp(selector, cmsg_cstring_bytes("subscribe"))) {
return;
}
if ((msglen == 6) && !cmsg_bytes_cmp(selector, cmsg_cstring_bytes("subscribe"))) {
subscription_t *sub = handle_subscribe_message(&q->subscriptions, args);
if (sub != NULL) {
enqueue(&q->waiter_q, sub);
throck_shovel(q);
}
} else if ((msglen == 2) && !cmsg_bytes_cmp(selector, cmsg_cstring_bytes("unsubscribe"))) {
handle_unsubscribe_message(&q->subscriptions, args);
} else {
warn("Message not understood in queue; selector <<%.*s>>, length %u\n",
selector.len, selector.bytes,
msglen);
return;
}
if ((msglen == 2) && !cmsg_bytes_cmp(selector, cmsg_cstring_bytes("unsubscribe"))) {
handle_unsubscribe_message(&q->subscriptions, args);
return;
}
warn("Message not understood in queue; selector <<%.*s>>, length %u\n",
selector.len, selector.bytes,
msglen);
}
static node_class_t queue_class = {