redis async client 与自有框架集成

hiredis的异步接口已经支持ae libuv libev 和 libevent集成，具体头文件可以参见redis/deps/hiredis/adapters，样例参见redis/deps/hiredis/examples.

完整样例参见： https://github.com/DavadDi/study_example/tree/master/async_redis_client

参照hireids的异步接口和libevent的集成可以很容易和其他网络框架集成，例如asio或者ace等。 以下样例为自己编写reactor框架的集成方式，

支持自动重练和asyncRedisContext对象的创建和释放，重练使用退步算法，最大连接时间间隔为32秒。

使用方式：
　　将redis_client.hpp 放到 hiredis的adapter目录即可。

#ifndef redis_client_h
#define redis_client_h

#include "reactor/define.hpp"
#include "reactor/event_handler.hpp"
#include "hiredis.h"
#include "async.h"

using namespace reactor;

static  void redisReactorAddRead(void *arg);
static  void redisReactorDelRead(void *arg);
static  void redisReactorAddWrite(void *arg);
static  void redisReactorDelWrite(void *arg);
static  void redisReactorCleanup(void *arg);

void connectCallBack(const redisAsyncContext *c, int status);
void disconnectCallBack(const redisAsyncContext *c, int status);

void get_call_fun(redisAsyncContext *c, void *r, void *arg)
{
    redisReply *reply = (redisReply *)r;
    std::string *key_str = (std::string *)arg;

    if (reply == NULL)
    {
        delete key_str;
        return;
    }

    LOG_INFO("[%s] -> %s
", key_str->c_str(), reply->str);

    delete key_str;

    /* Disconnect after receiving the reply to GET */
    // redisAsyncDisconnect(c);
}

// -------------------------------------------------------------------
// !!NOTE, if obj conneted to server faild and unregister from epoll, 
// prog exit, this object my leak memory
// -------------------------------------------------------------------

class CRedisClient : public CEventHandler
{
    public:
        // enum {MAX_BUF_SIZE = 4096};
        typedef CEventHandler super;

        CRedisClient(const char *srv_ip, uint16_t srv_port, reactor::CReactor *reactor) 
            : super(reactor)
        {
            m_srv_ip_str = srv_ip;
            m_srv_port = srv_port;
        }

        int connect()
        {
            LOG_DEBUG("Enter CRedisClient connect()");
            m_client_status = CONNECT_STATUS::CLIENT_CONNECTING;

            clear_redis_context();

            m_redis_context = redisAsyncConnect(m_srv_ip_str.c_str(), m_srv_port);
            if (m_redis_context == nullptr)
            {
                return -1;
            }

            if (m_redis_context->err)
            {
                LOG_INFO("Connect to %s:%d Error: %s", 
                        m_srv_ip_str.c_str(), m_srv_port, m_redis_context->errstr);

                return -1;
            }

            if (m_timer_id == 0)
            {
                m_timer_id = this->reactor()->regist_timer(this, m_timeout_value); // only one time
                LOG_DEBUG("Client regist timer to reactor id %d, timeout %d", m_timer_id, m_timeout_value);
            }

            this->attach();

            return 0;
        }

        virtual ~CRedisClient()
        {
            // maybe should not free redis context in deconstuct!!
            m_delete_redis_context = true;
            clear_redis_context();
        }

        virtual int open(void *data = nullptr)
        {
            m_client_status = CONNECT_STATUS::CLIENT_CONNECTED;

            m_delete_redis_context = false;

            if (m_timer_id == 0)
            {
                m_timer_id = this->reactor()->regist_timer(this, m_timeout_value); // only one time
                LOG_DEBUG("Client regist timer to reactor id %d, timeout %d", 
                                                    m_timer_id, m_timeout_value);
            }
            
            LOG_INFO("Connect to RedisServer %s:%d succeed!!", 
                                               m_srv_ip_str.c_str(), m_srv_port);
            
            return 0;
        }

        virtual int handle_input(socket_t socket)
        {
            redisAsyncHandleRead(m_redis_context);
            return 0;
        }

        virtual int handle_output(socket_t socket)
        {
            redisAsyncHandleWrite(m_redis_context);
            return 0;
        }

        virtual int handle_timeout(uint32_t tm, void *data = nullptr)
        {
            // LOG_DEBUG("Enter into timeout function....");
            if (m_client_status == CONNECT_STATUS::CLIENT_CONNECTED)
            {
                /* just for test */
                std::string key = std::to_string(tm);

                LOG_DEBUG("Set key %s", key.c_str());
                redisAsyncCommand(m_redis_context, NULL, NULL, "SET %s %s",key.c_str(), "aaa");
                redisAsyncCommand(m_redis_context, get_call_fun, (char*)new string(key), "GET %s", key.c_str());
            }
            else
            {
                static uint32_t last_tm = 0;
                if ((tm - last_tm) >= m_timeout_interval)
                { 
                    //reconnect
                    LOG_DEBUG("Start reconnect now ...");
                    this->connect();

                    m_timeout_interval = m_timeout_interval * 2;
                    if (m_timeout_interval > 32)
                    {
                        m_timeout_interval = 1;
                    }

                    last_tm = tm;
                }
            }

            return 0;
        }

        virtual int handle_close(socket_t socket = INVALID_SOCKET, uint32_t mask = 0)
        {
            LOG_DEBUG("Enter into handle_close()"); 
            m_client_status = CONNECT_STATUS::CLIENT_UNCONNECTED;
            
            // epoll call delete this handler
            if (mask & RE_MASK_DEL)
            {
                LOG_DEBUG("Call RE_MASK_DEL now");

                if (this->m_timer_id && (this->reactor() != nullptr))
                {
                    this->reactor()->unregist_timer(this->m_timer_id);
                    this->m_timer_id = 0;
                }

                delete this;
                return 0;
            }
            
            this->reactor()->del_event(this,0);
            return 0;
        }

        void clear_redis_context()
        {
            if (m_delete_redis_context && m_redis_context != nullptr)
            {
                LOG_DEBUG("Call redisAsynFree() now");
                redisAsyncFree(m_redis_context);
                m_redis_context = nullptr;
            }
        }

        int attach()
        {
            LOG_DEBUG("Enter attatch function... ");

            redisContext *context = &(m_redis_context->c);
            if (m_redis_context->ev.data != NULL)
            {
                return -1;
            }

            // set callback function
            redisAsyncSetConnectCallback(m_redis_context,connectCallBack);
            redisAsyncSetDisconnectCallback(m_redis_context,disconnectCallBack);

            this->set_handle(context->fd); // set handler

            m_redis_context->ev.addRead = redisReactorAddRead;
            m_redis_context->ev.delRead = redisReactorDelRead;
            m_redis_context->ev.addWrite = redisReactorAddWrite;
            m_redis_context->ev.delWrite = redisReactorDelWrite;
            m_redis_context->ev.cleanup = redisReactorCleanup;
            m_redis_context->ev.data = this;

            LOG_DEBUG("ac->ev.data %p", m_redis_context->ev.data);

            this->add_read();
            this->add_write();

            return 0;
        }

        void add_read()
        {
            LOG_TRACE_METHOD(__func__);

            if( (this->m_current_event_mask & reactor::EVENT_READ) > 0)
            {
                LOG_DEBUG("EV_READ(0x%0x) already in event_mask 0x%x",
                        reactor::EVENT_READ, this->m_current_event_mask);

                return;
            }

            this->reactor()->add_event(this, reactor::EVENT_READ);
        }

        void del_read()
        {
            LOG_TRACE_METHOD(__func__);
            this->reactor()->mod_event(this, this->m_current_event_mask&(~reactor::EVENT_READ));
        }

        void add_write()
        {
            LOG_TRACE_METHOD(__func__);
            this->schedule_write();
        }

        void del_write()
        {
            LOG_TRACE_METHOD(__func__);
            this->cancel_schedule_write();
        }

        void clean_up()
        {
            LOG_TRACE_METHOD(__func__);
        }

        // note!!!
        // connenct not succeed. we can free redis context. ]
        // But if connect succeed and borken, we don't connect 

    protected:
        std::string m_srv_ip_str;
        uint16_t    m_srv_port;

        CONNECT_STATUS  m_client_status = CONNECT_STATUS::CLIENT_UNCONNECTED;

        int m_timer_id  = 0;
        uint32_t m_timeout_value = 1;
        uint32_t  m_timeout_interval = 1;

        bool    m_delete_redis_context = true;
        redisAsyncContext *m_redis_context = nullptr;
};

static void redisReactorAddRead(void *arg) 
{
    LOG_DEBUG("Enter redisReactorAddRead() arg %p", arg);
    CRedisClient *event_handler = (CRedisClient *)arg;
    event_handler->add_read();
}

static void redisReactorDelRead(void *arg) 
{
    CRedisClient *event_handler = (CRedisClient *)arg;
    event_handler->del_read();
}

static void redisReactorAddWrite(void *arg) 
{
    CRedisClient *event_handler = (CRedisClient *)arg;
    event_handler->add_write();
}

static void redisReactorDelWrite(void *arg) 
{
    CRedisClient *event_handler = (CRedisClient *)arg;
    event_handler->del_write();
}

static void redisReactorCleanup(void *arg) 
{
    CRedisClient *event_handler = (CRedisClient *)arg;
    event_handler->clean_up();
}

void connectCallBack(const redisAsyncContext *ac, int status)
{
    if (status != REDIS_OK)
    {
        LOG_ERROR("connectCallBack() Error: %s", ac->errstr);
        return;
    }

    CRedisClient *event_handler = (CRedisClient *)ac->ev.data;
    event_handler->open();

    LOG_INFO("RedisClient Connected...");    
}

void disconnectCallBack(const redisAsyncContext *ac, int status)
{ 
    CRedisClient *event_handler = (CRedisClient *)ac->ev.data;
    event_handler->handle_close(0,0);

    if (status != REDIS_OK)
    {
        LOG_INFO("disconnectCallBack()!! Error: %s", ac->errstr);
        return;
    }

    LOG_INFO("RedisClient Disconnected...");
}

#endif /* redis_client_h */

使用的程序样例：

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <signal.h>

//#include <hiredis.h>
//#include <async.h>

#include <adapters/redis_client.hpp>
//#include "redis_client.hpp"

#include <signal.h>

static void signal_handler(int sig)
{
    if (sig == SIGINT)
    {
        reactor::CReactor::instance()->end_event_loop();
    }
}

/*
void get_call_fun(redisAsyncContext *c, void *r, void *arg) 
{
    redisReply *reply = (redisReply *)r;
    std::string *key_str = (std::string *)arg;

    if (reply == NULL) 
    {
        delete key_str;
        return;
    }

    LOG_INFO("[%s] -> %s
", key_str->c_str(), reply->str);

    delete key_str;

    // Disconnect after receiving the reply to GET 
    // redisAsyncDisconnect(c);
}
*/


int main (int argc, char **argv) 
{
    signal(SIGPIPE, SIG_IGN);
    signal(SIGINT, signal_handler);

    CLoggerMgr logger("reactor.prop");

    reactor::CReactor *rt = reactor::CReactor::instance();
    CRedisClient *redis_client = new CRedisClient("127.0.0.1", 6379, rt);
    redis_client->connect();

    rt->run_event_loop();

    return 0;
}