#include #include #include "adapters/libevent.h" #include #include #include "common.h" using namespace hiredis; struct AsyncClient; typedef std::function ConnectionCallback; typedef std::function DisconnectCallback; typedef std::function CommandCallback; static void realConnectCb(const redisAsyncContext*, int); static void realDisconnectCb(const redisAsyncContext*, int); static void realCommandCb(redisAsyncContext*, void*, void*); struct CmdData { AsyncClient *client; CommandCallback cb; }; struct AsyncClient { AsyncClient(const redisOptions& options, event_base* b) { ac = redisAsyncConnectWithOptions(&options); redisLibeventAttach(ac, b); redisAsyncSetConnectCallback(ac, realConnectCb); ac->data = this; } AsyncClient(const ClientSettings& settings, event_base *b) { redisOptions options = { 0 }; ac = redisAsyncConnectWithOptions(&options); redisLibeventAttach(ac, b); redisAsyncSetConnectCallback(ac, realConnectCb); if (settings.is_ssl()) { redisSecureConnection(&ac->c, settings.ssl_ca(), settings.ssl_cert(), settings.ssl_key(), NULL); } ac->data = this; } AsyncClient(redisAsyncContext *ac) : ac(ac) { redisAsyncSetDisconnectCallback(ac, realDisconnectCb); } void onConnect(ConnectionCallback cb) { conncb = cb; } ~AsyncClient() { if (ac != NULL) { auto tmpac = ac; ac = NULL; redisAsyncDisconnect(tmpac); } } void cmd(CommandCallback cb, const char *fmt, ...) { auto data = new CmdData {this, cb }; va_list ap; va_start(ap, fmt); redisvAsyncCommand(ac, realCommandCb, data, fmt, ap); va_end(ap); } void disconnect(DisconnectCallback cb) { disconncb = cb; redisAsyncDisconnect(ac); } void disconnect() { redisAsyncDisconnect(ac); } ConnectionCallback conncb; DisconnectCallback disconncb; redisAsyncContext *ac; }; static void realConnectCb(const redisAsyncContext *ac, int status) { auto self = reinterpret_cast(ac->data); if (self->conncb) { self->conncb(self, status == 0); } } static void realDisconnectCb(const redisAsyncContext *ac, int status) { auto self = reinterpret_cast(ac->data); if (self->disconncb) { self->disconncb(self, status == 0); } } static void realCommandCb(redisAsyncContext *ac, void *r, void *ctx) { auto *d = reinterpret_cast(ctx); auto *rep = reinterpret_cast(r); auto *self = reinterpret_cast(ac->data); d->cb(self, rep); delete d; } class AsyncTest : public ::testing::Test { protected: void SetUp() override { libevent = event_base_new(); } void TearDown() override { event_base_free(libevent); libevent = NULL; } void wait() { event_base_dispatch(libevent); } event_base *libevent; }; TEST_F(AsyncTest, testAsync) { redisOptions options = {0}; struct timeval tv = {0}; tv.tv_sec = 1; options.timeout = &tv; settings_g.initOptions(options); AsyncClient client(options, libevent); client.onConnect([](AsyncClient*, bool status) { printf("Status: %d\n", status); }); client.cmd([](AsyncClient *c, redisReply*){ printf("Got reply!\n"); c->disconnect(); }, "PING"); wait(); }