摘要:兩個(gè)函數(shù)是可選回調(diào)函數(shù)。附帶了一組可信任證書(shū)。應(yīng)該注意的是,驗(yàn)證失敗并不意味著連接不能使用。在對(duì)證書(shū)進(jìn)行驗(yàn)證時(shí),有一些安全性檢查并沒(méi)有執(zhí)行,包括證書(shū)的失效檢查和對(duì)證書(shū)中通用名的有效性驗(yàn)證。
前言
swoole_client 提供了 tcp/udp socket 的客戶(hù)端的封裝代碼,使用時(shí)僅需 new swoole_client 即可。 swoole 的 socket client 對(duì)比 PHP 提供的 stream 族函數(shù)有哪些好處:
stream 函數(shù)存在超時(shí)設(shè)置的陷阱和 Bug,一旦沒(méi)處理好會(huì)導(dǎo)致 Server 端長(zhǎng)時(shí)間阻塞
fread 有 8192 長(zhǎng)度限制,無(wú)法支持 UDP 的大包
swoole_client 支持 waitall,在有確定包長(zhǎng)度時(shí)可一次取完,不必循環(huán)讀取
swoole_client 支持 UDP connect,解決了 UDP 串包問(wèn)題
swoole_client 是純 C 的代碼,專(zhuān)門(mén)處理 socket,stream 函數(shù)非常復(fù)雜。swoole_client 性能更好
除了普通的同步阻塞+select 的使用方法外,swoole_client 還支持異步非阻塞回調(diào)。
swoole_client::__construct 構(gòu)造函數(shù)構(gòu)造函數(shù)的邏輯很簡(jiǎn)單,就是更新 swoole_client_class_entry_ptr 指針的 type 屬性與 key 屬性。
當(dāng) type 是異步客戶(hù)端的時(shí)候,不能在 CLI 模式下使用 SWOOLE_KEEP。
#define php_swoole_socktype(type) (type & (~SW_FLAG_SYNC) & (~SW_FLAG_ASYNC) & (~SW_FLAG_KEEP) & (~SW_SOCK_SSL)) static PHP_METHOD(swoole_client, __construct) { long async = 0; long type = 0; char *id = NULL; zend_size_t len = 0; if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "l|ls", &type, &async, &id, &len) == FAILURE) { swoole_php_fatal_error(E_ERROR, "socket type param is required."); RETURN_FALSE; } if (async == 1) { type |= SW_FLAG_ASYNC; } if ((type & SW_FLAG_ASYNC)) { if ((type & SW_FLAG_KEEP) && SWOOLE_G(cli)) { swoole_php_fatal_error(E_ERROR, "The "SWOOLE_KEEP" flag can only be used in the php-fpm or apache environment."); } php_swoole_check_reactor(); } int client_type = php_swoole_socktype(type); if (client_type < SW_SOCK_TCP || client_type > SW_SOCK_UNIX_STREAM) { swoole_php_fatal_error(E_ERROR, "Unknown client type "%d".", client_type); } zend_update_property_long(swoole_client_class_entry_ptr, getThis(), ZEND_STRL("type"), type TSRMLS_CC); if (id) { zend_update_property_stringl(swoole_client_class_entry_ptr, getThis(), ZEND_STRL("id"), id, len TSRMLS_CC); } else { zend_update_property_null(swoole_client_class_entry_ptr, getThis(), ZEND_STRL("id") TSRMLS_CC); } //init swoole_set_object(getThis(), NULL); swoole_set_property(getThis(), client_property_callback, NULL); #ifdef SWOOLE_SOCKETS_SUPPORT swoole_set_property(getThis(), client_property_socket, NULL); #endif RETURN_TRUE; }swoole_client->set 屬性設(shè)置
設(shè)置客戶(hù)端參數(shù),必須在 connect 前執(zhí)行。本函數(shù)用于從 zend 中讀出 client 的屬性,并且和用戶(hù)的配置參數(shù)合并。
static PHP_METHOD(swoole_client, set) { zval *zset; if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "z", &zset) == FAILURE) { return; } if (Z_TYPE_P(zset) != IS_ARRAY) { RETURN_FALSE; } zval *zsetting = php_swoole_read_init_property(swoole_client_class_entry_ptr, getThis(), ZEND_STRL("setting") TSRMLS_CC); sw_php_array_merge(Z_ARRVAL_P(zsetting), Z_ARRVAL_P(zset)); RETURN_TRUE; } static sw_inline zval* php_swoole_read_init_property(zend_class_entry *scope, zval *object, const char *p, size_t pl TSRMLS_DC) { zval *property = sw_zend_read_property(scope, object, p, pl, 1 TSRMLS_CC); if (property == NULL || ZVAL_IS_NULL(property)) { SW_MAKE_STD_ZVAL(property); array_init(property); zend_update_property(scope, object, p, pl, property TSRMLS_CC); sw_zval_ptr_dtor(&property); return sw_zend_read_property(scope, object, p, pl, 1 TSRMLS_CC); } else { return property; } }swoole_client->on 注冊(cè)異步事件回調(diào)函數(shù)
注冊(cè)異步事件回調(diào)函數(shù),函數(shù)主要更新 swoole_client_class_entry_ptr 中的各個(gè)屬性,并將其屬性回調(diào)函數(shù)賦值給 client_property_callback 當(dāng)中。
static PHP_METHOD(swoole_client, on) { char *cb_name; zend_size_t cb_name_len; zval *zcallback; if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "sz", &cb_name, &cb_name_len, &zcallback) == FAILURE) { return; } zval *ztype = sw_zend_read_property(swoole_client_class_entry_ptr, getThis(), SW_STRL("type")-1, 0 TSRMLS_CC); client_callback *cb = (client_callback *) swoole_get_property(getThis(), client_property_callback); if (!cb) { cb = (client_callback *) emalloc(sizeof(client_callback)); bzero(cb, sizeof(client_callback)); swoole_set_property(getThis(), client_property_callback, cb); } if (strncasecmp("connect", cb_name, cb_name_len) == 0) { zend_update_property(swoole_client_class_entry_ptr, getThis(), ZEND_STRL("onConnect"), zcallback TSRMLS_CC); cb->onConnect = sw_zend_read_property(swoole_client_class_entry_ptr, getThis(), ZEND_STRL("onConnect"), 0 TSRMLS_CC); sw_copy_to_stack(cb->onConnect, cb->_onConnect); #ifdef PHP_SWOOLE_ENABLE_FASTCALL cb->cache_onConnect = func_cache; #endif } else if (strncasecmp("receive", cb_name, cb_name_len) == 0) { zend_update_property(swoole_client_class_entry_ptr, getThis(), ZEND_STRL("onReceive"), zcallback TSRMLS_CC); cb->onReceive = sw_zend_read_property(swoole_client_class_entry_ptr, getThis(), ZEND_STRL("onReceive"), 0 TSRMLS_CC); sw_copy_to_stack(cb->onReceive, cb->_onReceive); #ifdef PHP_SWOOLE_ENABLE_FASTCALL cb->cache_onReceive = func_cache; #endif } else if (strncasecmp("close", cb_name, cb_name_len) == 0) { zend_update_property(swoole_client_class_entry_ptr, getThis(), ZEND_STRL("onClose"), zcallback TSRMLS_CC); cb->onClose = sw_zend_read_property(swoole_client_class_entry_ptr, getThis(), ZEND_STRL("onClose"), 0 TSRMLS_CC); sw_copy_to_stack(cb->onClose, cb->_onClose); #ifdef PHP_SWOOLE_ENABLE_FASTCALL cb->cache_onClose = func_cache; #endif } else if (strncasecmp("error", cb_name, cb_name_len) == 0) { zend_update_property(swoole_client_class_entry_ptr, getThis(), ZEND_STRL("onError"), zcallback TSRMLS_CC); cb->onError = sw_zend_read_property(swoole_client_class_entry_ptr, getThis(), ZEND_STRL("onError"), 0 TSRMLS_CC); sw_copy_to_stack(cb->onError, cb->_onError); #ifdef PHP_SWOOLE_ENABLE_FASTCALL cb->cache_onError = func_cache; #endif } else if (strncasecmp("bufferFull", cb_name, cb_name_len) == 0) { zend_update_property(swoole_client_class_entry_ptr, getThis(), ZEND_STRL("onBufferFull"), zcallback TSRMLS_CC); cb->onBufferFull = sw_zend_read_property(swoole_client_class_entry_ptr, getThis(), ZEND_STRL("onBufferFull"), 0 TSRMLS_CC); sw_copy_to_stack(cb->onBufferFull, cb->_onBufferFull); #ifdef PHP_SWOOLE_ENABLE_FASTCALL cb->cache_onBufferFull = func_cache; #endif } else if (strncasecmp("bufferEmpty", cb_name, cb_name_len) == 0) { zend_update_property(swoole_client_class_entry_ptr, getThis(), ZEND_STRL("onBufferEmpty"), zcallback TSRMLS_CC); cb->onBufferEmpty = sw_zend_read_property(swoole_client_class_entry_ptr, getThis(), ZEND_STRL("onBufferEmpty"), 0 TSRMLS_CC); sw_copy_to_stack(cb->onBufferEmpty, cb->_onBufferEmpty); #ifdef PHP_SWOOLE_ENABLE_FASTCALL cb->cache_onBufferEmpty = func_cache; #endif } else { swoole_php_fatal_error(E_WARNING, "Unknown event callback type name "%s".", cb_name); RETURN_FALSE; } RETURN_TRUE; }swoole_client->connect connect
connect 是客戶(hù)端模塊核心的函數(shù)。
PHP 內(nèi)部函數(shù)使用 zend_parse_parameters() API 接受參數(shù),將輸入?yún)?shù)轉(zhuǎn)換成 c 變量。不幸的是,每次調(diào)用這個(gè)函數(shù)時(shí)都要對(duì)這個(gè)這個(gè)字符串進(jìn)行解析,這會(huì)加重性能開(kāi)銷(xiāo)。在PHP7中新提供的方式。是為了提高參數(shù)解析的性能。對(duì)應(yīng)經(jīng)常使用的方法,建議使用 FAST ZPP 方式。
利用 php_swoole_client_new 函數(shù)創(chuàng)建一個(gè) swClient 客戶(hù)端對(duì)象
如果是異步的 TCP 客戶(hù)端,設(shè)置 sock_flag 為 1,為后面的異步 connect 的參數(shù)
如果客戶(hù)端當(dāng)前狀態(tài)是激活而且是保持長(zhǎng)連接的,直接返回成功,不需要再次連接。
根據(jù)配置參數(shù),利用函數(shù) php_swoole_client_check_setting 設(shè)置 swClient 對(duì)象的屬性
如果客戶(hù)端是異步的:
如果是 TCP 客戶(hù)端,需要驗(yàn)證是否設(shè)置了 onConnect、onError、onClose 三個(gè)回調(diào)函數(shù),如果沒(méi)有返回錯(cuò)誤。onBufferFull、onBufferEmpty 是可選回調(diào)函數(shù)。
如果是 UDP 客戶(hù)端,需要驗(yàn)證是否設(shè)置了 onReceive 等函數(shù),否則返回錯(cuò)誤。onConnect、onClose 兩個(gè)函數(shù)是可選回調(diào)函數(shù)。
值得注意的是,swClient 中的 onConnect 等函數(shù)并沒(méi)有直接使用用戶(hù)的回調(diào)函數(shù),而是使用 client_onConnect 等函數(shù),將用戶(hù)的回調(diào)函數(shù)放在了 cli->object 的屬性中。
最后調(diào)用 connect 函數(shù)與服務(wù)端進(jìn)行連接。
static PHP_METHOD(swoole_client, connect) { zend_long port = 0, sock_flag = 0; char *host = NULL; zend_size_t host_len; double timeout = SW_CLIENT_DEFAULT_TIMEOUT; #ifdef FAST_ZPP ZEND_PARSE_PARAMETERS_START(1, 4) Z_PARAM_STRING(host, host_len) Z_PARAM_OPTIONAL Z_PARAM_LONG(port) Z_PARAM_DOUBLE(timeout) Z_PARAM_LONG(sock_flag) ZEND_PARSE_PARAMETERS_END(); #else if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "s|ldl", &host, &host_len, &port, &timeout, &sock_flag) == FAILURE) { return; } #endif swClient *cli = (swClient *) swoole_get_object(getThis()); cli = php_swoole_client_new(getThis(), host, host_len, port); swoole_set_object(getThis(), cli); if (cli->type == SW_SOCK_TCP || cli->type == SW_SOCK_TCP6) { if (cli->async == 1) { //for tcp: nonblock //for udp: have udp connect sock_flag = 1; } } if (cli->keep == 1 && cli->socket->active == 1) { zend_update_property_bool(swoole_client_class_entry_ptr, getThis(), SW_STRL("reuse")-1, 1 TSRMLS_CC); RETURN_TRUE; } else if (cli->socket->active == 1) { swoole_php_fatal_error(E_WARNING, "connection to the server has already been established."); RETURN_FALSE; } zval *zset = sw_zend_read_property(swoole_client_class_entry_ptr, getThis(), ZEND_STRL("setting"), 1 TSRMLS_CC); if (zset && !ZVAL_IS_NULL(zset)) { php_swoole_client_check_setting(cli, zset TSRMLS_CC); } //nonblock async if (cli->async) { client_callback *cb = (client_callback *) swoole_get_property(getThis(), 0); if (!cb) { swoole_php_fatal_error(E_ERROR, "no event callback function."); RETURN_FALSE; } if (swSocket_is_stream(cli->type)) { if (!cb->onConnect) { swoole_php_fatal_error(E_ERROR, "no "onConnect" callback function."); RETURN_FALSE; } if (!cb->onError) { swoole_php_fatal_error(E_ERROR, "no "onError" callback function."); RETURN_FALSE; } if (!cb->onClose) { swoole_php_fatal_error(E_ERROR, "no "onClose" callback function."); RETURN_FALSE; } cli->onConnect = client_onConnect; cli->onClose = client_onClose; cli->onError = client_onError; cli->onReceive = client_onReceive; cli->reactor_fdtype = PHP_SWOOLE_FD_STREAM_CLIENT; if (cb->onBufferFull) { cli->onBufferFull = client_onBufferFull; } if (cb->onBufferEmpty) { cli->onBufferEmpty = client_onBufferEmpty; } } else { if (!cb || !cb->onReceive) { swoole_php_fatal_error(E_ERROR, "no "onReceive" callback function."); RETURN_FALSE; } if (cb->onConnect) { cli->onConnect = client_onConnect; } if (cb->onClose) { cli->onClose = client_onClose; } cli->onReceive = client_onReceive; cli->reactor_fdtype = PHP_SWOOLE_FD_DGRAM_CLIENT; } zval *zobject = getThis(); cli->object = zobject; sw_copy_to_stack(cli->object, cb->_object); sw_zval_add_ref(&zobject); } //nonblock async if (cli->connect(cli, host, port, timeout, sock_flag) < 0) { if (errno == 0 ) { if (SwooleG.error == SW_ERROR_DNSLOOKUP_RESOLVE_FAILED) { swoole_php_error(E_WARNING, "connect to server[%s:%d] failed. Error: %s[%d]", host, (int )port, hstrerror(h_errno), h_errno); } zend_update_property_long(swoole_client_class_entry_ptr, getThis(), SW_STRL("errCode")-1, SwooleG.error TSRMLS_CC); } else { swoole_php_sys_error(E_WARNING, "connect to server[%s:%d] failed.", host, (int )port); zend_update_property_long(swoole_client_class_entry_ptr, getThis(), SW_STRL("errCode")-1, errno TSRMLS_CC); } RETURN_FALSE; } RETURN_TRUE; }php_swoole_client_new
php_swoole_client_new 會(huì)新建一個(gè) swClient 客戶(hù)端對(duì)象,具體流程如下:
首先取出 type 屬性,觀察是否是異步客戶(hù)端 SW_FLAG_ASYNC
接著取出 id 屬性,觀察是否存在著 connection_id,如果沒(méi)有傳入 connection_id 就要根據(jù)主域與端口號(hào)來(lái)創(chuàng)建 connection_id
如果當(dāng)前客戶(hù)端要保持長(zhǎng)連接,要試圖從 php_sw_long_connections 中根據(jù) connection_id 取出客戶(hù)端對(duì)象,復(fù)用連接
如果 php_sw_long_connections 哈希表中并沒(méi)有 connection_id,那么新申請(qǐng)一個(gè) swClient 放入哈希表中,并調(diào)到 create_socket 創(chuàng)建客戶(hù)端對(duì)象。
如果取出了客戶(hù)端對(duì)象,那么要嘗試接受數(shù)據(jù),如果報(bào)錯(cuò),說(shuō)明連接已經(jīng)失效,需要關(guān)閉該連接,并調(diào)到 create_socket 創(chuàng)建一個(gè)新的客戶(hù)端對(duì)象;如果還可以使用連接,就遞增 reuse_count,
如果當(dāng)前客戶(hù)端不需要復(fù)用長(zhǎng)連接,那么就利用 swClient_create 創(chuàng)建一個(gè)新的客戶(hù)端
swClient* php_swoole_client_new(zval *object, char *host, int host_len, int port) { zval *ztype; int async = 0; char conn_key[SW_LONG_CONNECTION_KEY_LEN]; int conn_key_len = 0; uint64_t tmp_buf; int ret; ztype = sw_zend_read_property(Z_OBJCE_P(object), object, SW_STRL("type")-1, 0 TSRMLS_CC); long type = Z_LVAL_P(ztype); //new flag, swoole-1.6.12+ if (type & SW_FLAG_ASYNC) { async = 1; } swClient *cli; bzero(conn_key, SW_LONG_CONNECTION_KEY_LEN); zval *connection_id = sw_zend_read_property(Z_OBJCE_P(object), object, ZEND_STRL("id"), 1 TSRMLS_CC); if (connection_id == NULL || ZVAL_IS_NULL(connection_id)) { conn_key_len = snprintf(conn_key, SW_LONG_CONNECTION_KEY_LEN, "%s:%d", host, port) + 1; } else { conn_key_len = snprintf(conn_key, SW_LONG_CONNECTION_KEY_LEN, "%s", Z_STRVAL_P(connection_id)) + 1; } //keep the tcp connection if (type & SW_FLAG_KEEP) { swClient *find = swHashMap_find(php_sw_long_connections, conn_key, conn_key_len); if (find == NULL) { cli = (swClient*) pemalloc(sizeof(swClient), 1); if (swHashMap_add(php_sw_long_connections, conn_key, conn_key_len, cli) == FAILURE) { swoole_php_fatal_error(E_WARNING, "failed to add swoole_client_create_socket to hashtable."); } goto create_socket; } else { cli = find; //try recv, check connection status ret = recv(cli->socket->fd, &tmp_buf, sizeof(tmp_buf), MSG_DONTWAIT | MSG_PEEK); if (ret == 0 || (ret < 0 && swConnection_error(errno) == SW_CLOSE)) { cli->close(cli); goto create_socket; } cli->reuse_count ++; zend_update_property_long(Z_OBJCE_P(object), object, ZEND_STRL("reuseCount"), cli->reuse_count TSRMLS_CC); } } else { cli = (swClient*) emalloc(sizeof(swClient)); create_socket: if (swClient_create(cli, php_swoole_socktype(type), async) < 0) { swoole_php_fatal_error(E_WARNING, "swClient_create() failed. Error: %s [%d]", strerror(errno), errno); zend_update_property_long(Z_OBJCE_P(object), object, ZEND_STRL("errCode"), errno TSRMLS_CC); return NULL; } //don"t forget free it cli->server_str = sw_strndup(conn_key, conn_key_len); cli->server_strlen = conn_key_len; } zend_update_property_long(Z_OBJCE_P(object), object, ZEND_STRL("sock"), cli->socket->fd TSRMLS_CC); if (type & SW_FLAG_KEEP) { cli->keep = 1; } #ifdef SW_USE_OPENSSL if (type & SW_SOCK_SSL) { cli->open_ssl = 1; } #endif return cli; }swClient_create
創(chuàng)建客戶(hù)端就是利用 type 類(lèi)型來(lái)創(chuàng)建不同的 socket 套接字
如果是異步客戶(hù)端,設(shè)置客戶(hù)端的 reactor 對(duì)象(可以分為 reactor 線程和其他進(jìn)程 reactor)
設(shè)置 cli->socket 屬性
如果是異步客戶(hù)端,利用 cli->reactor->setHandle 設(shè)定 reactor 的回調(diào)函數(shù)
根據(jù)異步或者同步、數(shù)據(jù)流或者數(shù)據(jù)報(bào),設(shè)置 swClient 的各種函數(shù)
int swClient_create(swClient *cli, int type, int async) { int _domain; int _type; bzero(cli, sizeof(swClient)); switch (type) { case SW_SOCK_TCP: _domain = AF_INET; _type = SOCK_STREAM; break; case SW_SOCK_TCP6: _domain = AF_INET6; _type = SOCK_STREAM; break; case SW_SOCK_UNIX_STREAM: _domain = AF_UNIX; _type = SOCK_STREAM; break; case SW_SOCK_UDP: _domain = AF_INET; _type = SOCK_DGRAM; break; case SW_SOCK_UDP6: _domain = AF_INET6; _type = SOCK_DGRAM; break; case SW_SOCK_UNIX_DGRAM: _domain = AF_UNIX; _type = SOCK_DGRAM; break; default: return SW_ERR; } #ifdef SOCK_CLOEXEC int sockfd = socket(_domain, _type | SOCK_CLOEXEC, 0); #else int sockfd = socket(_domain, _type, 0); #endif if (sockfd < 0) { swWarn("socket() failed. Error: %s[%d]", strerror(errno), errno); return SW_ERR; } if (async) { if (swIsMaster() && SwooleTG.type == SW_THREAD_REACTOR) { cli->reactor = SwooleTG.reactor; } else { cli->reactor = SwooleG.main_reactor; } cli->socket = swReactor_get(cli->reactor, sockfd); } else { cli->socket = sw_malloc(sizeof(swConnection)); } cli->buffer_input_size = SW_CLIENT_BUFFER_SIZE; bzero(cli->socket, sizeof(swConnection)); cli->socket->fd = sockfd; cli->socket->object = cli; if (async) { swSetNonBlock(cli->socket->fd); if (!swReactor_handle_isset(cli->reactor, SW_FD_STREAM_CLIENT)) { cli->reactor->setHandle(cli->reactor, SW_FD_STREAM_CLIENT | SW_EVENT_READ, swClient_onStreamRead); cli->reactor->setHandle(cli->reactor, SW_FD_DGRAM_CLIENT | SW_EVENT_READ, swClient_onDgramRead); cli->reactor->setHandle(cli->reactor, SW_FD_STREAM_CLIENT | SW_EVENT_WRITE, swClient_onWrite); cli->reactor->setHandle(cli->reactor, SW_FD_STREAM_CLIENT | SW_EVENT_ERROR, swClient_onError); } } if (swSocket_is_stream(type)) { cli->recv = swClient_tcp_recv_no_buffer; if (async) { cli->connect = swClient_tcp_connect_async; cli->send = swClient_tcp_send_async; cli->sendfile = swClient_tcp_sendfile_async; cli->pipe = swClient_tcp_pipe; cli->socket->dontwait = 1; } else { cli->connect = swClient_tcp_connect_sync; cli->send = swClient_tcp_send_sync; cli->sendfile = swClient_tcp_sendfile_sync; } cli->reactor_fdtype = SW_FD_STREAM_CLIENT; } else { cli->connect = swClient_udp_connect; cli->recv = swClient_udp_recv; cli->send = swClient_udp_send; cli->reactor_fdtype = SW_FD_DGRAM_CLIENT; } cli->_sock_domain = _domain; cli->_sock_type = _type; cli->close = swClient_close; cli->type = type; cli->async = async; cli->protocol.package_length_type = "N"; cli->protocol.package_length_size = 4; cli->protocol.package_body_offset = 0; cli->protocol.package_max_length = SW_BUFFER_INPUT_SIZE; cli->protocol.onPackage = swClient_onPackage; return SW_OK; }swClient_tcp_connect_async 異步客戶(hù)端數(shù)據(jù)流連接
對(duì)于異步的數(shù)據(jù)流連接來(lái)說(shuō),首先要驗(yàn)證 onConnect、onError、onClose 回調(diào)函數(shù)不能少。
swClient_inet_addr 用于為 cli->server_addr.addr 賦值,主要是要利用 htons、inet_pton 轉(zhuǎn)化數(shù)值。
對(duì)于異步的客戶(hù)端來(lái)說(shuō) cli->wait_dns 是 1,需要 AIO 模塊來(lái)異步加載 DNS,進(jìn)行 swAio_dispatch 之后本函數(shù)就會(huì)立刻返回 true
當(dāng) AIO 模塊解析了 DNS 之后,cli->wait_dns 會(huì)被重置為 0,再次調(diào)用本函數(shù)swClient_tcp_connect_async
調(diào)用 connect 函數(shù)進(jìn)行建立連接,遇到 EINTR 信號(hào)中斷要進(jìn)行重試
當(dāng)錯(cuò)誤是 EINPROGRESS 的時(shí)候,將套接字放入 reactor 中,并設(shè)置超時(shí)時(shí)間。當(dāng)我們以非阻塞的方式來(lái)進(jìn)行連接的時(shí)候,返回的結(jié)果如果是 -1, 這并不代表這次連接發(fā)生了錯(cuò)誤,如果它的返回結(jié)果是 EINPROGRESS,那么就代表連接還在進(jìn)行中。 后面可以將套接字放入 reactor 中,如果可以寫(xiě),說(shuō)明連接完成了。
static int swClient_tcp_connect_async(swClient *cli, char *host, int port, double timeout, int nonblock) { int ret; cli->timeout = timeout; if (!cli->buffer) { //alloc input memory buffer cli->buffer = swString_new(cli->buffer_input_size); if (!cli->buffer) { return SW_ERR; } } if (!(cli->onConnect && cli->onError && cli->onClose)) { swWarn("onConnect/onError/onClose callback have not set."); return SW_ERR; } if (cli->onBufferFull && cli->buffer_high_watermark == 0) { cli->buffer_high_watermark = cli->socket->buffer_size * 0.8; } if (swClient_inet_addr(cli, host, port) < 0) { return SW_ERR; } if (cli->wait_dns) { if (SwooleAIO.init == 0) { swAio_init(); } swAio_event ev; bzero(&ev, sizeof(swAio_event)); int len = strlen(cli->server_host); if (strlen(cli->server_host) < SW_IP_MAX_LENGTH) { ev.nbytes = SW_IP_MAX_LENGTH; } else { ev.nbytes = len + 1; } ev.buf = sw_malloc(ev.nbytes); if (!ev.buf) { swWarn("malloc failed."); return SW_ERR; } memcpy(ev.buf, cli->server_host, len); ((char *) ev.buf)[len] = 0; ev.flags = cli->_sock_domain; ev.type = SW_AIO_GETHOSTBYNAME; ev.object = cli; ev.fd = cli->socket->fd; ev.callback = swClient_onResolveCompleted; if (swAio_dispatch(&ev) < 0) { sw_free(ev.buf); return SW_ERR; } else { return SW_OK; } } while (1) { ret = connect(cli->socket->fd, (struct sockaddr *) &cli->server_addr.addr, cli->server_addr.len); if (ret < 0) { if (errno == EINTR) { continue; } SwooleG.error = errno; } break; } if ((ret < 0 && errno == EINPROGRESS) || ret == 0) { if (cli->reactor->add(cli->reactor, cli->socket->fd, cli->reactor_fdtype | SW_EVENT_WRITE) < 0) { return SW_ERR; } if (timeout > 0) { if (SwooleG.timer.fd == 0) { swTimer_init((int) (timeout * 1000)); } cli->timer = SwooleG.timer.add(&SwooleG.timer, (int) (timeout * 1000), 0, cli, swClient_onTimeout); } return SW_OK; } return ret; }swClient_inet_addr 轉(zhuǎn)化地址
為了解決大端小端問(wèn)題,就不能直接在 connect 函數(shù)參數(shù)中傳數(shù)組,而是應(yīng)該利用 htons、inet_pton 等函數(shù)進(jìn)行轉(zhuǎn)化。
static int swClient_inet_addr(swClient *cli, char *host, int port) { ... cli->server_host = host; cli->server_port = port; void *addr = NULL; if (cli->type == SW_SOCK_TCP || cli->type == SW_SOCK_UDP) { cli->server_addr.addr.inet_v4.sin_family = AF_INET; cli->server_addr.addr.inet_v4.sin_port = htons(port); cli->server_addr.len = sizeof(cli->server_addr.addr.inet_v4); addr = &cli->server_addr.addr.inet_v4.sin_addr.s_addr; if (inet_pton(AF_INET, host, addr)) { return SW_OK; } } else if (cli->type == SW_SOCK_TCP6 || cli->type == SW_SOCK_UDP6) { cli->server_addr.addr.inet_v6.sin6_family = AF_INET6; cli->server_addr.addr.inet_v6.sin6_port = htons(port); cli->server_addr.len = sizeof(cli->server_addr.addr.inet_v6); addr = cli->server_addr.addr.inet_v6.sin6_addr.s6_addr; if (inet_pton(AF_INET6, host, addr)) { return SW_OK; } } else if (cli->type == SW_SOCK_UNIX_STREAM || cli->type == SW_SOCK_UNIX_DGRAM) { cli->server_addr.addr.un.sun_family = AF_UNIX; strncpy(cli->server_addr.addr.un.sun_path, host, sizeof(cli->server_addr.addr.un.sun_path) - 1); cli->server_addr.addr.un.sun_path[sizeof(cli->server_addr.addr.un.sun_path) - 1] = 0; cli->server_addr.len = sizeof(cli->server_addr.addr.un.sun_path); return SW_OK; } else { return SW_ERR; } if (!cli->async) { ... } else { cli->wait_dns = 1; } return SW_OK; }AIO 異步 DNS 解析
所謂的 AIO 異步并不是操作系統(tǒng)中真正的異步系統(tǒng)調(diào)用,而是 swoole 利用線程池 + reactor 實(shí)現(xiàn)的異步任務(wù)系統(tǒng),當(dāng)線程完成任務(wù)后,就會(huì)執(zhí)行相應(yīng)的 callback 函數(shù),這里 DNS 異步解析事件的回調(diào)函數(shù)就是 swClient_onResolveCompleted:
static void swClient_onResolveCompleted(swAio_event *event) { swConnection *socket = swReactor_get(SwooleG.main_reactor, event->fd); if (socket->removed) { sw_free(event->buf); return; } swClient *cli = event->object; cli->wait_dns = 0; if (event->error == 0) { swClient_tcp_connect_async(cli, event->buf, cli->server_port, cli->timeout, 1); } else { SwooleG.error = SW_ERROR_DNSLOOKUP_RESOLVE_FAILED; cli->socket->removed = 1; cli->close(cli); if (cli->onError) { cli->onError(cli); } } sw_free(event->buf); }DNS 解析
DNS 具體的解析函數(shù)是 swoole_gethostbyname:
獲取 DNS 的方法有兩種:gethostbyname2_r 與 gethostbyname2,gethostbyname2_r 是線程安全函數(shù),可以用于多線程。
gethostbyname2_r 中的第四個(gè)參數(shù) buf 用于存放臨時(shí)數(shù)據(jù)
主機(jī)的 ip 地址可能會(huì)有多個(gè),函數(shù)只把第一個(gè) ip 地址復(fù)制到 addr 中。
struct hostent { char *h_name; //主機(jī)的規(guī)范名 char **h_aliases; //主機(jī)的別名 int h_addrtype;//主機(jī)ip地址的類(lèi)型,到底是(AF_INET),還是(AF_INET6) int h_length;//主機(jī)ip地址的長(zhǎng)度 char **h_addr_list;//主機(jī)的ip地址 }; #ifdef HAVE_GETHOSTBYNAME2_R int swoole_gethostbyname(int flags, char *name, char *addr) { int __af = flags & (~SW_DNS_LOOKUP_RANDOM); int index = 0; int rc, err; int buf_len = 256; struct hostent hbuf; struct hostent *result; char *buf = (char*) sw_malloc(buf_len); memset(buf, 0, buf_len); while ((rc = gethostbyname2_r(name, __af, &hbuf, buf, buf_len, &result, &err)) == ERANGE) { buf_len *= 2; void *tmp = sw_realloc(buf, buf_len); if (NULL == tmp) { sw_free(buf); return SW_ERR; } else { buf = tmp; } } if (0 != rc || NULL == result) { sw_free(buf); return SW_ERR; } union { char v4[INET_ADDRSTRLEN]; char v6[INET6_ADDRSTRLEN]; } addr_list[SW_DNS_HOST_BUFFER_SIZE]; int i = 0; for (i = 0; i < SW_DNS_HOST_BUFFER_SIZE; i++) { if (hbuf.h_addr_list[i] == NULL) { break; } if (__af == AF_INET) { memcpy(addr_list[i].v4, hbuf.h_addr_list[i], hbuf.h_length); } else { memcpy(addr_list[i].v6, hbuf.h_addr_list[i], hbuf.h_length); } } if (__af == AF_INET) { memcpy(addr, addr_list[index].v4, hbuf.h_length); } else { memcpy(addr, addr_list[index].v6, hbuf.h_length); } sw_free(buf); return SW_OK; } #else int swoole_gethostbyname(int flags, char *name, char *addr) { int __af = flags & (~SW_DNS_LOOKUP_RANDOM); int index = 0; struct hostent *host_entry; if (!(host_entry = gethostbyname2(name, __af))) { return SW_ERR; } union { char v4[INET_ADDRSTRLEN]; char v6[INET6_ADDRSTRLEN]; } addr_list[SW_DNS_HOST_BUFFER_SIZE]; int i = 0; for (i = 0; i < SW_DNS_HOST_BUFFER_SIZE; i++) { if (host_entry->h_addr_list[i] == NULL) { break; } if (__af == AF_INET) { memcpy(addr_list[i].v4, host_entry->h_addr_list[i], host_entry->h_length); } else { memcpy(addr_list[i].v6, host_entry->h_addr_list[i], host_entry->h_length); } } if (__af == AF_INET) { memcpy(addr, addr_list[index].v4, host_entry->h_length); } else { memcpy(addr, addr_list[index].v6, host_entry->h_length); } return SW_OK; } #endifswClient_tcp_connect_sync 同步數(shù)據(jù)流連接
與異步的 TCP 連接類(lèi)似,函數(shù)首先要調(diào)用 swClient_inet_addr 來(lái)轉(zhuǎn)化主域與端口號(hào),如果 inet_pton 函數(shù)返回成功,說(shuō)明并不需要進(jìn)行 DNS 解析,直接返回;否則就要同步調(diào)用 swoole_gethostbyname 函數(shù)進(jìn)行 DNS 的解析。
利用 swSocket_set_timeout 函數(shù)為套接字設(shè)置超時(shí)時(shí)間。
如果使用 openssl 要進(jìn)行 SSL 握手。
static int swClient_inet_addr(swClient *cli, char *host, int port) { ... if (!cli->async) { if (swoole_gethostbyname(cli->_sock_domain, host, addr) < 0) { SwooleG.error = SW_ERROR_DNSLOOKUP_RESOLVE_FAILED; return SW_ERR; } } ... } int swSocket_set_timeout(int sock, double timeout) { int ret; struct timeval timeo; timeo.tv_sec = (int) timeout; timeo.tv_usec = (int) ((timeout - timeo.tv_sec) * 1000 * 1000); ret = setsockopt(sock, SOL_SOCKET, SO_SNDTIMEO, (void *) &timeo, sizeof(timeo)); if (ret < 0) { swWarn("setsockopt(SO_SNDTIMEO) failed. Error: %s[%d]", strerror(errno), errno); return SW_ERR; } ret = setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, (void *) &timeo, sizeof(timeo)); if (ret < 0) { swWarn("setsockopt(SO_RCVTIMEO) failed. Error: %s[%d]", strerror(errno), errno); return SW_ERR; } return SW_OK; } static int swClient_tcp_connect_sync(swClient *cli, char *host, int port, double timeout, int nonblock) { int ret, n; char buf[1024]; cli->timeout = timeout; if (swClient_inet_addr(cli, host, port) < 0) { return SW_ERR; } if (nonblock == 1) { swSetNonBlock(cli->socket->fd); } else { if (cli->timeout > 0) { swSocket_set_timeout(cli->socket->fd, timeout); } } while (1) { ret = connect(cli->socket->fd, (struct sockaddr *) &cli->server_addr.addr, cli->server_addr.len); #endif if (ret < 0) { if (errno == EINTR) { continue; } } break; } if (ret >= 0) { cli->socket->active = 1; #ifdef SW_USE_OPENSSL if (cli->open_ssl) { if (swClient_enable_ssl_encrypt(cli) < 0) { return SW_ERR; } if (swClient_ssl_handshake(cli) < 0) { return SW_ERR; } } #endif } return ret; }SSL 客戶(hù)端加密
其過(guò)程與服務(wù)端 SSL 隧道加密流程相似,首先需要?jiǎng)?chuàng)建上下文 swSSL_get_context;如果要驗(yàn)證對(duì)端證書(shū),需要調(diào)用 swSSL_set_capath 設(shè)置證書(shū)地址;如果使用 http2 協(xié)商,要使用 SSL_CTX_set_alpn_protos 函數(shù)設(shè)置 h2
創(chuàng)建上下文之后,就可以利用 swSSL_create 創(chuàng)建 SSL 對(duì)象綁定套接字,并利用 SSL_set_tlsext_host_name 設(shè)置 SSL 主域,SSL_set_tlsext_host_name(s,name) 函數(shù)來(lái)設(shè)置ClientHello 中的 Server Name
SNI 是 Server Name Indication 的縮寫(xiě),是為了解決一個(gè)服務(wù)器使用多個(gè)域名和證書(shū)的 SSL/TLS 擴(kuò)展。它允許客戶(hù)端在發(fā)起 SSL 握手請(qǐng)求時(shí)(客戶(hù)端發(fā)出 ClientHello 消息中)提交請(qǐng)求的 HostName 信息,使得服務(wù)器能夠切換到正確的域并返回相應(yīng)的證書(shū)。在 SNI 出現(xiàn)之前,HostName 信息只存在于 HTTP 請(qǐng)求中,但 SSL/TLS 層無(wú)法獲知這一信息。通過(guò)將HostName 的信息加入到 SNI 擴(kuò)展中,SSL/TLS 允許服務(wù)器使用一個(gè) IP 為不同的域名提供不同的證書(shū),從而能夠與使用同一個(gè) IP 的多個(gè)“虛擬主機(jī)”更方便地建立安全連接。
swSSL_connect 函數(shù)進(jìn)行 SSL 握手連接
如果要驗(yàn)證服務(wù)端證書(shū),則調(diào)用 swClient_ssl_verify 函數(shù)。
int swClient_enable_ssl_encrypt(swClient *cli) { cli->ssl_context = swSSL_get_context(&cli->ssl_option); if (cli->ssl_context == NULL) { return SW_ERR; } if (cli->ssl_option.verify_peer) { if (swSSL_set_capath(&cli->ssl_option, cli->ssl_context) < 0) { return SW_ERR; } } cli->socket->ssl_send = 1; #if defined(SW_USE_HTTP2) && defined(SW_USE_OPENSSL) && OPENSSL_VERSION_NUMBER >= 0x10002000L if (cli->http2) { if (SSL_CTX_set_alpn_protos(cli->ssl_context, (const unsigned char *) "x02h2", 3) < 0) { return SW_ERR; } } #endif return SW_OK; } int swClient_ssl_handshake(swClient *cli) { if (!cli->socket->ssl) { if (swSSL_create(cli->socket, cli->ssl_context, SW_SSL_CLIENT) < 0) { return SW_ERR; } #ifdef SSL_CTRL_SET_TLSEXT_HOSTNAME if (cli->ssl_option.tls_host_name) { SSL_set_tlsext_host_name(cli->socket->ssl, cli->ssl_option.tls_host_name); } #endif } if (swSSL_connect(cli->socket) < 0) { return SW_ERR; } if (cli->socket->ssl_state == SW_SSL_STATE_READY && cli->ssl_option.verify_peer) { if (swClient_ssl_verify(cli, cli->ssl_option.allow_self_signed) < 0) { return SW_ERR; } } return SW_OK; }
swSSL_set_capath 提供可信任證書(shū)庫(kù):
// 在創(chuàng)建上下文結(jié)構(gòu)之后,必須加載一個(gè)可信任證書(shū)庫(kù)。這是成功驗(yàn)證每個(gè)證書(shū)所必需的。 // 如果不能確認(rèn)證書(shū)是可信任的,那么 OpenSSL 會(huì)將證書(shū)標(biāo)記為無(wú)效(但連接仍可以繼續(xù))。 // OpenSSL 附帶了一組可信任證書(shū)。它們位于源文件目錄樹(shù)的 certs/demo 目錄中。 // 不過(guò),每個(gè)證書(shū)都是一個(gè)獨(dú)立的文件,也就是說(shuō),需要多帶帶加載每一個(gè)證書(shū)。 // 在 certs 目錄下,還有一個(gè) expired 存放過(guò)期證書(shū)的子目錄,試圖加載這些證書(shū)將會(huì)出錯(cuò)。 // 在數(shù)字證書(shū)進(jìn)行信任驗(yàn)證之前, // 必須為在為安全連接設(shè)置時(shí)創(chuàng)建的 OpenSSL SSL_CTX 對(duì)象提供一個(gè)默認(rèn)的信任證書(shū), // 這可以使用幾種方法來(lái)提供, // 但是最簡(jiǎn)單的方法是將這個(gè)證書(shū)保存為一個(gè) PEM 文件, // 并使用 // SSL_CTX_load_verify_locations( ctx, file, path ); // 將其加載到 OpenSSL 中。 // 該函數(shù)有三個(gè)參數(shù): // ctx - 上下文指針( SSL_CTX_new 函數(shù)返回 ); // file - 包含一個(gè)或多個(gè) PEM 格式的證書(shū)的文件的路徑(必需); // path - 到一個(gè)或多個(gè) PEM 格式文件的路徑,不過(guò)文件名必須使用特定的格式(可為 NULL); // 如果指定成功,則返回 1 ,如果遇到問(wèn)題,則返回 0 。 // 盡管當(dāng)信任證書(shū)在一個(gè)目錄中有多個(gè)多帶帶的文件時(shí)更容易添加或更新, // 但是您不太可能會(huì)如此頻繁地更新信任證書(shū),因此不必?fù)?dān)心這個(gè)問(wèn)題。 int swSSL_set_capath(swSSL_option *cfg, SSL_CTX *ctx) { if (cfg->cafile || cfg->capath) { if (!SSL_CTX_load_verify_locations(ctx, cfg->cafile, cfg->capath)) { return SW_ERR; } } else { if (!SSL_CTX_set_default_verify_paths(ctx)) { swWarn("Unable to set default verify locations and no CA settings specified."); return SW_ERR; } } if (cfg->verify_depth > 0) { SSL_CTX_set_verify_depth(ctx, cfg->verify_depth); } return SW_OK; }
swSSL_connect 函數(shù)就是簡(jiǎn)單調(diào)用 SSL_connect 來(lái)連接服務(wù)端:
int swSSL_connect(swConnection *conn) { int n = SSL_connect(conn->ssl); if (n == 1) { conn->ssl_state = SW_SSL_STATE_READY; conn->ssl_want_read = 0; conn->ssl_want_write = 0; #ifdef SW_LOG_TRACE_OPEN const char *ssl_version = SSL_get_version(conn->ssl); const char *ssl_cipher = SSL_get_cipher_name(conn->ssl); swTraceLog(SW_TRACE_SSL, "connected (%s %s)", ssl_version, ssl_cipher); #endif return SW_OK; } long err = SSL_get_error(conn->ssl, n); if (err == SSL_ERROR_WANT_READ) { conn->ssl_want_read = 1; conn->ssl_want_write = 0; conn->ssl_state = SW_SSL_STATE_WAIT_STREAM; return SW_OK; } else if (err == SSL_ERROR_WANT_WRITE) { conn->ssl_want_read = 0; conn->ssl_want_write = 1; conn->ssl_state = SW_SSL_STATE_WAIT_STREAM; return SW_OK; } else if (err == SSL_ERROR_ZERO_RETURN) { swDebug("SSL_connect(fd=%d) closed.", conn->fd); return SW_ERR; } else if (err == SSL_ERROR_SYSCALL) { if (n) { SwooleG.error = errno; return SW_ERR; } } swWarn("SSL_connect(fd=%d) failed. Error: %s[%ld|%d].", conn->fd, ERR_reason_error_string(err), err, errno); return SW_ERR; }
swClient_ssl_verify 驗(yàn)證連接的有效性:
// 連接建立后,必須檢查證書(shū),以確定它是否有效。 // 實(shí)際上,OpenSSL 為我們完成了這項(xiàng)任務(wù)。 // 如果證書(shū)有致命的問(wèn)題(例如,哈希值無(wú)效),那么將無(wú)法建立連接。 // 但是,如果證書(shū)的問(wèn)題并不是致命的(當(dāng)它已經(jīng)過(guò)期或者尚不合法時(shí)),那么仍可以繼續(xù)使用連接。 // 可以將 SSL 結(jié)構(gòu)作為惟一參數(shù), // 調(diào)用 SSL_get_verify_result 來(lái)查明證書(shū)是否通過(guò)了 OpenSSL 的檢驗(yàn)。 // 如果證書(shū)通過(guò)了包括信任檢查在內(nèi)的 OpenSSL 的內(nèi)部檢查,則返回 X509_V_OK。 // 如果有地方出了問(wèn)題,則返回一個(gè)錯(cuò)誤代碼,在 OpenSSL 文檔的 verify 部分中都進(jìn)行了介紹。 // 注:該錯(cuò)誤代碼被記錄在命令行工具的 verify 選項(xiàng)下。 // 應(yīng)該注意的是,驗(yàn)證失敗并不意味著連接不能使用。 // 是否應(yīng)該使用連接取決于驗(yàn)證結(jié)果和安全方面的考慮。 // 例如,失敗的信任驗(yàn)證可能只是意味著沒(méi)有可信任的證書(shū)。 // 連接仍然可用,只是需要從思想上提高安全意識(shí)。 // OpenSSL 在對(duì)證書(shū)進(jìn)行驗(yàn)證時(shí),有一些安全性檢查并沒(méi)有執(zhí)行, // 包括證書(shū)的失效檢查和對(duì)證書(shū)中通用名的有效性驗(yàn)證。 int swClient_ssl_verify(swClient *cli, int allow_self_signed) { if (swSSL_verify(cli->socket, allow_self_signed) < 0) { return SW_ERR; } if (cli->ssl_option.tls_host_name && swSSL_check_host(cli->socket, cli->ssl_option.tls_host_name) < 0) { return SW_ERR; } return SW_OK; } int swSSL_verify(swConnection *conn, int allow_self_signed) { int err = SSL_get_verify_result(conn->ssl); switch (err) { case X509_V_OK: return SW_OK; case X509_V_ERR_DEPTH_ZERO_SELF_SIGNED_CERT: if (allow_self_signed) { return SW_OK; } else { return SW_ERR; } default: swoole_error_log(SW_LOG_NOTICE, SW_ERROR_SSL_VEFIRY_FAILED, "Could not verify peer: code:%d %s", err, X509_verify_cert_error_string(err)); return SW_ERR; } return SW_ERR; }
驗(yàn)證了證書(shū)的有效性之后,還要驗(yàn)證域名的統(tǒng)一。高版本可以直接使用 X509_check_host 函數(shù)驗(yàn)證。
int swSSL_check_host(swConnection *conn, char *tls_host_name) { X509 *cert = SSL_get_peer_certificate(conn->ssl); if (cert == NULL) { return SW_ERR; } #ifdef X509_CHECK_FLAG_ALWAYS_CHECK_SUBJECT /* X509_check_host() is only available in OpenSSL 1.0.2+ */ if (X509_check_host(cert, tls_host_name, strlen(tls_host_name), 0, NULL) != 1) { swWarn("X509_check_host(): no match"); goto failed; } goto found; #else ... failed: X509_free(cert); return SW_ERR; found: X509_free(cert); return SW_OK; }swClient_udp_connect 數(shù)據(jù)報(bào)連接
對(duì)于 UDP 來(lái)說(shuō),swoole 并不會(huì) connect 服務(wù)端進(jìn)行三次握手,只是會(huì)調(diào)用 bind 來(lái)綁定本機(jī)的一個(gè)隨機(jī)端口,這個(gè)時(shí)候客戶(hù)端可以接受任何服務(wù)端發(fā)來(lái)的消息。
如果調(diào)用 connect 函數(shù)的時(shí)候指定了第 5 個(gè)參數(shù)為 true,那么就會(huì)調(diào)用 connect 函數(shù),這時(shí)候雖然客戶(hù)端仍然不會(huì)進(jìn)行三次握手,但是卻只能接受特定服務(wù)端發(fā)來(lái)的消息
調(diào)用了 connect 之后的 UDP 特點(diǎn):
不需要給輸出操作指定目的IP和目的端口,寫(xiě)到UDP的緩沖區(qū)里的數(shù)據(jù),將自動(dòng)發(fā)送到你調(diào)用connect指定的IP和端口。
在一個(gè)已連接的UDP套接字上,內(nèi)核由輸入操作返回的數(shù)據(jù)報(bào)只有那些來(lái)自connect所指定的協(xié)議地址的數(shù)據(jù)報(bào)。目的地為這個(gè)已連接的UDP套接字的本地協(xié)議地址(IP和端口),遠(yuǎn)端地址不是該套接字早先connect到的協(xié)議地址的數(shù)據(jù)報(bào),不會(huì)投遞到該套接字。這樣就限制了已連接的UDP套接字能且只能與一個(gè)對(duì)端交換數(shù)據(jù)報(bào)。
由已連接的套接字引發(fā)的異步錯(cuò)誤發(fā)回給他們所在的進(jìn)程,而未連接的UDP套接字不接受任何異步錯(cuò)誤。
讀寫(xiě)的操作接口方法增多了,除了可以使用sendto和recvfrom的接口外,還可以使用tcp的那套操作接口--read/readv/readmsg和write/writev等
UDP 客戶(hù)端可選設(shè)置 onConnect,socket 創(chuàng)建成功會(huì)立即回調(diào) onConnect
static int swClient_udp_connect(swClient *cli, char *host, int port, double timeout, int udp_connect) { if (swClient_inet_addr(cli, host, port) < 0) { return SW_ERR; } cli->socket->active = 1; cli->timeout = timeout; int bufsize = SwooleG.socket_buffer_size; if (timeout > 0) { swSocket_set_timeout(cli->socket->fd, timeout); } if (cli->type == SW_SOCK_UNIX_DGRAM) { struct sockaddr_un* client_addr = &cli->socket->info.addr.un; sprintf(client_addr->sun_path, "/tmp/swoole-client.%d.%d.sock", getpid(), cli->socket->fd); client_addr->sun_family = AF_UNIX; unlink(client_addr->sun_path); if (bind(cli->socket->fd, (struct sockaddr *) client_addr, sizeof(cli->socket->info.addr.un)) < 0) { swSysError("bind(%s) failed.", client_addr->sun_path); return SW_ERR; } } if (udp_connect != 1) { goto connect_ok; } if (connect(cli->socket->fd, (struct sockaddr *) (&cli->server_addr), cli->server_addr.len) == 0) { swSocket_clean(cli->socket->fd); connect_ok: setsockopt(cli->socket->fd, SOL_SOCKET, SO_SNDBUF, &bufsize, sizeof(bufsize)); setsockopt(cli->socket->fd, SOL_SOCKET, SO_RCVBUF, &bufsize, sizeof(bufsize)); if (cli->async && cli->onConnect) { if (cli->reactor->add(cli->reactor, cli->socket->fd, cli->reactor_fdtype | SW_EVENT_READ) < 0) { return SW_ERR; } execute_onConnect(cli); } return SW_OK; } else { swSysError("connect() failed."); cli->socket->active = 0; cli->socket->removed = 1; return SW_ERR; } }swoole_client->isConnected 異步連接判定
static PHP_METHOD(swoole_client, isConnected) { swClient *cli = (swClient *) swoole_get_object(getThis()); if (!cli) { RETURN_FALSE; } if (!cli->socket) { RETURN_FALSE; } RETURN_BOOL(cli->socket->active); }
文章版權(quán)歸作者所有,未經(jīng)允許請(qǐng)勿轉(zhuǎn)載,若此文章存在違規(guī)行為,您可以聯(lián)系管理員刪除。
轉(zhuǎn)載請(qǐng)注明本文地址:http://specialneedsforspecialkids.com/yun/29461.html
摘要:當(dāng)此時(shí)的套接字不可寫(xiě)的時(shí)候,會(huì)自動(dòng)放入緩沖區(qū)中。當(dāng)大于高水線時(shí),會(huì)自動(dòng)調(diào)用回調(diào)函數(shù)。寫(xiě)就緒狀態(tài)當(dāng)監(jiān)控到套接字進(jìn)入了寫(xiě)就緒狀態(tài)時(shí),就會(huì)調(diào)用函數(shù)。如果為,說(shuō)明此時(shí)異步客戶(hù)端雖然建立了連接,但是還沒(méi)有調(diào)用回調(diào)函數(shù),因此這時(shí)要調(diào)用函數(shù)。 前言 上一章我們說(shuō)了客戶(hù)端的連接 connect,對(duì)于同步客戶(hù)端來(lái)說(shuō),連接已經(jīng)建立成功;但是對(duì)于異步客戶(hù)端來(lái)說(shuō),此時(shí)可能還在進(jìn)行 DNS 的解析,on...
摘要:新建可以看到,自動(dòng)采用包長(zhǎng)檢測(cè)的方法該函數(shù)主要功能是設(shè)置各種回調(diào)函數(shù)值得注意的是第三個(gè)參數(shù)代表是否異步。發(fā)送數(shù)據(jù)函數(shù)并不是直接發(fā)送數(shù)據(jù),而是將數(shù)據(jù)存儲(chǔ)在,等著寫(xiě)事件就緒之后調(diào)用發(fā)送數(shù)據(jù)。 swReactorThread_dispatch 發(fā)送數(shù)據(jù) reactor 線程會(huì)通過(guò) swReactorThread_dispatch 發(fā)送數(shù)據(jù),當(dāng)采用 stream 發(fā)送數(shù)據(jù)的時(shí)候,會(huì)調(diào)用 sw...
摘要:判斷客戶(hù)端是否配置了檢測(cè)或者長(zhǎng)度檢測(cè),如果配置了就調(diào)用接受完整的數(shù)據(jù)包,這兩天會(huì)調(diào)用,進(jìn)而調(diào)用函數(shù)。異步客戶(hù)端接受數(shù)據(jù)異步的客戶(hù)端接受數(shù)據(jù)調(diào)用的和同步的客戶(hù)端相同,都是調(diào)用函數(shù)。 recv 接受數(shù)據(jù) 客戶(hù)端接受數(shù)據(jù)需要指定緩存區(qū)最大長(zhǎng)度,就是下面的 buf_len,flags 用于指定是否設(shè)置 waitall 標(biāo)志,如果設(shè)定了 waitall 就必須設(shè)定準(zhǔn)確的 size,否則會(huì)一直等...
摘要:進(jìn)程可以使用函數(shù)向進(jìn)程投遞新的任務(wù)。當(dāng)前的進(jìn)程在調(diào)用回調(diào)函數(shù)時(shí)會(huì)將進(jìn)程狀態(tài)切換為忙碌,這時(shí)將不再接收新的,當(dāng)函數(shù)返回時(shí)會(huì)將進(jìn)程狀態(tài)切換為空閑然后繼續(xù)接收新的。當(dāng)進(jìn)程投遞的任務(wù)在中完成時(shí),進(jìn)程會(huì)通過(guò)方法將任務(wù)處理的結(jié)果發(fā)送給進(jìn)程。 swoole——從入門(mén)到放棄(一) 一、swoole的源碼包安裝 下載swoole源碼:git clone https://gitee.com/swoole...
摘要:對(duì)于服務(wù)端來(lái)說(shuō),緩存默認(rèn)是不能使用的,可以通過(guò)調(diào)用函數(shù)來(lái)進(jìn)行設(shè)置生效。在回調(diào)函數(shù)中,首先申請(qǐng)一個(gè)大數(shù)數(shù)據(jù)結(jié)構(gòu),然后將其設(shè)定為,該值表示公鑰指數(shù),然后利用函數(shù)生成秘鑰。此時(shí)需要調(diào)用函數(shù)將新的連接與綁定。 前言 上一篇文章我們講了 OpenSSL 的原理,接下來(lái),我們來(lái)說(shuō)說(shuō)如何利用 openssl 第三方庫(kù)進(jìn)行開(kāi)發(fā),來(lái)為 tcp 層進(jìn)行 SSL 隧道加密 OpenSSL 初始化 在 sw...
閱讀 2797·2023-04-25 23:08
閱讀 1583·2021-11-23 09:51
閱讀 1564·2021-10-27 14:18
閱讀 3115·2019-08-29 13:25
閱讀 2831·2019-08-29 13:14
閱讀 2895·2019-08-26 18:36
閱讀 2193·2019-08-26 12:11
閱讀 811·2019-08-26 11:29