Commit d5fc7d8c authored by Pieter Noordhuis's avatar Pieter Noordhuis
Browse files

Update libev and libevent examples to work with async.h

parent ac13c9f0
#include <sys/types.h> #include <sys/types.h>
#include <ev.h> #include <ev.h>
#include <hiredis.h> #include <hiredis.h>
#include <async.h>
/* Prototype for the error callback. */ typedef struct redisLibevEvents {
typedef void (redisErrorCallback)(const redisContext*); redisAsyncContext *context;
typedef struct libevRedisEvents {
redisContext *context;
redisErrorCallback *err;
struct ev_loop *loop; struct ev_loop *loop;
int reading, writing;
ev_io rev, wev; ev_io rev, wev;
} libevRedisEvents; } redisLibevEvents;
void libevRedisReadEvent(struct ev_loop *loop, ev_io *watcher, int revents) { void redisLibevReadEvent(struct ev_loop *loop, ev_io *watcher, int revents) {
((void)loop); ((void)revents); ((void)loop); ((void)revents);
libevRedisEvents *e = watcher->data; redisLibevEvents *e = watcher->data;
redisAsyncHandleRead(e->context);
if (redisBufferRead(e->context) == REDIS_ERR) {
redisDisconnect(e->context);
e->err(e->context);
} else {
if (redisProcessCallbacks(e->context) == REDIS_ERR) {
redisDisconnect(e->context);
e->err(e->context);
}
}
} }
void libevRedisWriteEvent(struct ev_loop *loop, ev_io *watcher, int revents) { void redisLibevWriteEvent(struct ev_loop *loop, ev_io *watcher, int revents) {
((void)loop); ((void)revents); ((void)loop); ((void)revents);
libevRedisEvents *e = watcher->data; redisLibevEvents *e = watcher->data;
int done = 0; redisAsyncHandleWrite(e->context);
}
if (redisBufferWrite(e->context, &done) == REDIS_ERR) { void redisLibevAddRead(void *privdata) {
redisDisconnect(e->context); redisLibevEvents *e = privdata;
e->err(e->context); if (!e->reading) {
} else { e->reading = 1;
/* Stop firing the write event when done */ ev_io_start(e->loop,&e->rev);
if (done) {
ev_io_stop(e->loop,&e->wev);
ev_io_start(e->loop,&e->rev);
}
} }
} }
void libevRedisCommandCallback(redisContext *c, void *privdata) { void redisLibevDelRead(void *privdata) {
((void)c); redisLibevEvents *e = privdata;
libevRedisEvents *e = privdata; if (e->reading) {
ev_io_start(e->loop,&e->wev); e->reading = 0;
ev_io_stop(e->loop,&e->rev);
}
} }
void libevRedisDisconnectCallback(redisContext *c, void *privdata) { void redisLibevAddWrite(void *privdata) {
((void)c); redisLibevEvents *e = privdata;
libevRedisEvents *e = privdata; if (!e->writing) {
ev_io_stop(e->loop,&e->rev); e->writing = 1;
ev_io_stop(e->loop,&e->wev); ev_io_start(e->loop,&e->wev);
}
} }
void libevRedisFreeCallback(redisContext *c, void *privdata) { void redisLibevDelWrite(void *privdata) {
((void)c); redisLibevEvents *e = privdata;
libevRedisEvents *e = privdata; if (e->writing) {
e->writing = 0;
ev_io_stop(e->loop,&e->wev);
}
}
void redisLibevCleanup(void *privdata) {
redisLibevEvents *e = privdata;
redisLibevDelRead(privdata);
redisLibevDelWrite(privdata);
free(e); free(e);
} }
redisContext *libevRedisConnect(struct ev_loop *loop, redisErrorCallback *err, const char *ip, int port) { int redisLibevAttach(redisAsyncContext *ac, struct ev_loop *loop) {
libevRedisEvents *e; redisContext *c = &(ac->c);
redisContext *c = redisConnectNonBlock(ip, port, NULL); redisLibevEvents *e;
if (c->error != NULL) {
err(c); /* Nothing should be attached when something is already attached */
redisFree(c); if (ac->data != NULL)
return NULL; return REDIS_ERR;
}
/* Create container for context and r/w events */ /* Create container for context and r/w events */
e = malloc(sizeof(*e)); e = malloc(sizeof(*e));
e->context = c; e->context = ac;
e->err = err;
e->loop = loop; e->loop = loop;
e->reading = e->writing = 0;
e->rev.data = e; e->rev.data = e;
e->wev.data = e; e->wev.data = e;
/* Register callbacks */ /* Register functions to start/stop listening for events */
redisSetDisconnectCallback(c,libevRedisDisconnectCallback,e); ac->evAddRead = redisLibevAddRead;
redisSetCommandCallback(c,libevRedisCommandCallback,e); ac->evDelRead = redisLibevDelRead;
redisSetFreeCallback(c,libevRedisFreeCallback,e); ac->evAddWrite = redisLibevAddWrite;
ac->evDelWrite = redisLibevDelWrite;
ac->evCleanup = redisLibevCleanup;
ac->data = e;
/* Initialize read/write events */ /* Initialize read/write events */
ev_io_init(&e->rev,libevRedisReadEvent,c->fd,EV_READ); ev_io_init(&e->rev,redisLibevReadEvent,c->fd,EV_READ);
ev_io_init(&e->wev,libevRedisWriteEvent,c->fd,EV_WRITE); ev_io_init(&e->wev,redisLibevWriteEvent,c->fd,EV_WRITE);
return c; return REDIS_OK;
} }
#include <sys/types.h> #include <sys/types.h>
#include <event.h> #include <event.h>
#include <hiredis.h> #include <hiredis.h>
#include <async.h>
/* Prototype for the error callback. */ typedef struct redisLibeventEvents {
typedef void (redisErrorCallback)(const redisContext*); redisAsyncContext *context;
typedef struct libeventRedisEvents {
redisContext *context;
redisErrorCallback *err;
struct event rev, wev; struct event rev, wev;
} libeventRedisEvents; } redisLibeventEvents;
void libeventRedisReadEvent(int fd, short event, void *arg) { void redisLibeventReadEvent(int fd, short event, void *arg) {
((void)fd); ((void)event); ((void)fd); ((void)event);
libeventRedisEvents *e = arg; redisLibeventEvents *e = arg;
redisAsyncHandleRead(e->context);
/* Always re-schedule read events */
event_add(&e->rev,NULL);
if (redisBufferRead(e->context) == REDIS_ERR) {
redisDisconnect(e->context);
e->err(e->context);
} else {
if (redisProcessCallbacks(e->context) == REDIS_ERR) {
redisDisconnect(e->context);
e->err(e->context);
}
}
} }
void libeventRedisWriteEvent(int fd, short event, void *arg) { void redisLibeventWriteEvent(int fd, short event, void *arg) {
((void)fd); ((void)event); ((void)fd); ((void)event);
libeventRedisEvents *e = arg; redisLibeventEvents *e = arg;
int done = 0; redisAsyncHandleWrite(e->context);
}
if (redisBufferWrite(e->context,&done) == REDIS_ERR) { void redisLibeventAddRead(void *privdata) {
redisDisconnect(e->context); redisLibeventEvents *e = privdata;
e->err(e->context); event_add(&e->rev,NULL);
} else {
/* Schedule write event again when writing is not done. */
if (!done) {
event_add(&e->wev,NULL);
} else {
event_add(&e->rev,NULL);
}
}
} }
/* Schedule to be notified on a write event, so the outgoing buffer void redisLibeventDelRead(void *privdata) {
* can be flushed to the socket. */ redisLibeventEvents *e = privdata;
void libeventRedisCommandCallback(redisContext *c, void *privdata) { event_del(&e->rev);
((void)c); }
libeventRedisEvents *e = privdata;
void redisLibeventAddWrite(void *privdata) {
redisLibeventEvents *e = privdata;
event_add(&e->wev,NULL); event_add(&e->wev,NULL);
} }
/* Remove event handlers when the context gets disconnected. */ void redisLibeventDelWrite(void *privdata) {
void libeventRedisDisconnectCallback(redisContext *c, void *privdata) { redisLibeventEvents *e = privdata;
((void)c);
libeventRedisEvents *e = privdata;
event_del(&e->rev);
event_del(&e->wev); event_del(&e->wev);
} }
/* Free the libeventRedisEvents struct when the context is free'd. */ void redisLibeventCleanup(void *privdata) {
void libeventRedisFreeCallback(redisContext *c, void *privdata) { redisLibeventEvents *e = privdata;
((void)c); event_del(&e->rev);
libeventRedisEvents *e = privdata; event_del(&e->wev);
free(e); free(e);
} }
redisContext *libeventRedisConnect(struct event_base *base, redisErrorCallback *err, const char *ip, int port) { int redisLibeventAttach(redisAsyncContext *ac, struct event_base *base) {
libeventRedisEvents *e; redisContext *c = &(ac->c);
redisContext *c = redisConnectNonBlock(ip,port,NULL); redisLibeventEvents *e;
if (c->error != NULL) {
err(c); /* Nothing should be attached when something is already attached */
redisFree(c); if (ac->data != NULL)
return NULL; return REDIS_ERR;
}
/* Create container for context and r/w events */ /* Create container for context and r/w events */
e = malloc(sizeof(*e)); e = malloc(sizeof(*e));
e->context = c; e->context = ac;
e->err = err;
/* Register callbacks */ /* Register functions to start/stop listening for events */
redisSetDisconnectCallback(c,libeventRedisDisconnectCallback,e); ac->evAddRead = redisLibeventAddRead;
redisSetCommandCallback(c,libeventRedisCommandCallback,e); ac->evDelRead = redisLibeventDelRead;
redisSetFreeCallback(c,libeventRedisFreeCallback,e); ac->evAddWrite = redisLibeventAddWrite;
ac->evDelWrite = redisLibeventDelWrite;
ac->evCleanup = redisLibeventCleanup;
ac->data = e;
/* Initialize and install read/write events */ /* Initialize and install read/write events */
event_set(&e->rev,c->fd,EV_READ,libeventRedisReadEvent,e); event_set(&e->rev,c->fd,EV_READ,redisLibeventReadEvent,e);
event_set(&e->wev,c->fd,EV_WRITE,libeventRedisWriteEvent,e); event_set(&e->wev,c->fd,EV_WRITE,redisLibeventWriteEvent,e);
event_base_set(base,&e->rev); event_base_set(base,&e->rev);
event_base_set(base,&e->wev); event_base_set(base,&e->wev);
return c; return REDIS_OK;
} }
...@@ -2,30 +2,37 @@ ...@@ -2,30 +2,37 @@
#include <stdlib.h> #include <stdlib.h>
#include <string.h> #include <string.h>
#include <hiredis/libev.h> #include <hiredis/libev.h>
#include <async.h>
#include <signal.h> #include <signal.h>
void getCallback(redisContext *c, redisReply *reply, void *privdata) { void getCallback(redisAsyncContext *c, redisReply *reply, void *privdata) {
if (reply == NULL) return; /* Error */ printf("argv[%s]: %s\n", (char*)privdata, reply->str);
printf("argv[%s]: %s\n", (char*)privdata, reply->reply);
/* Disconnect after receiving the reply to GET */ /* Disconnect after receiving the reply to GET */
redisDisconnect(c); redisAsyncDisconnect(c);
} }
void errorCallback(const redisContext *c) { void disconnectCallback(const redisAsyncContext *c, int status) {
printf("Error: %s\n", c->error); if (status != REDIS_OK) {
printf("Error: %s\n", c->error);
}
} }
int main (int argc, char **argv) { int main (int argc, char **argv) {
signal(SIGPIPE, SIG_IGN); signal(SIGPIPE, SIG_IGN);
struct ev_loop *loop = ev_default_loop(0); struct ev_loop *loop = ev_default_loop(0);
redisContext *c = libevRedisConnect(loop, errorCallback, "127.0.0.1", 6379); redisAsyncContext *c = redisAsyncConnect("127.0.0.1", 6379);
if (c == NULL) return 1; if (c->error != NULL) {
/* Let *c leak for now... */
printf("Error: %s\n", c->error);
return 1;
}
redisCommand(c, "SET key %b", argv[argc-1], strlen(argv[argc-1])); redisLibevAttach(c,loop);
redisCommandWithCallback(c, getCallback, (char*)"end-1", "GET key"); redisAsyncSetDisconnectCallback(c,disconnectCallback);
redisAsyncCommand(c, NULL, NULL, "SET key %b", argv[argc-1], strlen(argv[argc-1]));
redisAsyncCommand(c, getCallback, (char*)"end-1", "GET key");
ev_loop(loop, 0); ev_loop(loop, 0);
redisFree(c);
return 0; return 0;
} }
...@@ -2,30 +2,37 @@ ...@@ -2,30 +2,37 @@
#include <stdlib.h> #include <stdlib.h>
#include <string.h> #include <string.h>
#include <hiredis/libevent.h> #include <hiredis/libevent.h>
#include <async.h>
#include <signal.h> #include <signal.h>
void getCallback(redisContext *c, redisReply *reply, void *privdata) { void getCallback(redisAsyncContext *c, redisReply *reply, void *privdata) {
if (reply == NULL) return; /* Error */ printf("argv[%s]: %s\n", (char*)privdata, reply->str);
printf("argv[%s]: %s\n", (const char*)privdata, reply->reply);
/* Disconnect after receiving the reply to GET */ /* Disconnect after receiving the reply to GET */
redisDisconnect(c); redisAsyncDisconnect(c);
} }
void errorCallback(const redisContext *c) { void disconnectCallback(const redisAsyncContext *c, int status) {
printf("Error: %s\n", c->error); if (status != REDIS_OK) {
printf("Error: %s\n", c->error);
}
} }
int main (int argc, char **argv) { int main (int argc, char **argv) {
signal(SIGPIPE, SIG_IGN); signal(SIGPIPE, SIG_IGN);
struct event_base *base = event_base_new(); struct event_base *base = event_base_new();
redisContext *c = libeventRedisConnect(base, errorCallback, "127.0.0.1", 6379); redisAsyncContext *c = redisAsyncConnect("127.0.0.1", 6379);
if (c == NULL) return 1; if (c->error != NULL) {
/* Let *c leak for now... */
printf("Error: %s\n", c->error);
return 1;
}
redisCommand(c, "SET key %b", argv[argc-1], strlen(argv[argc-1])); redisLibeventAttach(c,base);
redisCommandWithCallback(c, getCallback, (char*)"end-1", "GET key"); redisAsyncSetDisconnectCallback(c,disconnectCallback);
redisAsyncCommand(c, NULL, NULL, "SET key %b", argv[argc-1], strlen(argv[argc-1]));
redisAsyncCommand(c, getCallback, (char*)"end-1", "GET key");
event_base_dispatch(base); event_base_dispatch(base);
redisFree(c);
return 0; return 0;
} }
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment