workflow源码解析：http客户端（一）

1、示例程序

程序从stdin读取http/https URL，抓取网页并把内容打印到stdout，并将请求和响应的http header打印在stderr。
为了简单起见，程序用Ctrl-C退出，但会保证所有资源先被完全释放。

#include <signal.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <string>
#include "workflow/HttpMessage.h"
#include "workflow/HttpUtil.h"
#include "workflow/WFTaskFactory.h"#ifndef _WIN32
#include <unistd.h>
#endif#define REDIRECT_MAX    5
#define RETRY_MAX       2void wget_callback(WFHttpTask *task)
{protocol::HttpRequest *req = task->get_req();protocol::HttpResponse *resp = task->get_resp();int state = task->get_state();int error = task->get_error();switch (state){case WFT_STATE_SYS_ERROR:fprintf(stderr, "system error: %s\n", strerror(error));break;case WFT_STATE_DNS_ERROR:fprintf(stderr, "DNS error: %s\n", gai_strerror(error));break;case WFT_STATE_SSL_ERROR:fprintf(stderr, "SSL error: %d\n", error);break;case WFT_STATE_TASK_ERROR:fprintf(stderr, "Task error: %d\n", error);break;case WFT_STATE_SUCCESS:break;}if (state != WFT_STATE_SUCCESS){fprintf(stderr, "Failed. Press Ctrl-C to exit.\n");return;}std::string name;std::string value;/* Print request. */fprintf(stderr, "%s %s %s\r\n", req->get_method(),req->get_http_version(),req->get_request_uri());protocol::HttpHeaderCursor req_cursor(req);while (req_cursor.next(name, value))fprintf(stderr, "%s: %s\r\n", name.c_str(), value.c_str());fprintf(stderr, "\r\n");/* Print response header. */fprintf(stderr, "%s %s %s\r\n", resp->get_http_version(),resp->get_status_code(),resp->get_reason_phrase());protocol::HttpHeaderCursor resp_cursor(resp);while (resp_cursor.next(name, value))fprintf(stderr, "%s: %s\r\n", name.c_str(), value.c_str());fprintf(stderr, "\r\n");/* Print response body. */const void *body;size_t body_len;resp->get_parsed_body(&body, &body_len);fwrite(body, 1, body_len, stdout);fflush(stdout);fprintf(stderr, "\nSuccess. Press Ctrl-C to exit.\n");
}void sig_handler(int signo) { }int main(int argc, char *argv[])
{WFHttpTask *task;if (argc != 2){fprintf(stderr, "USAGE: %s <http URL>\n", argv[0]);exit(1);}signal(SIGINT, sig_handler);std::string url = argv[1];if (strncasecmp(argv[1], "http://", 7) != 0 &&strncasecmp(argv[1], "https://", 8) != 0){url = "http://" + url;}task = WFTaskFactory::create_http_task(url, REDIRECT_MAX, RETRY_MAX,wget_callback);protocol::HttpRequest *req = task->get_req();req->add_header_pair("Accept", "*/*");req->add_header_pair("User-Agent", "Wget/1.14 (linux-gnu)");req->add_header_pair("Connection", "close");task->start();
#ifndef _WIN32pause();
#elsegetchar();
#endifreturn 0;
}

2、类继承关系

在这里插入图片描述

3、源码分析

create_http_task

#HttpTaskImpl.cc
WFHttpTask *WFTaskFactory::create_http_task(const std::string& url,int redirect_max,int retry_max,http_callback_t callback)
{auto *task = new ComplexHttpTask(redirect_max,retry_max,std::move(callback));ParsedURI uri;URIParser::parse(url, uri);task->init(std::move(uri));task->set_keep_alive(HTTP_KEEPALIVE_DEFAULT);return task;
}

ComplexHttpTask

