Add code example for pub/sub about hiredis HOT 17 CLOSED

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <signal.h>
#include "hiredis/hiredis.h"
#include "hiredis/async.h"
#include "hiredis/adapters/libevent.h"

void onMessage(redisAsyncContext *c, void *reply, void *privdata) {
    redisReply *r = reply;
    if (reply == NULL) return;

    if (r->type == REDIS_REPLY_ARRAY) {
        for (int j = 0; j < r->elements; j++) {
            printf("%u) %s\n", j, r->element[j]->str);
        }
    }
}

int main (int argc, char **argv) {
    signal(SIGPIPE, SIG_IGN);
    struct event_base *base = event_base_new();

    redisAsyncContext *c = redisAsyncConnect("127.0.0.1", 6379);
    if (c->err) {
        printf("error: %s\n", c->errstr);
        return 1;
    }

    redisLibeventAttach(c, base);
    redisAsyncCommand(c, onMessage, NULL, "SUBSCRIBE testtopic");
    event_base_dispatch(base);
    return 0;
}

from hiredis.

The problem in the documentation is that it isn't written anywhere that you can SUBSCRIBE for something, but then cannot do anything else than subscribing to other events.

Try something like

redisAsyncCommand(c, onMessage, NULL, "SUBSCRIBE testtopic");
redisAsyncCommand(c, onAnotherMessage, NULL, "SUBSCRIBE anothertopic");
redisAsyncCommand(c, onReply, NULL, "SET toto 5");
redisAsyncCommand(c, onReply, NULL, "PUBLISH testtopic \"hello\"");
redisAsyncCommand(c, onReply, NULL, "GET toto");

And only the 2 first commands will work. The other respond with REDIS_OK but the onReply() callback gets a NULL reply.

The only way to do things correctly is to use 2 redis contexts, one for the subscribes (and MONITOR), the other for the rest. Maybe the doc should be updated to reflect this limitation?

from hiredis.

On Ubuntu 14.04, libevent-dev libraries need to be installed to successfully compile this example. Also, -levent flag needs to be passed during compilation.

from hiredis.

How do you capture an event for when the socket eventually closes?

from hiredis.

I think it's a good suggestion. I'm quite confused about pub/sub in writing practical code

from hiredis.

+1 for this, I'm having difficulties with async libev pub/sub

from hiredis.

Flagged for ng.

from hiredis.

Is there a doc for EACH api?

from hiredis.

How do you capture an event for when the socket eventually closes?

Or how do you detect rapidly a timeout and redo a new connection? For example if a firewall in between closes the connection and drop packets after.

By having only the subscribe on that connection it makes it impossible as I understand it to send something to check for timeouts. Also TCP keepalives do not seem to exists with ASYNC and anyway these are hard to configure properly on linux inside a docker container for example (the intervals, retry, etc.)

Normally with TCP sockets you could have a short read/write timeout on the socket and after you send something you can see if it timeouts and quickly try to reconnect to another server or do something else...

If there was something else that could send a ping command asynchronously and configure the read/write timeouts to trigger disconnections that would be nice.

from hiredis.

@Gerporgl
In my own implementation I utilized redisAsyncSetDiconnectCallback I have not researched the mechanism of how this works.
nidhhoggr/twemproxy_sentinel@602e07c

from hiredis.

In my case the disconnect event does not seem to be fired fast enough (or not at all in some cases) for this case of what I always called a "silent disconnection", in general it takes way to long (I waited more than 5 minutes and never received the event, but then after like 10-15 minutes I finally got one)

After reading a bit more about the subject, I noticed that you can still do PINGs command while subscribed, and found this post kind of mentioning you can subscribe to your own PING response event as if it was a published message:
#351

Essentially what seems to work well now is to send PINGs for example at 1 second interval, and if you don't receive anymore ping response (or you allow a certain tolerance threshold) then you assume the connection is dead, do your cleanup, then reconnect. That seems a lot more robust and quick.

Using libevent and hiredis you can do a complete async subscribe implementation using timer events for PING while handling the normal message delivery asynchronously.

