recv
接受數據客戶端接受數據須要指定緩存區最大長度,就是下面的 buf_len
,flags
用於指定是否設置 waitall
標誌,若是設定了 waitall
就必須設定準確的 size
,不然會一直等待,直到接收的數據長度達到 size
。php
客戶端啓用了 EOF/Length
檢測後,無需設置 size
和 waitall
參數。擴展層會返回完整的數據包或者返回false
。react
開啓了 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
接受數據,直到達到超時時間。swoole
值得注意的是這個函數中的 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
:socket
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
握手(具體緣由不太清楚)。tcp
判斷客戶端是否配置了 EOF
檢測或者長度檢測,若是配置了就調用 swProtocol_recv_check_eof
/swProtocol_recv_check_length
接受完整的數據包,這兩天會調用 swClient_onPackage
,進而調用 onReceive
函數。函數
若是沒有配置,那麼就簡單的調用 swConnection_recv
接受數據,接受到數據以後會調用 onReceive
。ui
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
函數。code
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; }