#HttpTaskImpl.cc
class ComplexHttpTask : public WFComplexClientTask<HttpRequest, HttpResponse>
{
public:ComplexHttpTask(int redirect_max,int retry_max,http_callback_t&& callback):WFComplexClientTask(retry_max, std::move(callback)),redirect_max_(redirect_max),redirect_count_(0){HttpRequest *client_req = this->get_req();client_req->set_method(HttpMethodGet);client_req->set_http_version("HTTP/1.1");}protected:virtual CommMessageOut *message_out();virtual CommMessageIn *message_in();virtual int keep_alive_timeout();virtual bool init_success();virtual void init_failed();virtual bool finish_once();protected:bool need_redirect(ParsedURI& uri);bool redirect_url(HttpResponse *client_resp, ParsedURI& uri);void set_empty_request();void check_response();private:int redirect_max_;int redirect_count_;
};

WFComplexClientTask

#WFTaskFactory.inl
template<class REQ, class RESP, typename CTX = bool>
class WFComplexClientTask : public WFClientTask<REQ, RESP>
{
protected:using task_callback_t = std::function<void (WFNetworkTask<REQ, RESP> *)>;public:WFComplexClientTask(int retry_max, task_callback_t&& cb):WFClientTask<REQ, RESP>(NULL, WFGlobal::get_scheduler(), std::move(cb)){type_ = TT_TCP;fixed_addr_ = false;retry_max_ = retry_max;retry_times_ = 0;redirect_ = false;ns_policy_ = NULL;router_task_ = NULL;}protected:// new api for childrenvirtual bool init_success() { return true; }virtual void init_failed() {}virtual bool check_request() { return true; }virtual WFRouterTask *route();virtual bool finish_once() { return true; }public:void init(const ParsedURI& uri){uri_ = uri;init_with_uri();}void init(ParsedURI&& uri){uri_ = std::move(uri);init_with_uri();}void init(TransportType type,const struct sockaddr *addr,socklen_t addrlen,const std::string& info);void set_transport_type(TransportType type){type_ = type;}TransportType get_transport_type() const { return type_; }virtual const ParsedURI *get_current_uri() const { return &uri_; }void set_redirect(const ParsedURI& uri){redirect_ = true;init(uri);}void set_redirect(TransportType type, const struct sockaddr *addr,socklen_t addrlen, const std::string& info){redirect_ = true;init(type, addr, addrlen, info);}bool is_fixed_addr() const { return this->fixed_addr_; }protected:void set_fixed_addr(int fixed) { this->fixed_addr_ = fixed; }void set_info(const std::string& info){info_.assign(info);}void set_info(const char *info){info_.assign(info);}protected:virtual void dispatch();virtual SubTask *done();void clear_resp(){size_t size = this->resp.get_size_limit();this->resp.~RESP();new(&this->resp) RESP();this->resp.set_size_limit(size);}void disable_retry(){retry_times_ = retry_max_;}protected:TransportType type_;ParsedURI uri_;std::string info_;bool fixed_addr_;bool redirect_;CTX ctx_;int retry_max_;int retry_times_;WFNSPolicy *ns_policy_;WFRouterTask *router_task_;RouteManager::RouteResult route_result_;WFNSTracing tracing_;public:CTX *get_mutable_ctx() { return &ctx_; }private:void clear_prev_state();void init_with_uri();bool set_port();void router_callback(void *t);void switch_callback(void *t);
};

WFClientTask

