kapitan/IXWebSocket
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

wip

Browse Source
This commit is contained in:
Alexandre Konieczny 2019-04-24 16:43:22 +02:00
parent ec3896e61b
commit 1f518aa95d
6 changed files with 109 additions and 64 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
ixwebsocket/IXWebSocket.cpp
View File

@ -142,6 +142,7 @@ namespace ix
{
bool automaticReconnection = _automaticReconnection;
// This value needs to be forced when shutting down, it is restored later
_automaticReconnection = false;
@ -269,12 +270,13 @@ namespace ix
if (_stop) return;
// 2. Poll to see if there's any new data available
_ws.poll();
WebSocketTransport::PollPostTreatment pollPostTreatment = _ws.poll();
if (_stop) return;
//if (_stop) return;
// 3. Dispatch the incoming messages
_ws.dispatch(
pollPostTreatment,
[this](const std::string& msg,
size_t wireSize,
bool decompressionError,

2
ixwebsocket/IXWebSocketServer.cpp
View File

@ -43,7 +43,7 @@ namespace ix
client->close();
}
SocketServer::stop();
//SocketServer::stop();
}
void WebSocketServer::enablePong()

112
ixwebsocket/IXWebSocketTransport.cpp
View File

@ -68,7 +68,9 @@ namespace ix
const int WebSocketTransport::kDefaultPingIntervalSecs(-1);
const int WebSocketTransport::kDefaultPingTimeoutSecs(-1);
const bool WebSocketTransport::kDefaultEnablePong(true);
const int WebSocketTransport::kClosingMaximumWaitingDelayInMs(100);
constexpr size_t WebSocketTransport::kChunkSize;
const uint16_t WebSocketTransport::kInternalErrorCode(1011);
const uint16_t WebSocketTransport::kAbnormalCloseCode(1006);
const uint16_t WebSocketTransport::kProtocolErrorCode(1002);
@ -80,13 +82,13 @@ namespace ix
WebSocketTransport::WebSocketTransport() :
_useMask(true),
_readyState(CLOSED),
_treatAbnormalCloseAfterDispatch(false),
_closeCode(kInternalErrorCode),
_closeReason(kInternalErrorMessage),
_closeWireSize(0),
_closeRemote(false),
_enablePerMessageDeflate(false),
_requestInitCancellation(false),
_closingTimePoint(std::chrono::steady_clock::now()),
_enablePong(kDefaultEnablePong),
_pingIntervalSecs(kDefaultPingIntervalSecs),
_pingTimeoutSecs(kDefaultPingTimeoutSecs),
@ -243,7 +245,14 @@ namespace ix
return now - _lastReceivePongTimePoint > std::chrono::seconds(_pingTimeoutSecs);
}
void WebSocketTransport::poll()
bool WebSocketTransport::closingDelayExceeded()
{
std::lock_guard<std::mutex> lock(_closingTimePointMutex);
auto now = std::chrono::steady_clock::now();
return now - _closingTimePoint > std::chrono::milliseconds(kClosingMaximumWaitingDelayInMs);
}
WebSocketTransport::PollPostTreatment WebSocketTransport::poll()
{
PollResultType pollResult = _socket->poll(_pingIntervalOrTimeoutGCDSecs);
@ -300,24 +309,12 @@ namespace ix
}
else if (ret <= 0)
{
_socket->close();
// if there are received data pending to be processed, then delay the abnormal closure
// to after dispatch (other close code/reason could be read from the buffer)
if (_rxbuf.size() > 0)
{
_treatAbnormalCloseAfterDispatch = true;
_socket->close();
setReadyState(CLOSING);
}
// no received data pending processing, so we can close directly
else
{
_treatAbnormalCloseAfterDispatch = false;
internalClose(kAbnormalCloseCode, kAbnormalCloseMessage, 0, true);
}
break;
return CHECK_OR_RAISE_ABNORMAL_CLOSE_AFTER_DISPATCH;
}
else
{
@ -336,12 +333,14 @@ namespace ix
_socket->close();
}
// Avoid a race condition where we get stuck in select
// while closing.
if (_readyState == CLOSING)
if (_readyState == CLOSING && closingDelayExceeded())
{
// close code and reason were set when calling close()
_socket->close();
setReadyState(CLOSED);
}
return NONE;
}
bool WebSocketTransport::isSendBufferEmpty() const
@ -403,7 +402,7 @@ namespace ix
// | Payload Data continued ... |
// +---------------------------------------------------------------+
//
void WebSocketTransport::dispatch(const OnMessageCallback& onMessageCallback)
void WebSocketTransport::dispatch(WebSocketTransport::PollPostTreatment pollPostTreatment, const OnMessageCallback& onMessageCallback)
{
while (true)
{
@ -558,9 +557,25 @@ namespace ix
std::string reason(_rxbuf.begin()+ws.header_size + 2,
_rxbuf.begin()+ws.header_size + (size_t) ws.N);
bool remote = true;
// We receive a CLOSE frame from remote and are NOT the ones who triggered the close
if (_readyState != CLOSING)
{
//send back the CLOSE frame
sendCloseFrame(code, reason);
internalClose(code, reason, _rxbuf.size(), remote);
_socket->wakeUpFromPoll(Socket::kCloseRequest);
bool remote = true;
closeSocketAndSwitchToClosedState(code, reason, _rxbuf.size(), remote);
}
// we got the CLOSE frame answer from our close, so we can close the connection if
// the code/reason are the same
else if (_closeCode == code && _closeReason == reason)
{
bool remote = false;
closeSocketAndSwitchToClosedState(code, reason, _rxbuf.size(), remote);
}
}
else
{
@ -574,13 +589,11 @@ namespace ix
_rxbuf.begin() + ws.header_size + (size_t) ws.N);
}
// if an abnormal closure was raised, and nothing else triggered a CLOSED state in
// the received and processed data, then close using abnormal close code and message
if (_readyState == CLOSING && _treatAbnormalCloseAfterDispatch)
// if an abnormal closure was raised in poll, and nothing else triggered a CLOSED state in
// the received and processed data, then close uising abnormal close code and message
if (_readyState != CLOSED && _readyState != CLOSING && pollPostTreatment == CHECK_OR_RAISE_ABNORMAL_CLOSE_AFTER_DISPATCH)
{
_treatAbnormalCloseAfterDispatch = false;
internalClose(kAbnormalCloseCode, kAbnormalCloseMessage, 0, true);
closeSocketAndSwitchToClosedState(kAbnormalCloseCode, kAbnormalCloseMessage, 0, true);
}
}
@ -876,12 +889,9 @@ namespace ix
}
}
void WebSocketTransport::close(uint16_t code, const std::string& reason, size_t closeWireSize)
void WebSocketTransport::sendCloseFrame(uint16_t code, const std::string& reason)
{
_requestInitCancellation = true;
if (_readyState == CLOSING || _readyState == CLOSED) return;
// See list of close events here:
// https://developer.mozilla.org/en-US/docs/Web/API/CloseEvent
@ -894,17 +904,12 @@ 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)
void WebSocketTransport::closeSocketAndSwitchToClosedState(uint16_t code, const std::string& reason, size_t closeWireSize, bool remote)
{
_socket->close();
{
std::lock_guard<std::mutex> lock(_closeDataMutex);
_closeCode = code;
@ -915,6 +920,29 @@ namespace ix
setReadyState(CLOSED);
}
void WebSocketTransport::close(uint16_t code, const std::string& reason, size_t closeWireSize, bool remote)
{
_requestInitCancellation = true;
if (_readyState == CLOSING || _readyState == CLOSED) return;
sendCloseFrame(code, reason);
setReadyState(CLOSING);
{
std::lock_guard<std::mutex> lock(_closeDataMutex);
_closeCode = code;
_closeReason = reason;
_closeWireSize = closeWireSize;
_closeRemote = remote;
}
{
std::lock_guard<std::mutex> lock(_closingTimePointMutex);
_closingTimePoint = std::chrono::steady_clock::now();
}
}
size_t WebSocketTransport::bufferedAmount() const
{
std::lock_guard<std::mutex> lock(_txbufMutex);

31
ixwebsocket/IXWebSocketTransport.h
View File

@ -56,6 +56,12 @@ namespace ix
FRAGMENT
};
enum PollPostTreatment
{
NONE,
CHECK_OR_RAISE_ABNORMAL_CLOSE_AFTER_DISPATCH
};
using OnMessageCallback = std::function<void(const std::string&,
size_t,
bool,
@ -78,7 +84,7 @@ namespace ix
WebSocketInitResult connectToSocket(int fd, // Server
int timeoutSecs);
void poll();
PollPostTreatment poll();
WebSocketSendInfo sendBinary(const std::string& message,
const OnProgressCallback& onProgressCallback);
WebSocketSendInfo sendText(const std::string& message,
@ -87,12 +93,13 @@ namespace ix
void close(uint16_t code = 1000,
const std::string& reason = "Normal closure",
size_t closeWireSize = 0);
size_t closeWireSize = 0,
bool remote = false);
ReadyStateValues getReadyState() const;
void setReadyState(ReadyStateValues readyStateValue);
void setOnCloseCallback(const OnCloseCallback& onCloseCallback);
void dispatch(const OnMessageCallback& onMessageCallback);
void dispatch(PollPostTreatment pollPostTreatment, const OnMessageCallback& onMessageCallback);
size_t bufferedAmount() const;
private:
@ -146,7 +153,6 @@ namespace ix
// Hold the state of the connection (OPEN, CLOSED, etc...)
std::atomic<ReadyStateValues> _readyState;
std::atomic<bool> _treatAbnormalCloseAfterDispatch;
OnCloseCallback _onCloseCallback;
uint16_t _closeCode;
@ -162,6 +168,10 @@ namespace ix
// Used to cancel dns lookup + socket connect + http upgrade
std::atomic<bool> _requestInitCancellation;
mutable std::mutex _closingTimePointMutex;
std::chrono::time_point<std::chrono::steady_clock>_closingTimePoint;
static const int kClosingMaximumWaitingDelayInMs;
// Constants for dealing with closing conneections
static const uint16_t kInternalErrorCode;
@ -201,10 +211,15 @@ 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);
// after calling close(), if no CLOSE frame answer is received back from the remote, we should close the connexion
bool closingDelayExceeded();
void sendCloseFrame(uint16_t code, const std::string& reason);
void closeSocketAndSwitchToClosedState(uint16_t code,
const std::string& reason,
size_t closeWireSize,
bool remote);
void sendOnSocket();
WebSocketSendInfo sendData(wsheader_type::opcode_type type,

10
test/CMakeLists.txt
View File

@ -35,12 +35,12 @@ set (SOURCES
# Some unittest don't work on windows yet
if (NOT WIN32)
list(APPEND SOURCES
IXWebSocketCloseTest.cpp
#IXWebSocketCloseTest.cpp
IXWebSocketServerTest.cpp
IXWebSocketPingTest.cpp
IXWebSocketPingTimeoutTest.cpp
cmd_websocket_chat.cpp
IXWebSocketTestConnectionDisconnection.cpp
#IXWebSocketPingTest.cpp
#IXWebSocketPingTimeoutTest.cpp
#cmd_websocket_chat.cpp
#IXWebSocketTestConnectionDisconnection.cpp
)
endif()

12
test/IXWebSocketCloseTest.cpp
View File

@ -131,7 +131,7 @@ namespace
_closeCode = closeInfo.code;
_closeReason = std::string(closeInfo.reason);
_closeRemote = closeInfo.remote;
_webSocket.disableAutomaticReconnection();
}
else if (messageType == ix::WebSocket_MessageType_Error)
@ -231,7 +231,7 @@ namespace
return true;
}
}
/*
TEST_CASE("Websocket_client_close_default", "[close]")
{
SECTION("Make sure that close code and reason was used and sent to server.")
@ -289,7 +289,7 @@ TEST_CASE("Websocket_client_close_default", "[close]")
ix::reportWebSocketTraffic();
}
}
*/
TEST_CASE("Websocket_client_close_params_given", "[close]")
{
SECTION("Make sure that close code and reason was used and sent to server.")
@ -324,7 +324,7 @@ TEST_CASE("Websocket_client_close_params_given", "[close]")
webSocketClient.stop(4000, "My reason");
ix::msleep(200);
ix::msleep(500);
// ensure client close is the same as values given
REQUIRE(webSocketClient.getCloseCode() == 4000);
@ -378,11 +378,11 @@ TEST_CASE("Websocket_server_close", "[close]")
REQUIRE(server.getClients().size() == 1);
ix::msleep(100);
ix::msleep(200);
server.stop();
ix::msleep(200);
ix::msleep(500);
// ensure client close is the same as values given
REQUIRE(webSocketClient.getCloseCode() == 1000);