Pubsub server #156
Pubsub server #156
Conversation
src/core/data/pubsub.c
Outdated
|
|
||
| if (processor->write(s) < 0) { | ||
| log_debug("handler signals channel termination"); | ||
| s->ch->state = CHANNEL_TERM; |
There was a problem hiding this comment.
c->state = CHANNEL_TERM to be consistent with the one above
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
| struct element el_r = {.type = ELEM_INT}; /* reply, an integer */ | ||
| struct listener *l; | ||
| struct topic *t; | ||
| uint32_t nsub = 0; |
There was a problem hiding this comment.
I believe this would be more readable without this temporary variable and just using el_r.num
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
| log_debug("listener not added"); | ||
| } | ||
| if (!listener_add_topic(l, t)) { | ||
| log_debug("topic not added"); |
There was a problem hiding this comment.
these errors should probably write an error message to the client instead of failing silently?
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
| if (prev == NULL) { | ||
| SLIST_REMOVE_HEAD(bucket, l_sle); | ||
| } else { | ||
| SLIST_REMOVE_AFTER(prev, l_sle); |
There was a problem hiding this comment.
what would happen if s was not on the list? prev'd be the last element and it'll segfault or something? should there be an ASSERT(l != NULL)?
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
| t->name.len = name->len; | ||
| t->name.data = cc_alloc(t->name.len); | ||
| if (t->name.data == NULL) { | ||
| cc_free(t); |
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
|
|
||
| /* | ||
| * a topic is an endpoint that clients can subscribe to, equivalent to | ||
| * "channel" in the original redis protocol. |
There was a problem hiding this comment.
is there any reason to call it topic instead of channel? do we refer to something else as channel? you could add the reason to the comment here
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
| if (!admin_init) { | ||
| log_warn("%s has never been setup", PUBSUB_ADMIN_MODULE_NAME); | ||
| } | ||
|
|
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
| #include <stream/cc_sockio.h> | ||
|
|
||
| /* | ||
| * a listener is a client that has subscribed to at least one channel |
There was a problem hiding this comment.
I thought that a listener may be subscribed to 0 channels (after unsubscribe it is not deleted)?
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
| @@ -10,7 +10,7 @@ | |||
| #include <sysexits.h> | |||
|
|
|||
| void | |||
| core_run(void *arg_worker) | |||
| worker_run(void *arg_worker) | |||
There was a problem hiding this comment.
why are we renaming this? I think this name can cause some confusion, since it creates threads for both worker and server threads.
There was a problem hiding this comment.
maybe rename these to core_worker_run and core_pubsub_run
|
Yao Yue seems not to be a GitHub user. You need a GitHub account to be able to sign the CLA. If you have already a GitHub account, please add the email address used for this commit to your account. You have signed the CLA already but the status is still pending? Let us recheck it. |
This is a pubsub implementation that's compatible with Redis' pub/sub, but it's a standalone server, not mixed with other redis features.
Also, the publisher does not need to wait for the message fanout to complete before receiving a response, which fully decouples publishing from subscribing.
This is a prototype and many error/corner cases are not yet handled. Nor are unit tests added for the corresponding modules.