#WFTask.inl
template<class REQ, class RESP>
class WFClientTask : public WFNetworkTask<REQ, RESP>
{
protected:virtual CommMessageOut *message_out(){/* By using prepare function, users can modify request after* the connection is established. */if (this->prepare)this->prepare(this);return &this->req;}virtual CommMessageIn *message_in() { return &this->resp; }protected:virtual WFConnection *get_connection() const{CommConnection *conn;if (this->target){conn = this->CommSession::get_connection();if (conn)return (WFConnection *)conn;}errno = ENOTCONN;return NULL;}protected:virtual SubTask *done(){SeriesWork *series = series_of(this);if (this->state == WFT_STATE_SYS_ERROR && this->error < 0){this->state = WFT_STATE_SSL_ERROR;this->error = -this->error;}if (this->callback)this->callback(this);delete this;return series->pop();}public:void set_prepare(std::function<void (WFNetworkTask<REQ, RESP> *)> prep){this->prepare = std::move(prep);}protected:std::function<void (WFNetworkTask<REQ, RESP> *)> prepare;public:WFClientTask(CommSchedObject *object, CommScheduler *scheduler,std::function<void (WFNetworkTask<REQ, RESP> *)>&& cb) :WFNetworkTask<REQ, RESP>(object, scheduler, std::move(cb)){}protected:virtual ~WFClientTask() { }
};

WFNetworkTask

#WFTask.h
template<class REQ, class RESP>
class WFNetworkTask : public CommRequest
{
public:/* start(), dismiss() are for client tasks only. */void start(){assert(!series_of(this));Workflow::start_series_work(this, nullptr);}void dismiss(){assert(!series_of(this));delete this;}public:REQ *get_req() { return &this->req; }RESP *get_resp() { return &this->resp; }public:void *user_data;public:int get_state() const { return this->state; }int get_error() const { return this->error; }/* Call when error is ETIMEDOUT, return values:* TOR_NOT_TIMEOUT, TOR_WAIT_TIMEOUT, TOR_CONNECT_TIMEOUT,* TOR_TRANSMIT_TIMEOUT (send or receive).* SSL connect timeout also returns TOR_CONNECT_TIMEOUT. */int get_timeout_reason() const { return this->timeout_reason; }/* Call only in callback or server's process. */long long get_task_seq() const{if (!this->target){errno = ENOTCONN;return -1;}return this->get_seq();}int get_peer_addr(struct sockaddr *addr, socklen_t *addrlen) const;virtual WFConnection *get_connection() const = 0;public:/* All in milliseconds. timeout == -1 for unlimited. */void set_send_timeout(int timeout) { this->send_timeo = timeout; }void set_receive_timeout(int timeout) { this->receive_timeo = timeout; }void set_keep_alive(int timeout) { this->keep_alive_timeo = timeout; }public:/* Do not reply this request. */void noreply(){if (this->state == WFT_STATE_TOREPLY)this->state = WFT_STATE_NOREPLY;}/* Push reply data synchronously. */virtual int push(const void *buf, size_t size){return this->scheduler->push(buf, size, this);}/* To check if the connection was closed before replying.Always returns 'true' in callback. */bool closed() const{if (this->state == WFT_STATE_TOREPLY)return !this->get_target()->has_idle_conn();elsereturn this->state != WFT_STATE_UNDEFINED;}public:void set_callback(std::function<void (WFNetworkTask<REQ, RESP> *)> cb){this->callback = std::move(cb);}protected:virtual int send_timeout() { return this->send_timeo; }virtual int receive_timeout() { return this->receive_timeo; }virtual int keep_alive_timeout() { return this->keep_alive_timeo; }protected:int send_timeo;int receive_timeo;int keep_alive_timeo;REQ req;RESP resp;std::function<void (WFNetworkTask<REQ, RESP> *)> callback;protected:WFNetworkTask(CommSchedObject *object, CommScheduler *scheduler,std::function<void (WFNetworkTask<REQ, RESP> *)>&& cb) :CommRequest(object, scheduler),callback(std::move(cb)){this->send_timeo = -1;this->receive_timeo = -1;this->keep_alive_timeo = 0;this->target = NULL;this->timeout_reason = TOR_NOT_TIMEOUT;this->user_data = NULL;this->state = WFT_STATE_UNDEFINED;this->error = 0;}virtual ~WFNetworkTask() { }
};

CommRequest

