diff --git a/Makefile b/Makefile index 2b7f50464..287be43c4 100644 --- a/Makefile +++ b/Makefile @@ -413,6 +413,9 @@ unittest: prepare libhv $(CC) -g -Wall -O0 -std=c99 -I. -Ibase -Issl -Ievent -o bin/hdns_benchmark unittest/hdns_benchmark.c -Llib -lhv -pthread ifeq ($(WITH_EVPP), yes) $(CXX) -g -Wall -O0 -std=c++11 -I. -Ibase -Issl -Ievent -Icpputil -Ievpp -o bin/tcpclient_dns_test unittest/tcpclient_dns_test.cpp -Llib -lhv -pthread +ifeq ($(WITH_HTTP_CLIENT), yes) + $(CXX) -g -Wall -O0 -std=c++11 -I. -Ibase -Issl -Ievent -Icpputil -Ievpp -Ihttp -Ihttp/client -o bin/websocket_reconnect_test unittest/websocket_reconnect_test.cpp -Llib -lhv -pthread +endif endif ifeq ($(WITH_EVPP), yes) ifeq ($(WITH_REDIS), yes) diff --git a/http/client/WebSocketClient.cpp b/http/client/WebSocketClient.cpp index 375bbbfcf..2d1fc16bc 100644 --- a/http/client/WebSocketClient.cpp +++ b/http/client/WebSocketClient.cpp @@ -13,6 +13,9 @@ WebSocketClient::WebSocketClient(EventLoopPtr loop) state = WS_CLOSED; ping_interval = DEFAULT_WS_PING_INTERVAL; ping_cnt = 0; + reconn_cur_delay_ = 0; + reconn_cur_retry_cnt_ = 0; + reconn_upgrading_ = false; } WebSocketClient::~WebSocketClient() { @@ -72,6 +75,13 @@ int WebSocketClient::open(const char* _url, const http_headers& headers) { onConnection = [this](const WebSocketChannelPtr& channel) { if (channel->isConnected()) { + // NOTE: TcpClient resets reconn_setting after onConnection returns, + // restore it on close unless the websocket upgrade succeeds. + if (reconn_setting) { + reconn_cur_delay_ = reconn_setting->cur_delay; + reconn_cur_retry_cnt_ = reconn_setting->cur_retry_cnt; + reconn_upgrading_ = true; + } state = CONNECTED; // websocket_handshake http_req_->headers["Connection"] = "Upgrade"; @@ -100,6 +110,11 @@ int WebSocketClient::open(const char* _url, const http_headers& headers) { http_resp_ = std::make_shared(); http_parser_->InitResponse(http_resp_.get()); } else { + if (reconn_upgrading_ && reconn_setting) { + reconn_setting->cur_delay = reconn_cur_delay_; + reconn_setting->cur_retry_cnt = reconn_cur_retry_cnt_; + } + reconn_upgrading_ = false; state = WS_CLOSED; if (onclose) onclose(); } @@ -175,6 +190,7 @@ int WebSocketClient::open(const char* _url, const http_headers& headers) { } }; state = WS_OPENED; + reconn_upgrading_ = false; // ping if (ping_interval > 0) { ping_cnt = 0; diff --git a/http/client/WebSocketClient.h b/http/client/WebSocketClient.h index da0a4afa8..58ec13891 100644 --- a/http/client/WebSocketClient.h +++ b/http/client/WebSocketClient.h @@ -64,6 +64,10 @@ class HV_EXPORT WebSocketClient : public TcpClientTmpl { // ping/pong int ping_interval; int ping_cnt; + // reconnect backoff saved until upgraded + uint32_t reconn_cur_delay_; + uint32_t reconn_cur_retry_cnt_; + bool reconn_upgrading_; }; } diff --git a/scripts/unittest.sh b/scripts/unittest.sh index 5a4d05668..67a41be2a 100755 --- a/scripts/unittest.sh +++ b/scripts/unittest.sh @@ -74,6 +74,9 @@ fi if [ -x bin/tcpclient_dns_test ]; then bin/tcpclient_dns_test fi +if [ -x bin/websocket_reconnect_test ]; then + bin/websocket_reconnect_test +fi for redis_test in redis_async_client_test redis_client_test redis_batch_test redis_subscriber_test lua_redis_test; do if [ -x bin/${redis_test} ]; then if [ "${redis_test}" = "lua_redis_test" ]; then diff --git a/unittest/CMakeLists.txt b/unittest/CMakeLists.txt index d0a1327fc..0f49d8ec9 100644 --- a/unittest/CMakeLists.txt +++ b/unittest/CMakeLists.txt @@ -189,6 +189,12 @@ add_executable(tcpclient_dns_test tcpclient_dns_test.cpp) target_include_directories(tcpclient_dns_test PRIVATE .. ../base ../ssl ../event ../cpputil ../evpp) target_link_libraries(tcpclient_dns_test ${HV_LIBRARIES}) set(EVPP_DNS_UNITTEST_TARGETS tcpclient_dns_test) +if(WITH_HTTP AND WITH_HTTP_CLIENT) +add_executable(websocket_reconnect_test websocket_reconnect_test.cpp) +target_include_directories(websocket_reconnect_test PRIVATE .. ../base ../ssl ../event ../cpputil ../evpp ../http ../http/client) +target_link_libraries(websocket_reconnect_test ${HV_LIBRARIES}) +set(EVPP_WS_UNITTEST_TARGETS websocket_reconnect_test) +endif() endif() if(UNIX) @@ -227,4 +233,5 @@ add_custom_target(unittest DEPENDS hdns_benchmark ${REDIS_UNITTEST_TARGETS} ${EVPP_DNS_UNITTEST_TARGETS} + ${EVPP_WS_UNITTEST_TARGETS} ) diff --git a/unittest/websocket_reconnect_test.cpp b/unittest/websocket_reconnect_test.cpp new file mode 100644 index 000000000..9953f2979 --- /dev/null +++ b/unittest/websocket_reconnect_test.cpp @@ -0,0 +1,63 @@ +/* + * websocket_reconnect_test: a server that accepts TCP but rejects the + * websocket upgrade must not reset the client's reconnect backoff. + */ + +#include +#include + +#include "TcpServer.h" +#include "WebSocketClient.h" +#include "htime.h" + +using namespace hv; + +int main() { + int port = 40834; + TcpServer srv; + if (srv.createsocket(port, "127.0.0.1") < 0) { + printf("createsocket failed on port %d\n", port); + return 1; + } + srv.onMessage = [](const SocketChannelPtr& channel, Buffer* buf) { + channel->write("HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\n\r\n"); + }; + srv.start(); + + const int ncloses = 4; + std::atomic close_cnt{0}; + std::atomic retries{-1}; + std::atomic cur_delay{-1}; + + WebSocketClient ws; + reconn_setting_t reconn; + reconn_setting_init(&reconn); + reconn.min_delay = 10; + reconn.max_delay = 10000; + reconn.delay_policy = 2; + ws.setReconnect(&reconn); + ws.onclose = [&]() { + if (++close_cnt == ncloses) { + retries = ws.reconnectRetries(); + cur_delay = ws.reconn_setting->cur_delay; + ws.setReconnect(NULL); + } + }; + ws.open("ws://127.0.0.1:40834/"); + + uint64_t start = gettick_ms(); + while (close_cnt < ncloses && gettick_ms() - start < 5000) { + hv_msleep(10); + } + ws.stop(); + srv.stop(); + + // 3 reconnects scheduled before the 4th close: delays 10, 20, 40. + if (close_cnt >= ncloses && retries == ncloses - 1 && cur_delay == 40) { + printf("websocket_reconnect_test PASSED\n"); + return 0; + } + printf("websocket_reconnect_test FAILED: closes=%d retries=%d cur_delay=%d\n", + close_cnt.load(), retries.load(), cur_delay.load()); + return 1; +}