from hiredis.

@Gerporgl
This was very helpful!
By any chance did you try to use the same redisAsyncCommand/ Async context mechanism for key-space event notifications?
I am trying to make that part work; but unfortunately not getting the messages.

from hiredis.

Have you tried this?

redis-cli config set notify-keyspace-events KEA

redisAsyncCommand(c, subscribeCallback, NULL, "PSUBSCRIBE __key*__:*");

from hiredis.

@joe-at-startupmedia
Thanks! I already tried this.. and it seems all i am getting back is first psubscribe response.
Any updates in the data is not actually received in the callback.
Here is sample code:

`
#define SUBSCRIBE_CHANNEL "SUBSCRIBE URLC_Updates"
#define SUBSCRIBE_KEYEVENT "PSUBSCRIBE 'key*:*'"
#define SCAN_DB "SCAN %d COUNT 100"
#define QUERY_KEY "GET %s"

void
onPubsubMessage(redisAsyncContext *c, void *reply, void *privdata)
{
redisReply *r = (redisReply *) reply;
if (reply == NULL) return;

/* What if reply type is something else... */
if (r->type == REDIS_REPLY_ARRAY) {

    for (int j = 0; j < r->elements; j++) {
        if (r->element[j]->type == REDIS_REPLY_STRING) { 
            myPubsubFile.open(PUBSUB_FILE, fstream::in | fstream::out | fstream::app);
            myPubsubFile << r->element[j]->str << endl;
            myPubsubFile.close();
        } else if (r->element[j]->type == REDIS_REPLY_INTEGER) {
            cout << "Integer : "<< r->element[j]->integer << endl;
        }
    }
}
return;

}
void
onKeyspaceMessage(redisAsyncContext *c, void *reply, void *privdata)
{
redisReply *r = (redisReply *) reply;
if (reply == NULL) return;

cout << "Got keyspace event notification from REDIS.. " << endl;

/* What if reply type is something else... */
if (r->type == REDIS_REPLY_ARRAY) {
    cout << "Type is an array.. " << endl;
    cout << "Number of elements here: " << r->elements << endl;

    for (int j = 0; j < r->elements; j++) {
        cout << "\t\t Type for element : " << r->element[j]->type << endl;
        if (r->element[j]->type == REDIS_REPLY_STRING) {
            cout << "\t\t\t" << r->element[j]->str << endl;
        } else if (r->element[j]->type == REDIS_REPLY_INTEGER) {
            cout << "\t\t\t" << "Integer : "<< r->element[j]->integer << endl;
        }   
    }   
} else {
    cout << "This is the type for response : " << r->type << endl;
}   
return;

}
void*
pubsubRecipient(void* arg)
{
struct event_base *base = event_base_new();

redisAsyncContext *c = redisAsyncConnect("127.0.0.1", 6379);
if (c->err) {
    printf("error: %s\n", c->errstr);
    return NULL;
}   

redisLibeventAttach(c, base);
redisAsyncCommand(c, onPubsubMessage    , NULL, SUBSCRIBE_CHANNEL);
redisAsyncCommand(c, onKeyspaceMessage  , NULL, SUBSCRIBE_KEYEVENT);

`

P.S. The config part, which enables key-space notifications in REDIS is in a separate process; and i can see the effect of that config, when i run an instance of redis-cli in separate terminal.

from hiredis.

Pardon If I am missing something here but your keyevent subscription string is:

 "PSUBSCRIBE 'key*:*'"

Should it not be changed to the following to match that pattern?

"PSUBSCRIBE __key*__:*"

from hiredis.

Ohk.. something weird is going on with this editor... the underscores surrounding key were present when i copied and pasted my code.

nevertheless, looks like the issue was -> ' ' <- (single quotes) surrounding my _ _ key * _ _ pattern.
redis-cli does accept them without any issues; but when we send it via redis command, there is something wrong here.

from hiredis.

Going through old issues. An example was added ages ago to the Wiki (thanks @aluiken)

from hiredis.