# CommRequest.h
class CommRequest : public SubTask, public CommSession
{
public:CommRequest(CommSchedObject *object, CommScheduler *scheduler){this->scheduler = scheduler;this->object = object;this->wait_timeout = 0;}CommSchedObject *get_request_object() const { return this->object; }void set_request_object(CommSchedObject *object) { this->object = object; }int get_wait_timeout() const { return this->wait_timeout; }void set_wait_timeout(int timeout) { this->wait_timeout = timeout; }public:virtual void dispatch(){if (this->scheduler->request(this, this->object, this->wait_timeout,&this->target) < 0){this->handle(CS_STATE_ERROR, errno);}}protected:int state;int error;protected:CommTarget *target;
#define TOR_NOT_TIMEOUT			0
#define TOR_WAIT_TIMEOUT		1
#define TOR_CONNECT_TIMEOUT		2
#define TOR_TRANSMIT_TIMEOUT	3int timeout_reason;protected:int wait_timeout;CommSchedObject *object;CommScheduler *scheduler;protected:virtual void handle(int state, int error);
};

(1)应用程序调用start()时，调用到 WFNetworkTask类的

`void start(){assert(!series_of(this));Workflow::start_series_work(this, nullptr);}`

（2）

inline void
Workflow::start_series_work(SubTask *first, series_callback_t callback)
{new SeriesWork(first, std::move(callback));first->dispatch();
}

(3)调用 WFComplexClientTask<REQ, RESP, CTX>::dispatch() （这里为什么不是调用 CommRequest 的 dispatch）

template<class REQ, class RESP, typename CTX>
void WFComplexClientTask<REQ, RESP, CTX>::dispatch()
{switch (this->state){case WFT_STATE_UNDEFINED:       // 第一次是这个状态if (this->check_request())   // 这里直接return true {if (this->route_result_.request_object)    // 第一次走着初始化是空的，直接到下面产生router_task_{case WFT_STATE_SUCCESS:   // 第二次就直接success了this->set_request_object(route_result_.request_object);this->WFClientTask<REQ, RESP>::dispatch(); //这里会调用到 CommRequest 的 dispatchreturn;}// 第一次直接过来了，产生route做dns解析// 产生一个router_task_插入到前面去做dns解析router_task_ = this->route();series_of(this)->push_front(this);series_of(this)->push_front(router_task_);}default:break;}this->subtask_done();
}

（4）CommRequest::dispatch 组成
dns解析完后，

this->WFClientTask<REQ, RESP>::dispatch();

调用CommRequest的dispatch

void CommRequest::dispatch()
{// 发送请求this->scheduler->request(this, this->object, this->wait_timeout,&this->target);...
}

（5）scheduler是在 WFComplexClientTask的初始化列表中创建，此时传入的是NULL

WFComplexClientTask(int retry_max, task_callback_t&& cb):WFClientTask<REQ, RESP>(NULL, WFGlobal::get_scheduler(), std::move(cb))

而什么时候才初始化呢? 在WFComplexClientTask的dispatch中set_request_object

template<class REQ, class RESP, typename CTX>
void WFComplexClientTask<REQ, RESP, CTX>::dispatch()
{switch (this->state){case WFT_STATE_UNDEFINED:if (this->check_request()){if (this->route_result_.request_object){case WFT_STATE_SUCCESS:this->set_request_object(route_result_.request_object);this->WFClientTask<REQ, RESP>::dispatch();return;}...
}

这里如何产生的route_result_.request_object，是通过dns去做的，这里先略过，在dns那一节详细阐述。

可以看出我们给个url，dns解析出来之后，我们有了request的目标了

（6）scheduler 的reques执行的是

/* wait_timeout in microseconds, -1 for no timeout. */int request(CommSession *session, CommSchedObject *object,int wait_timeout, CommTarget **target){int ret = -1;*target = object->acquire(wait_timeout);  //获取通信targetif (*target){ret = this->comm.request(session, *target); // 调用request去发request请求if (ret < 0)(*target)->release();}return ret;}