mirror of
https://github.com/sogou/workflow.git
synced 2026-10-26 09:39:24 +08:00
@@ -278,7 +278,7 @@ void WFComplexClientTask<REQ, RESP, CTX>::init(TransportType type,
|
||||
if (redirect_)
|
||||
clear_prev_state();
|
||||
|
||||
const auto *params = &WFGlobal::get_global_settings()->endpoint_params;
|
||||
const auto params = WFGlobal::get_global_settings()->endpoint_params;
|
||||
struct addrinfo addrinfo = { };
|
||||
addrinfo.ai_family = addr->sa_family;
|
||||
addrinfo.ai_socktype = SOCK_STREAM;
|
||||
@@ -287,8 +287,9 @@ void WFComplexClientTask<REQ, RESP, CTX>::init(TransportType type,
|
||||
|
||||
type_ = type;
|
||||
info_.assign(info);
|
||||
if (WFGlobal::get_route_manager()->get(type, &addrinfo, info_, params,
|
||||
route_result_) < 0)
|
||||
params.use_tls_sni = false;
|
||||
if (WFGlobal::get_route_manager()->get(type, &addrinfo, info_, ¶ms,
|
||||
"", route_result_) < 0)
|
||||
{
|
||||
this->state = WFT_STATE_SYS_ERROR;
|
||||
this->error = errno;
|
||||
|
||||
+16
-16
@@ -871,9 +871,7 @@ void Communicator::handle_listen_result(struct poller_result *res)
|
||||
CommService *service = (CommService *)res->data.context;
|
||||
struct CommConnEntry *entry;
|
||||
CommServiceTarget *target;
|
||||
struct poller_data data;
|
||||
int timeout;
|
||||
int ret;
|
||||
|
||||
switch (res->state)
|
||||
{
|
||||
@@ -884,30 +882,29 @@ void Communicator::handle_listen_result(struct poller_result *res)
|
||||
{
|
||||
if (service->ssl_ctx)
|
||||
{
|
||||
ret = __create_ssl(service->ssl_ctx, entry);
|
||||
if (ret >= 0)
|
||||
if (__create_ssl(service->ssl_ctx, entry) >= 0 &&
|
||||
service->init_ssl(entry->ssl) >= 0)
|
||||
{
|
||||
data.operation = PD_OP_SSL_ACCEPT;
|
||||
res->data.operation = PD_OP_SSL_ACCEPT;
|
||||
timeout = service->ssl_accept_timeout;
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
ret = 0;
|
||||
data.operation = PD_OP_READ;
|
||||
data.message = NULL;
|
||||
res->data.operation = PD_OP_READ;
|
||||
res->data.message = NULL;
|
||||
timeout = target->response_timeout;
|
||||
}
|
||||
|
||||
if (ret >= 0)
|
||||
if (res->data.operation != PD_OP_LISTEN)
|
||||
{
|
||||
data.fd = entry->sockfd;
|
||||
data.ssl = entry->ssl;
|
||||
data.context = entry;
|
||||
if (mpoller_add(&data, timeout, this->mpoller) >= 0)
|
||||
res->data.fd = entry->sockfd;
|
||||
res->data.ssl = entry->ssl;
|
||||
res->data.context = entry;
|
||||
if (mpoller_add(&res->data, timeout, this->mpoller) >= 0)
|
||||
{
|
||||
if (this->stop_flag)
|
||||
mpoller_del(data.fd, this->mpoller);
|
||||
mpoller_del(res->data.fd, this->mpoller);
|
||||
break;
|
||||
}
|
||||
}
|
||||
@@ -945,13 +942,16 @@ void Communicator::handle_connect_result(struct poller_result *res)
|
||||
case PR_ST_FINISHED:
|
||||
if (target->ssl_ctx && !entry->ssl)
|
||||
{
|
||||
ret = __create_ssl(target->ssl_ctx, entry);
|
||||
if (ret >= 0)
|
||||
if (__create_ssl(target->ssl_ctx, entry) >= 0 &&
|
||||
target->init_ssl(entry->ssl) >= 0)
|
||||
{
|
||||
ret = 0;
|
||||
res->data.operation = PD_OP_SSL_CONNECT;
|
||||
res->data.ssl = entry->ssl;
|
||||
timeout = target->ssl_connect_timeout;
|
||||
}
|
||||
else
|
||||
ret = -1;
|
||||
}
|
||||
else if ((session->out = session->message_out()) != NULL)
|
||||
{
|
||||
|
||||
@@ -67,6 +67,8 @@ private:
|
||||
return new CommConnection;
|
||||
}
|
||||
|
||||
virtual int init_ssl(SSL *ssl) { return 0; }
|
||||
|
||||
public:
|
||||
virtual void release() { }
|
||||
|
||||
@@ -195,6 +197,8 @@ private:
|
||||
return new CommConnection;
|
||||
}
|
||||
|
||||
virtual int init_ssl(SSL *ssl) { return 0; }
|
||||
|
||||
private:
|
||||
struct sockaddr *bind_addr;
|
||||
socklen_t addrlen;
|
||||
|
||||
@@ -1505,7 +1505,8 @@ void Communicator::handle_connect_result(struct poller_result *res)
|
||||
{
|
||||
if (target->ssl_ctx)
|
||||
{
|
||||
if (__ssl_connect(target->ssl_ctx, entry) >= 0)
|
||||
if (__ssl_connect(target->ssl_ctx, entry) >= 0 &&
|
||||
target->init_ssl(entry->ssl) >= 0)
|
||||
{
|
||||
if (handle_incoming_ssl_connect(entry) == false)
|
||||
return;
|
||||
@@ -1578,7 +1579,9 @@ void Communicator::handle_accept_result(struct poller_result *res)
|
||||
else
|
||||
timeout = target->response_timeout;
|
||||
|
||||
if (!service->ssl_ctx || __ssl_accept(service->ssl_ctx, entry) >= 0)
|
||||
if (!service->ssl_ctx ||
|
||||
(__ssl_accept(service->ssl_ctx, entry) >= 0 &&
|
||||
service->init_ssl(entry->ssl) >= 0))
|
||||
{
|
||||
if (this->poller->put_io(&data, timeout) >= 0)
|
||||
break;
|
||||
@@ -1686,7 +1689,7 @@ SOCKET Communicator::nonblock_connect(CommTarget *target)
|
||||
|
||||
SOCKET Communicator::nonblock_accept(CommService *service)
|
||||
{
|
||||
SOCKET sockfd = service->create_accept_fd();
|
||||
SOCKET sockfd = (SOCKET)service->create_accept_fd();
|
||||
|
||||
if (sockfd != INVALID_SOCKET)
|
||||
{
|
||||
|
||||
@@ -71,6 +71,8 @@ private:
|
||||
return new CommConnection;
|
||||
}
|
||||
|
||||
virtual int init_ssl(SSL *ssl) { return 0; }
|
||||
|
||||
public:
|
||||
virtual void release() { }
|
||||
|
||||
@@ -199,11 +201,13 @@ private:
|
||||
return new CommConnection;
|
||||
}
|
||||
|
||||
SOCKET create_accept_fd()
|
||||
virtual int create_accept_fd()
|
||||
{
|
||||
return socket(this->bind_addr->sa_family, SOCK_STREAM, 0);
|
||||
return (int)socket(this->bind_addr->sa_family, SOCK_STREAM, 0);
|
||||
}
|
||||
|
||||
virtual int init_ssl(SSL *ssl) { return 0; }
|
||||
|
||||
private:
|
||||
struct sockaddr *bind_addr;
|
||||
socklen_t addrlen;
|
||||
|
||||
@@ -41,6 +41,7 @@ struct EndpointParams
|
||||
int connect_timeout;
|
||||
int response_timeout;
|
||||
int ssl_connect_timeout;
|
||||
bool use_tls_sni;
|
||||
};
|
||||
|
||||
static constexpr struct EndpointParams ENDPOINT_PARAMS_DEFAULT =
|
||||
@@ -48,7 +49,8 @@ static constexpr struct EndpointParams ENDPOINT_PARAMS_DEFAULT =
|
||||
/* .max_connections = */ 200,
|
||||
/* .connect_timeout = */ 10 * 1000,
|
||||
/* .response_timeout = */ 10 * 1000,
|
||||
/* .ssl_connect_timeout = */ 10 * 1000
|
||||
/* .ssl_connect_timeout = */ 10 * 1000,
|
||||
/* .use_tls_sni = */ false
|
||||
};
|
||||
|
||||
#endif
|
||||
|
||||
@@ -67,6 +67,27 @@ private:
|
||||
}
|
||||
};
|
||||
|
||||
/* To support TLS SNI. */
|
||||
class RouteTargetSNI : public RouteManager::RouteTarget
|
||||
{
|
||||
private:
|
||||
virtual int init_ssl(SSL *ssl)
|
||||
{
|
||||
if (SSL_set_tlsext_host_name(ssl, this->hostname.c_str()) > 0)
|
||||
return 0;
|
||||
else
|
||||
return -1;
|
||||
}
|
||||
|
||||
private:
|
||||
std::string hostname;
|
||||
|
||||
public:
|
||||
RouteTargetSNI(const std::string& name) : hostname(name)
|
||||
{
|
||||
}
|
||||
};
|
||||
|
||||
// protocol_name\n user\n pass\n dbname\n ai_addr ai_addrlen \n....
|
||||
//
|
||||
|
||||
@@ -80,6 +101,8 @@ struct RouteParams
|
||||
int ssl_connect_timeout;
|
||||
int response_timeout;
|
||||
size_t max_connections;
|
||||
bool use_tls_sni;
|
||||
const std::string& hostname;
|
||||
};
|
||||
|
||||
class RouteResultEntry
|
||||
@@ -133,9 +156,12 @@ CommSchedTarget *RouteResultEntry::create_target(const struct RouteParams *param
|
||||
|
||||
switch (params->transport_type)
|
||||
{
|
||||
case TT_TCP:
|
||||
case TT_TCP_SSL:
|
||||
target = new RouteTargetTCP();
|
||||
if (params->use_tls_sni)
|
||||
target = new RouteTargetSNI(params->hostname);
|
||||
else
|
||||
case TT_TCP:
|
||||
target = new RouteTargetTCP();
|
||||
break;
|
||||
case TT_UDP:
|
||||
target = new RouteTargetUDP();
|
||||
@@ -350,7 +376,9 @@ static inline bool __addr_less(const struct addrinfo *x, const struct addrinfo *
|
||||
|
||||
static uint64_t __generate_key(TransportType type,
|
||||
const struct addrinfo *addrinfo,
|
||||
const std::string& other_info)
|
||||
const std::string& other_info,
|
||||
const struct EndpointParams *endpoint_params,
|
||||
const std::string& hostname)
|
||||
{
|
||||
std::string str = "TT";
|
||||
|
||||
@@ -362,6 +390,12 @@ static uint64_t __generate_key(TransportType type,
|
||||
str += '\n';
|
||||
}
|
||||
|
||||
if (type == TT_TCP_SSL && endpoint_params->use_tls_sni)
|
||||
{
|
||||
str += hostname;
|
||||
str += '\n';
|
||||
}
|
||||
|
||||
std::vector<const struct addrinfo *> sorted_addr;
|
||||
|
||||
for (const struct addrinfo *p = addrinfo; p; p = p->ai_next)
|
||||
@@ -394,6 +428,7 @@ int RouteManager::get(TransportType type,
|
||||
const struct addrinfo *addrinfo,
|
||||
const std::string& other_info,
|
||||
const struct EndpointParams *endpoint_params,
|
||||
const std::string& hostname,
|
||||
RouteResult& result)
|
||||
{
|
||||
result.cookie = NULL;
|
||||
@@ -404,7 +439,8 @@ int RouteManager::get(TransportType type,
|
||||
return -1;
|
||||
}
|
||||
|
||||
uint64_t md5_16 = __generate_key(type, addrinfo, other_info);
|
||||
uint64_t md5_16 = __generate_key(type, addrinfo, other_info,
|
||||
endpoint_params, hostname);
|
||||
rb_node **p = &cache_.rb_node;
|
||||
rb_node *parent = NULL;
|
||||
RouteResultEntry *entry;
|
||||
@@ -439,8 +475,7 @@ int RouteManager::get(TransportType type,
|
||||
ssl_connect_timeout = endpoint_params->ssl_connect_timeout;
|
||||
}
|
||||
|
||||
struct RouteParams params =
|
||||
{
|
||||
struct RouteParams params = {
|
||||
/* .transport_type = */ type,
|
||||
/* .addrinfo = */ addrinfo,
|
||||
/* .md5_16 = */ md5_16,
|
||||
@@ -448,8 +483,11 @@ int RouteManager::get(TransportType type,
|
||||
/* .connect_timeout = */ endpoint_params->connect_timeout,
|
||||
/* .ssl_connect_timeout = */ ssl_connect_timeout,
|
||||
/* .response_timeout = */ endpoint_params->response_timeout,
|
||||
/* .max_connections = */ endpoint_params->max_connections
|
||||
/* .max_connections = */ endpoint_params->max_connections,
|
||||
/* .use_tls_sni = */ endpoint_params->use_tls_sni,
|
||||
/* .hostname = */ hostname,
|
||||
};
|
||||
|
||||
if (StringUtil::start_with(other_info, "?maxconn="))
|
||||
{
|
||||
int maxconn = atoi(other_info.c_str() + 9);
|
||||
@@ -460,7 +498,7 @@ int RouteManager::get(TransportType type,
|
||||
params.max_connections = maxconn;
|
||||
}
|
||||
|
||||
entry = new RouteResultEntry();
|
||||
entry = new RouteResultEntry;
|
||||
if (entry->init(¶ms) >= 0)
|
||||
{
|
||||
rb_link_node(&entry->rb, parent, p);
|
||||
|
||||
@@ -61,6 +61,7 @@ public:
|
||||
const struct addrinfo *addrinfo,
|
||||
const std::string& other_info,
|
||||
const struct EndpointParams *endpoint_params,
|
||||
const std::string& hostname,
|
||||
RouteResult& result);
|
||||
|
||||
RouteManager()
|
||||
|
||||
@@ -125,7 +125,7 @@ void WFResolverTask::dispatch()
|
||||
}
|
||||
|
||||
if (route_manager->get(type_, addrinfo, info_, &endpoint_params_,
|
||||
this->result) < 0)
|
||||
host_, this->result) < 0)
|
||||
{
|
||||
this->state = WFT_STATE_SYS_ERROR;
|
||||
this->error = errno;
|
||||
@@ -228,7 +228,7 @@ void WFResolverTask::dns_callback_internal(DNSOutput *dns_out,
|
||||
(unsigned int)ttl_default,
|
||||
(unsigned int)ttl_min);
|
||||
if (route_manager->get(type_, addrinfo, info_, &endpoint_params_,
|
||||
this->result) < 0)
|
||||
host_, this->result) < 0)
|
||||
{
|
||||
this->state = WFT_STATE_SYS_ERROR;
|
||||
this->error = errno;
|
||||
|
||||
Reference in New Issue
Block a user