use evpp to refactor AsyncHttpClient

CMakeLists.txt

@@ -134,7 +134,7 @@ endif()
 
 # see Makefile
 set(ALL_SRCDIRS . base utils event protocol http http/client http/server consul examples)
-set(LIBHV_SRCDIRS . base utils event)
+set(LIBHV_SRCDIRS . base utils event evpp)
 set(LIBHV_HEADERS hv.h hconfig.h hexport.h)
 set(LIBHV_HEADERS ${LIBHV_HEADERS} ${BASE_HEADERS} ${UTILS_HEADERS} ${EVENT_HEADERS})
 

Makefile

@@ -4,7 +4,7 @@ include Makefile.vars
 MAKEF=$(MAKE) -f Makefile.in
 ALL_SRCDIRS=. base utils event protocol http http/client http/server consul examples
 
-LIBHV_SRCDIRS = . base utils event
+LIBHV_SRCDIRS = . base utils event evpp
 LIBHV_HEADERS = hv.h hconfig.h hexport.h
 LIBHV_HEADERS += $(BASE_HEADERS) $(UTILS_HEADERS) $(EVENT_HEADERS)
 

evpp/Event.h

@@ -11,7 +11,8 @@ namespace hv {
 struct Event;
 struct Timer;
 
-typedef uint64_t    TimerID;
+typedef uint64_t            TimerID;
+#define INVALID_TIMER_ID    ((TimerID)-1)
 
 typedef std::function<void(Event*)>     EventCallback;
 typedef std::function<void(TimerID)>    TimerCallback;

examples/http_client_test.cpp

@@ -2,29 +2,24 @@
 
 #include "hthread.h" // import hv_gettid
 
-static void onResponse(int state, HttpRequestPtr req, HttpResponsePtr resp, void* userdata) {
-    printf("test_http_async_client response thread tid=%ld\n", hv_gettid());
-    if (state != 0) {
-        printf("onError: %s:%d\n", http_client_strerror(state), state);
-    } else {
-        printf("onSuccess\n");
-        printf("%d %s\r\n", resp->status_code, resp->status_message());
-        printf("%s\n", resp->body.c_str());
-    }
-
-    int* finished = (int*)userdata;
-    *finished = 1;
-}
-
 static void test_http_async_client(int* finished) {
     printf("test_http_async_client request thread tid=%ld\n", hv_gettid());
     HttpRequestPtr req = HttpRequestPtr(new HttpRequest);
-    HttpResponsePtr resp = HttpResponsePtr(new HttpResponse);
     req->method = HTTP_POST;
     req->url = "127.0.0.1:8080/echo";
+    req->headers["Connection"] = "keep-alive";
     req->body = "this is an async request.";
     req->timeout = 10;
-    int ret = http_client_send_async(req, resp, onResponse, (void*)finished);
+    int ret = http_client_send_async(req, [finished](const HttpResponsePtr& resp) {
+        printf("test_http_async_client response thread tid=%ld\n", hv_gettid());
+        if (resp == NULL) {
+            printf("request failed!\n");
+        } else {
+            printf("%d %s\r\n", resp->status_code, resp->status_message());
+            printf("%s\n", resp->body.c_str());
+        }
+        *finished = 1;
+    });
     if (ret != 0) {
         printf("http_client_send_async error: %s:%d\n", http_client_strerror(ret), ret);
         *finished = 1;
@@ -60,9 +55,15 @@ static void test_http_sync_client() {
 
 int main() {
     int finished = 0;
-    test_http_async_client(&finished);
 
-    test_http_sync_client();
+    int cnt = 1;
+    for (int i = 0; i < cnt; ++i) {
+        test_http_async_client(&finished);
+
+        test_http_sync_client();
+
+        sleep(1);
+    }
 
     // demo wait async ResponseCallback
     while (!finished) {

http/HttpMessage.h

@@ -29,6 +29,7 @@
 #include <memory>
 #include <string>
 #include <map>
+#include <functional>
 
 #include "hexport.h"
 #include "hbase.h"
@@ -319,7 +320,6 @@ public:
 
 typedef std::shared_ptr<HttpRequest>    HttpRequestPtr;
 typedef std::shared_ptr<HttpResponse>   HttpResponsePtr;
-// state: 0 onSucceed other onError
-typedef void (*HttpResponseCallback)(int state, HttpRequestPtr req, HttpResponsePtr resp, void* userdata);
+typedef std::function<void(const HttpResponsePtr&)> HttpResponseCallback;
 
 #endif // HTTP_MESSAGE_H_

http/client/AsyncHttpClient.cpp

@@ -0,0 +1,151 @@
+#include "AsyncHttpClient.h"
+
+namespace hv {
+
+int AsyncHttpClient::sendInLoopImpl(const HttpRequestPtr& req, HttpResponseCallback resp_cb, uint64_t start_hrtime) {
+    // queueInLoop timeout?
+    uint64_t now_hrtime = hloop_now_hrtime(loop_thread.hloop());
+    int elapsed_ms = (now_hrtime - start_hrtime) / 1000;
+    int timeout_ms = req->timeout * 1000;
+    if (timeout_ms > 0 && elapsed_ms >= timeout_ms) {
+        hlogw("%s queueInLoop timeout!", req->url.c_str());
+        return -10;
+    }
+
+    req->ParseUrl();
+    sockaddr_u peeraddr;
+    memset(&peeraddr, 0, sizeof(peeraddr));
+    int ret = sockaddr_set_ipport(&peeraddr, req->host.c_str(), req->port);
+    if (ret != 0) {
+        hloge("unknown host %s", req->host.c_str());
+        return -20;
+    }
+
+    int connfd = -1;
+    hio_t* connio = NULL;
+    HttpClientContextPtr ctx = NULL;
+
+    // first get from conn_pools
+    char strAddr[SOCKADDR_STRLEN] = {0};
+    SOCKADDR_STR(&peeraddr, strAddr);
+    auto iter = conn_pools.find(strAddr);
+    if (iter != conn_pools.end()) {
+        if (iter->second.get(connfd)) {
+            // hlogd("get from conn_pools");
+            ctx = getContext(connfd);
+            ctx->req = req;
+            ctx->cb = resp_cb;
+        }
+    }
+
+    if (connfd < 0) {
+        // create socket
+        connfd = socket(peeraddr.sa.sa_family, SOCK_STREAM, 0);
+        if (connfd < 0) {
+            perror("socket");
+            return -30;
+        }
+        connio = hio_get(loop_thread.hloop(), connfd);
+        assert(connio != NULL);
+        hio_set_peeraddr(connio, &peeraddr.sa, sockaddr_len(&peeraddr));
+        // https
+        if (req->https) {
+            hio_enable_ssl(connio);
+        }
+    }
+
+    if (ctx == NULL) {
+        // new HttpClientContext
+        ctx.reset(new HttpClientContext);
+        ctx->req = req;
+        ctx->cb = resp_cb;
+        ctx->channel.reset(new SocketChannel(connio));
+        ctx->channel->onread = [this, ctx](Buffer* buf) {
+            const char* data = (const char*)buf->data();
+            int len = buf->size();
+            int nparse = ctx->parser->FeedRecvData(data, len);
+            if (nparse != len) {
+                ctx->errorCallback();
+                ctx->channel->close();
+                return;
+            }
+            if (ctx->parser->IsComplete()) {
+                std::string req_connection = ctx->req->GetHeader("Connection");
+                std::string resp_connection = ctx->resp->GetHeader("Connection");
+                ctx->successCallback();
+                if (stricmp(req_connection.c_str(), "keep-alive") == 0 &&
+                    stricmp(resp_connection.c_str(), "keep-alive") == 0) {
+                    // add into conn_pools to reuse
+                    // hlogd("add into conn_pools");
+                    conn_pools[ctx->channel->peeraddr()].add(ctx->channel->fd());
+                } else {
+                    ctx->channel->close();
+                }
+            }
+        };
+        ctx->channel->onclose = [this, ctx]() {
+            ctx->channel->status = SocketChannel::CLOSED;
+            removeContext(ctx);
+            ctx->errorCallback();
+        };
+        addContext(ctx);
+    }
+
+    // timer
+    if (timeout_ms > 0) {
+        ctx->timerID = setTimeout(timeout_ms - elapsed_ms, [ctx](TimerID timerID){
+            hlogw("%s timeout!", ctx->req->url.c_str());
+            if (ctx->channel) {
+                ctx->channel->close();
+            }
+        });
+    }
+
+    if (ctx->channel->isConnected()) {
+        // sendRequest
+        sendRequest(ctx);
+    } else {
+        // startConnect
+        hevent_set_userdata(connio, this);
+        hio_setcb_connect(connio, onconnect);
+        hio_connect(connio);
+    }
+
+    return 0;
+}
+
+void AsyncHttpClient::onconnect(hio_t* io) {
+    AsyncHttpClient* client = (AsyncHttpClient*)hevent_userdata(io);
+    HttpClientContextPtr ctx = client->getContext(hio_fd(io));
+    assert(ctx != NULL && ctx->req != NULL && ctx->channel != NULL);
+
+    ctx->channel->status = SocketChannel::CONNECTED;
+    client->sendRequest(ctx);
+    ctx->channel->startRead();
+}
+
+int AsyncHttpClient::sendRequest(const HttpClientContextPtr ctx) {
+    assert(ctx != NULL && ctx->req != NULL && ctx->channel != NULL);
+    SocketChannelPtr channel = ctx->channel;
+
+    if (ctx->parser == NULL) {
+        ctx->parser.reset(HttpParser::New(HTTP_CLIENT, (http_version)ctx->req->http_major));
+    }
+    if (ctx->resp == NULL) {
+        ctx->resp.reset(new HttpResponse);
+    }
+
+    ctx->parser->InitResponse(ctx->resp.get());
+    ctx->parser->SubmitRequest(ctx->req.get());
+
+    char* data = NULL;
+    size_t len = 0;
+    while (ctx->parser->GetSendData(&data, &len)) {
+        Buffer buf(data, len);
+        channel->write(&buf);
+    }
+
+    return 0;
+}
+
+}

http/client/AsyncHttpClient.h

@@ -0,0 +1,158 @@
+#ifndef HV_ASYNC_HTTP_CLIENT_H_
+#define HV_ASYNC_HTTP_CLIENT_H_
+
+#include <list>
+
+#include "EventLoopThread.h"
+#include "Channel.h"
+
+#include "HttpMessage.h"
+#include "HttpParser.h"
+
+// async => keepalive => connect_pool
+
+namespace hv {
+
+template<typename Conn>
+class ConnPool {
+public:
+    int size() {
+        return conns_.size();
+    }
+
+    bool get(Conn& conn) {
+        if (conns_.empty()) return false;
+        conn = conns_.front();
+        conns_.pop_front();
+        return true;
+    }
+
+    bool add(const Conn& conn) {
+        conns_.push_back(conn);
+        return true;
+    }
+
+    bool remove(const Conn& conn) {
+        auto iter = conns_.begin();
+        while (iter != conns_.end()) {
+            if (*iter == conn) {
+                iter = conns_.erase(iter);
+                return true;
+            } else {
+                ++iter;
+            }
+        }
+        return false;
+    }
+
+private:
+    std::list<Conn>  conns_;
+};
+
+struct HttpClientContext {
+    HttpRequestPtr          req;
+    HttpResponseCallback    cb;
+
+    SocketChannelPtr    channel;
+    HttpResponsePtr     resp;
+    HttpParserPtr       parser;
+
+    TimerID             timerID;
+
+    HttpClientContext() {
+        timerID = INVALID_TIMER_ID;
+    }
+
+    void callback() {
+        if (timerID != INVALID_TIMER_ID) {
+            killTimer(timerID);
+            timerID = INVALID_TIMER_ID;
+        }
+        if (cb) {
+            cb(resp);
+            // NOTE: ensure cb just call once
+            cb = NULL;
+        }
+    }
+
+    void successCallback() {
+        callback();
+        resp = NULL;
+    }
+
+    void errorCallback() {
+        resp = NULL;
+        callback();
+    }
+};
+typedef std::shared_ptr<HttpClientContext>  HttpClientContextPtr;
+
+class AsyncHttpClient {
+public:
+    AsyncHttpClient() {
+        loop_thread.start(true);
+    }
+    ~AsyncHttpClient() {
+        loop_thread.stop(true);
+    }
+
+    // thread-safe
+    int send(const HttpRequestPtr& req, HttpResponseCallback resp_cb) {
+        uint64_t start_hrtime = hloop_now_hrtime(loop_thread.hloop());
+        loop_thread.loop()->queueInLoop(std::bind(&AsyncHttpClient::sendInLoop, this, req, resp_cb, start_hrtime));
+    }
+
+protected:
+    void sendInLoop(const HttpRequestPtr& req, HttpResponseCallback resp_cb, uint64_t start_hrtime) {
+        int err = sendInLoopImpl(req, resp_cb, start_hrtime);
+        if (err != 0 && resp_cb) {
+            resp_cb(NULL);
+        }
+    }
+    // createsocket => startConnect =>
+    // onconnect => sendRequest => startRead =>
+    // onread => HttpParser => resp_cb
+    int sendInLoopImpl(const HttpRequestPtr& req, HttpResponseCallback resp_cb, uint64_t start_hrtime);
+
+    // InitResponse => SubmitRequest => while(GetSendData) write => startRead
+    static void onconnect(hio_t* io);
+    static int sendRequest(const HttpClientContextPtr ctx);
+
+    HttpClientContextPtr getContext(int fd) {
+        return fd < client_ctxs.capacity() ? client_ctxs[fd] : NULL;
+    }
+
+    void addContext(const HttpClientContextPtr& ctx) {
+        int fd = ctx->channel->fd();
+        if (fd >= client_ctxs.capacity()) {
+            client_ctxs.resize(2 * fd);
+        }
+        client_ctxs[fd] = ctx;
+        // NOTE: add into conn_pools after recv response completed
+        // conn_pools[ctx->channel->peeraddr()].add(fd);
+    }
+
+    void removeContext(const HttpClientContextPtr& ctx) {
+        int fd = ctx->channel->fd();
+        // NOTE: remove from conn_pools
+        auto iter = conn_pools.find(ctx->channel->peeraddr());
+        if (iter != conn_pools.end()) {
+            iter->second.remove(fd);
+        }
+        if (fd < client_ctxs.capacity()) {
+            client_ctxs[fd] = NULL;
+        }
+    }
+
+private:
+    EventLoopThread                         loop_thread;
+    // NOTE: just one loop thread, no need mutex.
+    // with fd as index
+    std::vector<HttpClientContextPtr>       client_ctxs;
+    // peeraddr => ConnPool
+    std::map<std::string, ConnPool<int>>    conn_pools;
+};
+
+}
+
+#endif // HV_ASYNC_HTTP_CLIENT_H_

http/client/http_client.cpp

@@ -13,8 +13,8 @@
 #include "HttpParser.h"
 
 // for async
-#include "hthread.h"
-#include "hloop.h"
+#include "AsyncHttpClient.h"
+
 struct http_client_s {
     std::string  host;
     int          port;
@@ -25,14 +25,13 @@ struct http_client_s {
 #ifdef WITH_CURL
     CURL* curl;
 #endif
-    int fd;
     // for sync
+    int             fd;
     hssl_t          ssl;
     HttpParserPtr   parser;
     // for async
-    std::mutex  mutex_;
-    hthread_t   thread_;
-    hloop_t*    loop_;
+    std::mutex                              mutex_;
+    std::shared_ptr<hv::AsyncHttpClient>    async_client_;
 
     http_client_s() {
         host = LOCALHOST;
@@ -44,9 +43,6 @@ struct http_client_s {
 #endif
         fd = -1;
         ssl = NULL;
-
-        thread_ = 0;
-        loop_ = NULL;
     }
 
     ~http_client_s() {
@@ -54,14 +50,6 @@ struct http_client_s {
     }
 
     void Close() {
-        if (loop_) {
-            hloop_stop(loop_);
-            loop_ = NULL;
-        }
-        if (thread_) {
-            hthread_join(thread_);
-            thread_ = 0;
-        }
 #ifdef WITH_CURL
         if (curl) {
             curl_easy_cleanup(curl);
@@ -451,182 +439,18 @@ const char* http_client_strerror(int errcode) {
 }
 #endif
 
-struct HttpContext {
-    HttpRequestPtr  req;
-    HttpResponsePtr resp;
-    HttpParserPtr   parser;
-
-    HttpResponseCallback cb;
-    void*                userdata;
-
-    hio_t*          io;
-    htimer_t*       timer;
-
-    HttpContext() {
-        io = NULL;
-        timer = NULL;
-        cb = NULL;
-        userdata = NULL;
-    }
-
-    ~HttpContext() {
-        killTimer();
-        // keep-alive
-        // closeIO();
-    }
-
-    void closeIO() {
-        if (io) {
-            hio_close(io);
-            io = NULL;
-        }
-    }
-
-    void killTimer() {
-        if (timer) {
-            htimer_del(timer);
-            timer = NULL;
-        }
-    }
-
-    void callback(int state) {
-        if (cb) {
-            cb(state, req, resp, userdata);
-            // NOTE: ensure cb only called once
-            cb = NULL;
-        }
-    }
-
-    void successCallback() {
-        killTimer();
-        callback(0);
-    }
-
-    void errorCallback(int error) {
-        closeIO();
-        callback(error);
-    }
-};
-
-static void on_close(hio_t* io) {
-    HttpContext* ctx = (HttpContext*)hevent_userdata(io);
-    if (ctx) {
-        hevent_set_userdata(io, NULL);
-        ctx->io = NULL;
-        int error = hio_error(io);
-        ctx->callback(error);
-        delete ctx;
-    }
-}
-
-static void on_recv(hio_t* io, void* buf, int readbytes) {
-    HttpContext* ctx = (HttpContext*)hevent_userdata(io);
-
-    int nparse = ctx->parser->FeedRecvData((const char*)buf, readbytes);
-    if (nparse != readbytes) {
-        ctx->errorCallback(ERR_PARSE);
-        return;
-    }
-    if (ctx->parser->IsComplete()) {
-        ctx->successCallback();
-        return;
-    }
-}
-
-static void on_connect(hio_t* io) {
-    HttpContext* ctx = (HttpContext*)hevent_userdata(io);
-
-    ctx->parser = HttpParserPtr(HttpParser::New(HTTP_CLIENT, (http_version)ctx->req->http_major));
-    ctx->parser->InitResponse(ctx->resp.get());
-    ctx->parser->SubmitRequest(ctx->req.get());
-
-    char* data = NULL;
-    size_t len = 0;
-    while (ctx->parser->GetSendData(&data, &len)) {
-        hio_write(io, data, len);
-    }
-
-    hio_setcb_read(io, on_recv);
-    hio_read(io);
-}
-
-static void on_timeout(htimer_t* timer) {
-    HttpContext* ctx = (HttpContext*)hevent_userdata(timer);
-    if (ctx) {
-        hevent_set_userdata(timer, NULL);
-        ctx->timer = NULL;
-        ctx->errorCallback(ERR_TASK_TIMEOUT);
-    }
-}
-
-static HTHREAD_ROUTINE(http_client_loop_thread) {
-    hloop_t* loop = (hloop_t*)userdata;
-    assert(loop != NULL);
-    hloop_run(loop);
-    return 0;
-}
-
-// hloop_new -> htread_create -> hloop_run ->
-// hio_connect -> on_connect -> hio_write -> hio_read -> on_recv ->
-// HttpResponseCallback -> on_close
-static int __http_client_send_async(http_client_t* cli, HttpRequestPtr req, HttpResponsePtr resp,
-        HttpResponseCallback cb, void* userdata) {
-    sockaddr_u peeraddr;
-    memset(&peeraddr, 0, sizeof(peeraddr));
-    req->ParseUrl();
-    int ret = sockaddr_set_ipport(&peeraddr, req->host.c_str(), req->port);
-    if (ret != 0) {
-        return ERR_INVALID_PARAM;
-    }
-    int connfd = socket(peeraddr.sa.sa_family, SOCK_STREAM, 0);
-    if (connfd < 0) {
-        return ERR_SOCKET;
-    }
-
+static int __http_client_send_async(http_client_t* cli, HttpRequestPtr req, HttpResponseCallback resp_cb) {
     cli->mutex_.lock();
-    if (cli->loop_ == NULL) {
-        cli->loop_ = hloop_new(HLOOP_FLAG_AUTO_FREE);
-    }
-    if (cli->thread_ == 0) {
-        cli->thread_ = hthread_create(http_client_loop_thread, cli->loop_);
+    if (cli->async_client_ == NULL) {
+        cli->async_client_.reset(new hv::AsyncHttpClient);
     }
     cli->mutex_.unlock();
 
-    hio_t* connio = hio_get(cli->loop_, connfd);
-    assert(connio != NULL);
-
-    hio_set_peeraddr(connio, &peeraddr.sa, sockaddr_len(&peeraddr));
-    hio_setcb_connect(connio, on_connect);
-    hio_setcb_close(connio, on_close);
-
-    // https
-    if (req->https) {
-        hio_enable_ssl(connio);
-    }
-
-    // new HttpContext
-    // delete on_close
-    HttpContext* ctx = new HttpContext;
-    ctx->req = req;
-    ctx->resp = resp;
-    ctx->cb = cb;
-    ctx->userdata = userdata;
-    ctx->io = connio;
-    hevent_set_userdata(connio, ctx);
-
-    // timeout
-    if (req->timeout > 0) {
-        ctx->timer = htimer_add(cli->loop_, on_timeout, req->timeout * 1000, 1);
-        assert(ctx->timer != NULL);
-        hevent_set_userdata(ctx->timer, ctx);
-    }
-
-    return hio_connect(connio);
+    return cli->async_client_->send(req, resp_cb);
 }
 
-int http_client_send_async(http_client_t* cli, HttpRequestPtr req, HttpResponsePtr resp,
-        HttpResponseCallback cb, void* userdata) {
-    if (!cli || !req || !resp) return ERR_NULL_POINTER;
+int http_client_send_async(http_client_t* cli, HttpRequestPtr req, HttpResponseCallback resp_cb) {
+    if (!cli || !req) return ERR_NULL_POINTER;
 
     if (req->url.empty() || *req->url.c_str() == '/') {
         req->host = cli->host;
@@ -644,17 +468,16 @@ int http_client_send_async(http_client_t* cli, HttpRequestPtr req, HttpResponseP
         }
     }
 
-    return __http_client_send_async(cli, req, resp, cb, userdata);
+    return __http_client_send_async(cli, req, resp_cb);
 }
 
-int http_client_send_async(HttpRequestPtr req, HttpResponsePtr resp,
-        HttpResponseCallback cb, void* userdata) {
-    if (!req || !resp) return ERR_NULL_POINTER;
+int http_client_send_async(HttpRequestPtr req, HttpResponseCallback resp_cb) {
+    if (req == NULL) return ERR_NULL_POINTER;
 
     if (req->timeout == 0) {
         req->timeout = DEFAULT_HTTP_TIMEOUT;
     }
 
     static http_client_t s_default_async_client;
-    return __http_client_send_async(&s_default_async_client, req, resp, cb, userdata);
+    return __http_client_send_async(&s_default_async_client, req, resp_cb);
 }

http/client/http_client.h

@@ -26,7 +26,7 @@ int main(int argc, char* argv[]) {
 }
 */
 
-#define DEFAULT_HTTP_TIMEOUT    10 // s
+#define DEFAULT_HTTP_TIMEOUT    30 // s
 typedef struct http_client_s http_client_t;
 
 HV_EXPORT http_client_t* http_client_new(const char* host = NULL, int port = DEFAULT_HTTP_PORT, int https = 0);
@@ -45,16 +45,14 @@ HV_EXPORT const char* http_client_get_header(http_client_t* cli, const char* key
 HV_EXPORT int http_client_send(http_client_t* cli, HttpRequest* req, HttpResponse* resp);
 
 // async
-// Intern will start an event-loop thread when http_client_send_async first called,
+// Intern will start an EventLoopThread when http_client_send_async first called,
 // http_client_del will destroy the thread.
-HV_EXPORT int http_client_send_async(http_client_t* cli, HttpRequestPtr req, HttpResponsePtr resp,
-                                    HttpResponseCallback cb = NULL, void* userdata = NULL);
+HV_EXPORT int http_client_send_async(http_client_t* cli, HttpRequestPtr req, HttpResponseCallback resp_cb = NULL);
 
 // top-level api
 // http_client_new -> http_client_send -> http_client_del
 HV_EXPORT int http_client_send(HttpRequest* req, HttpResponse* resp);
 // http_client_send_async(&default_async_client, ...)
-HV_EXPORT int http_client_send_async(HttpRequestPtr req, HttpResponsePtr resp,
-                                    HttpResponseCallback cb = NULL, void* userdata = NULL);
+HV_EXPORT int http_client_send_async(HttpRequestPtr req, HttpResponseCallback resp_cb = NULL);
 
 #endif  // HTTP_CLIENT_H_