add close params, fix close wrong code in callback after close() sent on remote side
This commit is contained in:
Alexandre Konieczny
parent
23384dcd6e
commit
776227edcb
@ -216,9 +216,9 @@ namespace ix
|
||||
return getReadyState() == WebSocket_ReadyState_Closing;
|
||||
}
|
||||
|
||||
void WebSocket::close()
|
||||
void WebSocket::close(uint16_t code, const std::string& reason)
|
||||
{
|
||||
_ws.close();
|
||||
_ws.close(code, reason);
|
||||
}
|
||||
|
||||
void WebSocket::reconnectPerpetuallyIfDisconnected()
|
||||
|
||||
@ -111,7 +111,8 @@ namespace ix
|
||||
WebSocketSendInfo sendText(const std::string& text,
|
||||
const OnProgressCallback& onProgressCallback = nullptr);
|
||||
WebSocketSendInfo ping(const std::string& text);
|
||||
void close();
|
||||
void close(uint16_t code = 1000,
|
||||
const std::string& reason = "Normal closure");
|
||||
|
||||
void setOnMessageCallback(const OnMessageCallback& callback);
|
||||
static void setTrafficTrackerCallback(const OnTrafficTrackerCallback& callback);
|
||||
|
||||
@ -80,6 +80,7 @@ namespace ix
|
||||
WebSocketTransport::WebSocketTransport() :
|
||||
_useMask(true),
|
||||
_readyState(CLOSED),
|
||||
_treatAbnormalCloseAfterDispatch(false),
|
||||
_closeCode(kInternalErrorCode),
|
||||
_closeReason(kInternalErrorMessage),
|
||||
_closeWireSize(0),
|
||||
@ -299,16 +300,20 @@ namespace ix
|
||||
}
|
||||
else if (ret <= 0)
|
||||
{
|
||||
_rxbuf.clear();
|
||||
_socket->close();
|
||||
|
||||
if (_rxbuf.size() > 0)
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(_closeDataMutex);
|
||||
_closeCode = kAbnormalCloseCode;
|
||||
_closeReason = kAbnormalCloseMessage;
|
||||
_closeWireSize = 0;
|
||||
_closeRemote = true;
|
||||
_treatAbnormalCloseAfterDispatch = true;
|
||||
|
||||
setReadyState(CLOSING);
|
||||
}
|
||||
setReadyState(CLOSED);
|
||||
else
|
||||
{
|
||||
_treatAbnormalCloseAfterDispatch = false;
|
||||
internalClose(kAbnormalCloseCode, kAbnormalCloseMessage, 0, true);
|
||||
}
|
||||
|
||||
break;
|
||||
}
|
||||
else
|
||||
@ -400,7 +405,7 @@ namespace ix
|
||||
while (true)
|
||||
{
|
||||
wsheader_type ws;
|
||||
if (_rxbuf.size() < 2) return; /* Need at least 2 */
|
||||
if (_rxbuf.size() < 2) break; /* Need at least 2 */
|
||||
const uint8_t * data = (uint8_t *) &_rxbuf[0]; // peek, but don't consume
|
||||
ws.fin = (data[0] & 0x80) == 0x80;
|
||||
ws.rsv1 = (data[0] & 0x40) == 0x40;
|
||||
@ -408,7 +413,7 @@ namespace ix
|
||||
ws.mask = (data[1] & 0x80) == 0x80;
|
||||
ws.N0 = (data[1] & 0x7f);
|
||||
ws.header_size = 2 + (ws.N0 == 126? 2 : 0) + (ws.N0 == 127? 8 : 0) + (ws.mask? 4 : 0);
|
||||
if (_rxbuf.size() < ws.header_size) return; /* Need: ws.header_size - _rxbuf.size() */
|
||||
if (_rxbuf.size() < ws.header_size) break; /* Need: ws.header_size - _rxbuf.size() */
|
||||
|
||||
//
|
||||
// Calculate payload length:
|
||||
@ -552,7 +557,8 @@ namespace ix
|
||||
|
||||
bool remote = true;
|
||||
|
||||
close(code, reason, _rxbuf.size(), remote);
|
||||
//std::cout << this << " CLOSE FROM REMOTE" << code << " / " << reason << std::endl;
|
||||
internalClose(code, reason, _rxbuf.size(), remote);
|
||||
}
|
||||
else
|
||||
{
|
||||
@ -565,6 +571,13 @@ namespace ix
|
||||
_rxbuf.erase(_rxbuf.begin(),
|
||||
_rxbuf.begin() + ws.header_size + (size_t) ws.N);
|
||||
}
|
||||
|
||||
if (_readyState == CLOSING && _treatAbnormalCloseAfterDispatch)
|
||||
{
|
||||
_treatAbnormalCloseAfterDispatch = false;
|
||||
|
||||
internalClose(kAbnormalCloseCode, kAbnormalCloseMessage, 0, true);
|
||||
}
|
||||
}
|
||||
|
||||
std::string WebSocketTransport::getMergedChunks() const
|
||||
@ -859,7 +872,7 @@ namespace ix
|
||||
}
|
||||
}
|
||||
|
||||
void WebSocketTransport::close(uint16_t code, const std::string& reason, size_t closeWireSize, bool remote)
|
||||
void WebSocketTransport::close(uint16_t code, const std::string& reason, size_t closeWireSize)
|
||||
{
|
||||
_requestInitCancellation = true;
|
||||
|
||||
@ -877,11 +890,17 @@ namespace ix
|
||||
|
||||
bool compress = false;
|
||||
sendData(wsheader_type::CLOSE, closure, compress);
|
||||
|
||||
setReadyState(CLOSING);
|
||||
|
||||
_socket->wakeUpFromPoll(Socket::kCloseRequest);
|
||||
_socket->close();
|
||||
|
||||
internalClose(code, reason, closeWireSize, false);
|
||||
}
|
||||
|
||||
void WebSocketTransport::internalClose(uint16_t code, const std::string& reason, size_t closeWireSize, bool remote)
|
||||
{
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(_closeDataMutex);
|
||||
_closeCode = code;
|
||||
@ -889,7 +908,6 @@ namespace ix
|
||||
_closeWireSize = closeWireSize;
|
||||
_closeRemote = remote;
|
||||
}
|
||||
|
||||
setReadyState(CLOSED);
|
||||
}
|
||||
|
||||
|
||||
@ -87,8 +87,7 @@ namespace ix
|
||||
|
||||
void close(uint16_t code = 1000,
|
||||
const std::string& reason = "Normal closure",
|
||||
size_t closeWireSize = 0,
|
||||
bool remote = false);
|
||||
size_t closeWireSize = 0);
|
||||
|
||||
ReadyStateValues getReadyState() const;
|
||||
void setReadyState(ReadyStateValues readyStateValue);
|
||||
@ -147,6 +146,7 @@ namespace ix
|
||||
|
||||
// Hold the state of the connection (OPEN, CLOSED, etc...)
|
||||
std::atomic<ReadyStateValues> _readyState;
|
||||
std::atomic<bool> _treatAbnormalCloseAfterDispatch;
|
||||
|
||||
OnCloseCallback _onCloseCallback;
|
||||
uint16_t _closeCode;
|
||||
@ -201,6 +201,11 @@ namespace ix
|
||||
// No PONG data was received through the socket for longer than ping timeout delay
|
||||
bool pingTimeoutExceeded();
|
||||
|
||||
void internalClose(uint16_t code,
|
||||
const std::string& reason,
|
||||
size_t closeWireSize,
|
||||
bool remote);
|
||||
|
||||
void sendOnSocket();
|
||||
WebSocketSendInfo sendData(wsheader_type::opcode_type type,
|
||||
const std::string& message,
|
||||
|
||||
@ -35,6 +35,7 @@ set (SOURCES
|
||||
# Some unittest don't work on windows yet
|
||||
if (NOT WIN32)
|
||||
list(APPEND SOURCES
|
||||
IXWebSocketCloseTest.cpp
|
||||
IXWebSocketServerTest.cpp
|
||||
IXWebSocketPingTest.cpp
|
||||
IXWebSocketPingTimeoutTest.cpp
|
||||
|
||||
407
test/IXWebSocketCloseTest.cpp
Normal file
407
test/IXWebSocketCloseTest.cpp
Normal file
@ -0,0 +1,407 @@
|
||||
/*
|
||||
* IXWebSocketCloseTest.cpp
|
||||
* Author: Alexandre Konieczny
|
||||
* Copyright (c) 2019 Machine Zone. All rights reserved.
|
||||
*/
|
||||
|
||||
#include <iostream>
|
||||
#include <sstream>
|
||||
#include <queue>
|
||||
#include <ixwebsocket/IXWebSocket.h>
|
||||
#include <ixwebsocket/IXWebSocketServer.h>
|
||||
|
||||
#include "IXTest.h"
|
||||
|
||||
#include "catch.hpp"
|
||||
|
||||
using namespace ix;
|
||||
|
||||
namespace
|
||||
{
|
||||
class WebSocketClient
|
||||
{
|
||||
public:
|
||||
WebSocketClient(int port);
|
||||
|
||||
void subscribe(const std::string& channel);
|
||||
void start();
|
||||
void stop();
|
||||
void stop(uint16_t code, const std::string& reason);
|
||||
bool isReady() const;
|
||||
void sendMessage(const std::string& text);
|
||||
|
||||
uint16_t getCloseCode();
|
||||
const std::string& getCloseReason();
|
||||
bool getCloseRemote();
|
||||
|
||||
private:
|
||||
ix::WebSocket _webSocket;
|
||||
int _port;
|
||||
|
||||
mutable std::mutex _mutexCloseData;
|
||||
uint16_t _closeCode;
|
||||
std::string _closeReason;
|
||||
bool _closeRemote;
|
||||
};
|
||||
|
||||
WebSocketClient::WebSocketClient(int port)
|
||||
: _port(port)
|
||||
, _closeCode(0)
|
||||
, _closeReason(std::string(""))
|
||||
, _closeRemote(false)
|
||||
{
|
||||
;
|
||||
}
|
||||
|
||||
bool WebSocketClient::isReady() const
|
||||
{
|
||||
return _webSocket.getReadyState() == ix::WebSocket_ReadyState_Open;
|
||||
}
|
||||
|
||||
uint16_t WebSocketClient::getCloseCode()
|
||||
{
|
||||
std::lock_guard<std::mutex> lck(_mutexCloseData);
|
||||
|
||||
return _closeCode;
|
||||
}
|
||||
|
||||
const std::string& WebSocketClient::getCloseReason()
|
||||
{
|
||||
std::lock_guard<std::mutex> lck(_mutexCloseData);
|
||||
|
||||
return _closeReason;
|
||||
}
|
||||
|
||||
bool WebSocketClient::getCloseRemote()
|
||||
{
|
||||
std::lock_guard<std::mutex> lck(_mutexCloseData);
|
||||
|
||||
return _closeRemote;
|
||||
}
|
||||
|
||||
void WebSocketClient::stop()
|
||||
{
|
||||
_webSocket.stop();
|
||||
}
|
||||
|
||||
void WebSocketClient::stop(uint16_t code, const std::string& reason)
|
||||
{
|
||||
_webSocket.close(code, reason);
|
||||
_webSocket.stop();
|
||||
}
|
||||
|
||||
void WebSocketClient::start()
|
||||
{
|
||||
std::string url;
|
||||
{
|
||||
std::stringstream ss;
|
||||
ss << "ws://localhost:"
|
||||
<< _port
|
||||
<< "/";
|
||||
|
||||
url = ss.str();
|
||||
}
|
||||
|
||||
_webSocket.setUrl(url);
|
||||
|
||||
std::stringstream ss;
|
||||
log(std::string("Connecting to url: ") + url);
|
||||
|
||||
_webSocket.setOnMessageCallback(
|
||||
[this](ix::WebSocketMessageType messageType,
|
||||
const std::string& str,
|
||||
size_t wireSize,
|
||||
const ix::WebSocketErrorInfo& error,
|
||||
const ix::WebSocketOpenInfo& openInfo,
|
||||
const ix::WebSocketCloseInfo& closeInfo)
|
||||
{
|
||||
std::stringstream ss;
|
||||
if (messageType == ix::WebSocket_MessageType_Open)
|
||||
{
|
||||
log("client connected");
|
||||
|
||||
_webSocket.disableAutomaticReconnection();
|
||||
}
|
||||
else if (messageType == ix::WebSocket_MessageType_Close)
|
||||
{
|
||||
log("client disconnected");
|
||||
|
||||
std::lock_guard<std::mutex> lck(_mutexCloseData);
|
||||
|
||||
_closeCode = closeInfo.code;
|
||||
_closeReason = std::string(closeInfo.reason);
|
||||
_closeRemote = closeInfo.remote;
|
||||
|
||||
_webSocket.disableAutomaticReconnection();
|
||||
}
|
||||
else if (messageType == ix::WebSocket_MessageType_Error)
|
||||
{
|
||||
ss << "Error ! " << error.reason;
|
||||
log(ss.str());
|
||||
|
||||
_webSocket.disableAutomaticReconnection();
|
||||
}
|
||||
else if (messageType == ix::WebSocket_MessageType_Pong)
|
||||
{
|
||||
ss << "Received pong message " << str;
|
||||
log(ss.str());
|
||||
}
|
||||
else if (messageType == ix::WebSocket_MessageType_Ping)
|
||||
{
|
||||
ss << "Received ping message " << str;
|
||||
log(ss.str());
|
||||
}
|
||||
else if (messageType == ix::WebSocket_MessageType_Message)
|
||||
{
|
||||
ss << "Received message " << str;
|
||||
log(ss.str());
|
||||
}
|
||||
else
|
||||
{
|
||||
ss << "Invalid ix::WebSocketMessageType";
|
||||
log(ss.str());
|
||||
}
|
||||
});
|
||||
|
||||
_webSocket.start();
|
||||
}
|
||||
|
||||
void WebSocketClient::sendMessage(const std::string& text)
|
||||
{
|
||||
_webSocket.send(text);
|
||||
}
|
||||
|
||||
bool startServer(ix::WebSocketServer& server,
|
||||
uint16_t& receivedCloseCode,
|
||||
std::string& receivedCloseReason,
|
||||
bool& receivedCloseRemote,
|
||||
std::mutex& mutexWrite)
|
||||
{
|
||||
// A dev/null server
|
||||
server.setOnConnectionCallback(
|
||||
[&server, &receivedCloseCode, &receivedCloseReason, &receivedCloseRemote, &mutexWrite](std::shared_ptr<ix::WebSocket> webSocket,
|
||||
std::shared_ptr<ConnectionState> connectionState)
|
||||
{
|
||||
webSocket->setOnMessageCallback(
|
||||
[webSocket, connectionState, &server, &receivedCloseCode, &receivedCloseReason, &receivedCloseRemote, &mutexWrite](ix::WebSocketMessageType messageType,
|
||||
const std::string& str,
|
||||
size_t wireSize,
|
||||
const ix::WebSocketErrorInfo& error,
|
||||
const ix::WebSocketOpenInfo& openInfo,
|
||||
const ix::WebSocketCloseInfo& closeInfo)
|
||||
{
|
||||
if (messageType == ix::WebSocket_MessageType_Open)
|
||||
{
|
||||
Logger() << "New server connection";
|
||||
Logger() << "id: " << connectionState->getId();
|
||||
Logger() << "Uri: " << openInfo.uri;
|
||||
Logger() << "Headers:";
|
||||
for (auto it : openInfo.headers)
|
||||
{
|
||||
Logger() << it.first << ": " << it.second;
|
||||
}
|
||||
}
|
||||
else if (messageType == ix::WebSocket_MessageType_Close)
|
||||
{
|
||||
log("Server closed connection");
|
||||
|
||||
//Logger() << closeInfo.code;
|
||||
//Logger() << closeInfo.reason;
|
||||
//Logger() << closeInfo.remote;
|
||||
|
||||
std::lock_guard<std::mutex> lck(mutexWrite);
|
||||
|
||||
receivedCloseCode = closeInfo.code;
|
||||
receivedCloseReason = std::string(closeInfo.reason);
|
||||
receivedCloseRemote = closeInfo.remote;
|
||||
}
|
||||
}
|
||||
);
|
||||
}
|
||||
);
|
||||
|
||||
auto res = server.listen();
|
||||
if (!res.first)
|
||||
{
|
||||
log(res.second);
|
||||
return false;
|
||||
}
|
||||
|
||||
server.start();
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
TEST_CASE("Websocket_client_close_default", "[close]")
|
||||
{
|
||||
SECTION("Make sure that close code and reason was used and sent to server.")
|
||||
{
|
||||
ix::setupWebSocketTrafficTrackerCallback();
|
||||
|
||||
int port = getFreePort();
|
||||
ix::WebSocketServer server(port);
|
||||
|
||||
uint16_t serverReceivedCloseCode(0);
|
||||
bool serverReceivedCloseRemote(false);
|
||||
std::string serverReceivedCloseReason("");
|
||||
std::mutex mutexWrite;
|
||||
|
||||
REQUIRE(startServer(server, serverReceivedCloseCode, serverReceivedCloseReason, serverReceivedCloseRemote, mutexWrite));
|
||||
|
||||
std::string session = ix::generateSessionId();
|
||||
WebSocketClient webSocketClient(port);
|
||||
|
||||
webSocketClient.start();
|
||||
|
||||
// Wait for all chat instance to be ready
|
||||
while (true)
|
||||
{
|
||||
if (webSocketClient.isReady()) break;
|
||||
ix::msleep(10);
|
||||
}
|
||||
|
||||
REQUIRE(server.getClients().size() == 1);
|
||||
|
||||
ix::msleep(100);
|
||||
|
||||
webSocketClient.stop();
|
||||
|
||||
ix::msleep(200);
|
||||
|
||||
// ensure client close is the same as values given
|
||||
REQUIRE(webSocketClient.getCloseCode() == 1000);
|
||||
REQUIRE(webSocketClient.getCloseReason() == "Normal closure");
|
||||
REQUIRE(webSocketClient.getCloseRemote() == false);
|
||||
|
||||
{
|
||||
std::lock_guard<std::mutex> lck(mutexWrite);
|
||||
|
||||
// Here we read the code/reason received by the server, and ensure that remote is true
|
||||
REQUIRE(serverReceivedCloseCode == 1000);
|
||||
REQUIRE(serverReceivedCloseReason == "Normal closure");
|
||||
REQUIRE(serverReceivedCloseRemote == true);
|
||||
}
|
||||
|
||||
// Give us 1000ms for the server to notice that clients went away
|
||||
ix::msleep(1000);
|
||||
REQUIRE(server.getClients().size() == 0);
|
||||
|
||||
ix::reportWebSocketTraffic();
|
||||
}
|
||||
}
|
||||
|
||||
TEST_CASE("Websocket_client_close_params_given", "[close]")
|
||||
{
|
||||
SECTION("Make sure that close code and reason was used and sent to server.")
|
||||
{
|
||||
ix::setupWebSocketTrafficTrackerCallback();
|
||||
|
||||
int port = getFreePort();
|
||||
ix::WebSocketServer server(port);
|
||||
|
||||
uint16_t serverReceivedCloseCode(0);
|
||||
bool serverReceivedCloseRemote(false);
|
||||
std::string serverReceivedCloseReason("");
|
||||
std::mutex mutexWrite;
|
||||
|
||||
REQUIRE(startServer(server, serverReceivedCloseCode, serverReceivedCloseReason, serverReceivedCloseRemote, mutexWrite));
|
||||
|
||||
std::string session = ix::generateSessionId();
|
||||
WebSocketClient webSocketClient(port);
|
||||
|
||||
webSocketClient.start();
|
||||
|
||||
// Wait for all chat instance to be ready
|
||||
while (true)
|
||||
{
|
||||
if (webSocketClient.isReady()) break;
|
||||
ix::msleep(10);
|
||||
}
|
||||
|
||||
REQUIRE(server.getClients().size() == 1);
|
||||
|
||||
ix::msleep(100);
|
||||
|
||||
webSocketClient.stop(4000, "My reason");
|
||||
|
||||
ix::msleep(200);
|
||||
|
||||
// ensure client close is the same as values given
|
||||
REQUIRE(webSocketClient.getCloseCode() == 4000);
|
||||
REQUIRE(webSocketClient.getCloseReason() == "My reason");
|
||||
REQUIRE(webSocketClient.getCloseRemote() == false);
|
||||
|
||||
{
|
||||
std::lock_guard<std::mutex> lck(mutexWrite);
|
||||
|
||||
// Here we read the code/reason received by the server, and ensure that remote is true
|
||||
REQUIRE(serverReceivedCloseCode == 4000);
|
||||
REQUIRE(serverReceivedCloseReason == "My reason");
|
||||
REQUIRE(serverReceivedCloseRemote == true);
|
||||
}
|
||||
|
||||
// Give us 1000ms for the server to notice that clients went away
|
||||
ix::msleep(1000);
|
||||
REQUIRE(server.getClients().size() == 0);
|
||||
|
||||
ix::reportWebSocketTraffic();
|
||||
}
|
||||
}
|
||||
|
||||
TEST_CASE("Websocket_server_close", "[close]")
|
||||
{
|
||||
SECTION("Make sure that close code and reason was read from server.")
|
||||
{
|
||||
ix::setupWebSocketTrafficTrackerCallback();
|
||||
|
||||
int port = getFreePort();
|
||||
ix::WebSocketServer server(port);
|
||||
|
||||
uint16_t serverReceivedCloseCode(0);
|
||||
bool serverReceivedCloseRemote(false);
|
||||
std::string serverReceivedCloseReason("");
|
||||
std::mutex mutexWrite;
|
||||
|
||||
REQUIRE(startServer(server, serverReceivedCloseCode, serverReceivedCloseReason, serverReceivedCloseRemote, mutexWrite));
|
||||
|
||||
std::string session = ix::generateSessionId();
|
||||
WebSocketClient webSocketClient(port);
|
||||
|
||||
webSocketClient.start();
|
||||
|
||||
// Wait for all chat instance to be ready
|
||||
while (true)
|
||||
{
|
||||
if (webSocketClient.isReady()) break;
|
||||
ix::msleep(10);
|
||||
}
|
||||
|
||||
REQUIRE(server.getClients().size() == 1);
|
||||
|
||||
ix::msleep(100);
|
||||
|
||||
server.stop();
|
||||
|
||||
ix::msleep(200);
|
||||
|
||||
// ensure client close is the same as values given
|
||||
REQUIRE(webSocketClient.getCloseCode() == 1000);
|
||||
REQUIRE(webSocketClient.getCloseReason() == "Normal closure");
|
||||
REQUIRE(webSocketClient.getCloseRemote() == true);
|
||||
|
||||
{
|
||||
std::lock_guard<std::mutex> lck(mutexWrite);
|
||||
|
||||
// Here we read the code/reason received by the server, and ensure that remote is true
|
||||
REQUIRE(serverReceivedCloseCode == 1000);
|
||||
REQUIRE(serverReceivedCloseReason == "Normal closure");
|
||||
REQUIRE(serverReceivedCloseRemote == false);
|
||||
}
|
||||
|
||||
// Give us 1000ms for the server to notice that clients went away
|
||||
ix::msleep(1000);
|
||||
REQUIRE(server.getClients().size() == 0);
|
||||
|
||||
ix::reportWebSocketTraffic();
|
||||
}
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user