資訊專欄INFORMATION COLUMN
摘要：判斷客戶端是否配置了檢測或者長度檢測，如果配置了就調用接受完整的數據包，這兩天會調用，進而調用函數。異步客戶端接受數據異步的客戶端接受數據調用的和同步的客戶端相同，都是調用函數。
recv 接受數據
客戶端接受數據需要指定緩存區最大長度，就是下面的 buf_len，flags 用于指定是否設置 waitall 標志，如果設定了 waitall 就必須設定準確的 size，否則會一直等待，直到接收的數據長度達到 size。
客戶端啟用了 EOF/Length 檢測后，無需設置 size 和 waitall 參數。擴展層會返回完整的數據包或者返回false。
開啟了 open_eof_check/open_length_check 選項之后，會自動進行包長檢測，過程和服務端類似，此處不需要多說。
static PHP_METHOD(swoole_client, recv)
{
zend_long buf_len = SW_PHP_CLIENT_BUFFER_SIZE;
zend_long flags = 0;
int ret;
char *buf = NULL;
if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "|ll", &buf_len, &flags) == FAILURE)
{
return;
}
//waitall
if (flags == 1)
{
flags = MSG_WAITALL;
}
swClient *cli = client_get_ptr(getThis() TSRMLS_CC);
swProtocol *protocol = &cli->protocol;
if (cli->open_eof_check)
{
if (cli->buffer == NULL)
{
cli->buffer = swString_new(SW_BUFFER_SIZE_BIG);
}
swString *buffer = cli->buffer;
int eof = -1;
if (buffer->length > 0)
{
goto find_eof;
}
while (1)
{
buf = buffer->str + buffer->length;
buf_len = buffer->size - buffer->length;
if (buf_len > SW_BUFFER_SIZE_BIG)
{
buf_len = SW_BUFFER_SIZE_BIG;
}
ret = cli->recv(cli, buf, buf_len, 0);
if (ret < 0)
{
swoole_php_error(E_WARNING, "recv() failed. Error: %s [%d]", strerror(errno), errno);
buffer->length = 0;
RETURN_FALSE;
}
else if (ret == 0)
{
buffer->length = 0;
RETURN_EMPTY_STRING();
}
buffer->length += ret;
if (buffer->length < protocol->package_eof_len)
{
continue;
}
find_eof: eof = swoole_strnpos(buffer->str, buffer->length, protocol->package_eof, protocol->package_eof_len);
if (eof >= 0)
{
eof += protocol->package_eof_len;
SW_RETVAL_STRINGL(buffer->str, eof, 1);
if (buffer->length > eof)
{
buffer->length -= eof;
memmove(buffer->str, buffer->str + eof, buffer->length);
}
else
{
buffer->length = 0;
}
return;
}
else
{
if (buffer->length == protocol->package_max_length)
{
swoole_php_error(E_WARNING, "no package eof");
buffer->length = 0;
RETURN_FALSE;
}
else if (buffer->length == buffer->size)
{
if (buffer->size < protocol->package_max_length)
{
int new_size = buffer->size * 2;
if (new_size > protocol->package_max_length)
{
new_size = protocol->package_max_length;
}
if (swString_extend(buffer, new_size) < 0)
{
buffer->length = 0;
RETURN_FALSE;
}
}
}
}
}
buffer->length = 0;
RETURN_FALSE;
}
else if (cli->open_length_check)
{
if (cli->buffer == NULL)
{
cli->buffer = swString_new(SW_BUFFER_SIZE_STD);
}
uint32_t header_len = protocol->package_length_offset + protocol->package_length_size;
ret = cli->recv(cli, cli->buffer->str, header_len, MSG_WAITALL);
if (ret <= 0)
{
goto check_return;
}
else if (ret != header_len)
{
ret = 0;
goto check_return;
}
buf_len = protocol->get_package_length(protocol, cli->socket, cli->buffer->str, ret);
//error package
if (buf_len < 0)
{
RETURN_EMPTY_STRING();
}
//empty package
else if (buf_len == header_len)
{
SW_RETURN_STRINGL(cli->buffer->str, header_len, 1);
}
else if (buf_len > protocol->package_max_length)
{
swoole_error_log(SW_LOG_WARNING, SW_ERROR_PACKAGE_LENGTH_TOO_LARGE, "Package is too big. package_length=%d", (int )buf_len);
RETURN_EMPTY_STRING();
}
buf = (char *) emalloc(buf_len + 1);
memcpy(buf, cli->buffer->str, header_len);
SwooleG.error = 0;
ret = cli->recv(cli, buf + header_len, buf_len - header_len, MSG_WAITALL);
if (ret > 0)
{
ret += header_len;
if (ret != buf_len)
{
ret = 0;
}
}
}
else
{
if (!(flags & MSG_WAITALL) && buf_len > SW_PHP_CLIENT_BUFFER_SIZE)
{
buf_len = SW_PHP_CLIENT_BUFFER_SIZE;
}
buf = (char *) emalloc(buf_len + 1);
SwooleG.error = 0;
ret = cli->recv(cli, buf, buf_len, flags);
}
check_return:
if (ret < 0)
{
SwooleG.error = errno;
swoole_php_error(E_WARNING, "recv() failed. Error: %s [%d]", strerror(SwooleG.error), SwooleG.error);
zend_update_property_long(swoole_client_class_entry_ptr, getThis(), SW_STRL("errCode")-1, SwooleG.error TSRMLS_CC);
swoole_efree(buf);
RETURN_FALSE;
}
else
{
if (ret == 0)
{
swoole_efree(buf);
RETURN_EMPTY_STRING();
}
else
{
buf[ret] = 0;
SW_RETVAL_STRINGL(buf, ret, 0);
}
}
}
swClient_tcp_recv_no_buffer 同步 TCP 客戶端接受函數
上一小節中的 cli->recv 實際調用的是 swClient_tcp_recv_no_buffer 函數，無論是同步客戶端還是異步客戶端都是這個函數，該函數會調用 swConnection_recv 接受數據，直到達到超時時間。
值得注意的是這個函數中的 flag 如果是 MSG_WAITALL 標志，recv 會等待所有的數據(長度為 len 的數據)全部到達之后才會返回。
static int swClient_tcp_recv_no_buffer(swClient *cli, char *data, int len, int flag)
{
int ret;
while (1)
{
ret = swConnection_recv(cli->socket, data, len, flag);
if (ret >= 0)
{
break;
}
if (errno == EINTR)
{
if (cli->interrupt_time <= 0)
{
cli->interrupt_time = swoole_microtime();
}
else if (swoole_microtime() > cli->interrupt_time + cli->timeout)
{
break;
}
else
{
continue;
}
}
#ifdef SW_USE_OPENSSL
if (errno == EAGAIN && cli->socket->ssl)
{
int timeout_ms = (int) (cli->timeout * 1000);
if (cli->socket->ssl_want_read && swSocket_wait(cli->socket->fd, timeout_ms, SW_EVENT_READ) == SW_OK)
{
continue;
}
else if (cli->socket->ssl_want_write && swSocket_wait(cli->socket->fd, timeout_ms, SW_EVENT_WRITE) == SW_OK)
{
continue;
}
}
#endif
break;
}
return ret;
}
swClient_udp_recv 同步 UDP 客戶端
對于 UDP 來說，cli->recv 就是函數 swClient_udp_recv, 本函數會嘗試調用兩次 recvfrom:
static int swClient_udp_recv(swClient *cli, char *data, int length, int flags)
{
cli->remote_addr.len = sizeof(cli->remote_addr.addr);
int ret = recvfrom(cli->socket->fd, data, length, flags, (struct sockaddr *) &cli->remote_addr.addr, &cli->remote_addr.len);
if (ret < 0)
{
if (errno == EINTR)
{
ret = recvfrom(cli->socket->fd, data, length, flags, (struct sockaddr *) &cli->remote_addr, &cli->remote_addr.len);
}
else
{
return SW_ERR;
}
}
return ret;
}
swClient_onStreamRead 異步 TCP 客戶端讀就緒
對于異步 TCP 數據的接受，首先與異步客戶端的寫就緒類似，首先要判斷當前的 SSL 的狀態是否是 SW_SSL_STATE_WAIT_STREAM，再次進行 SSL 握手（具體原因不太清楚）。
判斷客戶端是否配置了 EOF 檢測或者長度檢測，如果配置了就調用 swProtocol_recv_check_eof/swProtocol_recv_check_length 接受完整的數據包，這兩天會調用 swClient_onPackage，進而調用 onReceive 函數。
如果沒有配置，那么就簡單的調用 swConnection_recv 接受數據，接受到數據之后會調用 onReceive。
static int swClient_onPackage(swConnection *conn, char *data, uint32_t length)
{
swClient *cli = (swClient *) conn->object;
cli->onReceive(conn->object, data, length);
return conn->close_wait ? SW_ERR : SW_OK;
}
static int swClient_onStreamRead(swReactor *reactor, swEvent *event)
{
int n;
swClient *cli = event->socket->object;
char *buf = cli->buffer->str + cli->buffer->length;
long buf_size = cli->buffer->size - cli->buffer->length;
#ifdef SW_USE_OPENSSL
if (cli->open_ssl && cli->socket->ssl_state == SW_SSL_STATE_WAIT_STREAM)
{
if (swClient_ssl_handshake(cli) < 0)
{
goto connect_fail;
}
if (cli->socket->ssl_state != SW_SSL_STATE_READY)
{
return SW_OK;
}
//ssl handshake sucess
else if (cli->onConnect)
{
execute_onConnect(cli);
}
}
#endif
if (cli->open_eof_check || cli->open_length_check)
{
swConnection *conn = cli->socket;
swProtocol *protocol = &cli->protocol;
if (cli->open_eof_check)
{
n = swProtocol_recv_check_eof(protocol, conn, cli->buffer);
}
else
{
n = swProtocol_recv_check_length(protocol, conn, cli->buffer);
}
if (n < 0)
{
return cli->close(cli);
}
else
{
if (conn->removed == 0 && cli->remove_delay)
{
swClient_sleep(cli);
cli->remove_delay = 0;
}
return SW_OK;
}
}
#ifdef SW_CLIENT_RECV_AGAIN
recv_again:
#endif
n = swConnection_recv(event->socket, buf, buf_size, 0);
if (n < 0)
{
__error:
switch (swConnection_error(errno))
{
case SW_ERROR:
swSysError("Read from socket[%d] failed.", event->fd);
return SW_OK;
case SW_CLOSE:
goto __close;
case SW_WAIT:
return SW_OK;
default:
return SW_OK;
}
}
else if (n == 0)
{
__close:
return cli->close(cli);
}
else
{
cli->onReceive(cli, buf, n);
#ifdef SW_CLIENT_RECV_AGAIN
if (n == buf_size)
{
goto recv_again;
}
#endif
return SW_OK;
}
return SW_OK;
}
swClient_onDgramRead 異步 UDP 客戶端接受數據
異步的 UDP 客戶端接受數據調用的和同步的客戶端相同，都是調用 swClient_udp_recv 函數。
static int swClient_onDgramRead(swReactor *reactor, swEvent *event)
{
swClient *cli = event->socket->object;
char buffer[SW_BUFFER_SIZE_UDP];
int n = swClient_udp_recv(cli, buffer, sizeof(buffer), 0);
if (n < 0)
{
return SW_ERR;
}
else
{
cli->onReceive(cli, buffer, n);
}
return SW_OK;
}
文章版權歸作者所有，未經允許請勿轉載,若此文章存在違規行為，您可以聯系管理員刪除。
轉載請注明本文地址：http://specialneedsforspecialkids.com/yun/29497.html
摘要：新建可以看到，自動采用包長檢測的方法該函數主要功能是設置各種回調函數值得注意的是第三個參數代表是否異步。發送數據函數并不是直接發送數據，而是將數據存儲在，等著寫事件就緒之后調用發送數據。 swReactorThread_dispatch 發送數據 reactor 線程會通過 swReactorThread_dispatch 發送數據，當采用 stream 發送數據的時候，會調用 sw...
摘要：有研究過框架的同學就會發現，其實最核心的，就是用了拓展加上拓展來實現其底層的網絡服務和多進程調度。我們在模式下，測試起五個進程主進程要等待回收我們，這樣就很簡單的實現了一個多進程的協程服務。 有研究過Workman框架的同學就會發現，其實workman最核心的，就是用了php socket拓展加上pcntl拓展來實現其底層的網絡服務和多進程調度。那我們今天就來探討如何使用Swoole的...
摘要：之后如果仍然有剩余未發送的數據，那么就如果已經沒有剩余數據了，繼續去取下一個數據包。拿到后，要用函數轉化為相應的類型即可得到包長值。 swPort_onRead_check_eof EOF 自動分包 我們前面說過，swPort_onRead_raw 是最簡單的向 worker 進程發送數據包的方法，swoole 會將從客戶端接受到的數據包，立刻發送給 worker 進程,用戶自己把...
摘要：兩個函數是可選回調函數。附帶了一組可信任證書。應該注意的是，驗證失敗并不意味著連接不能使用。在對證書進行驗證時，有一些安全性檢查并沒有執行，包括證書的失效檢查和對證書中通用名的有效性驗證。 前言 swoole_client 提供了 tcp/udp socket 的客戶端的封裝代碼，使用時僅需 new swoole_client 即可。 swoole 的 socket client 對比...
摘要：對于服務端來說，緩存默認是不能使用的，可以通過調用函數來進行設置生效。在回調函數中，首先申請一個大數數據結構，然后將其設定為，該值表示公鑰指數，然后利用函數生成秘鑰。此時需要調用函數將新的連接與綁定。 前言 上一篇文章我們講了 OpenSSL 的原理，接下來，我們來說說如何利用 openssl 第三方庫進行開發，來為 tcp 層進行 SSL 隧道加密 OpenSSL 初始化 在 sw...
閱讀 914·2021-11-22 13:54
閱讀 2843·2021-09-28 09:36
閱讀 2979·2019-08-30 15:55
閱讀 1951·2019-08-30 15:44
閱讀 544·2019-08-29 12:31
閱讀 2564·2019-08-28 18:18
閱讀 1199·2019-08-26 13:58
閱讀 1383·2019-08-26 13:44
