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

Real ping (#32)

Browse Source 
* close method change and fix code

* missing mutex

* wip

* renaming and fixes

* renaming, fixes

* added enablePong/disablePong, add tests

* added new test cases

* add 1 test case

* fix gcd name to greatestCommonDivisor

* move ping and ping timeout checks into socket poll, local var in test cases and indent fixes

* indent issue
This commit is contained in:
Kumamon38
2019-04-18 18:24:16 +02:00
committed by Benjamin Sergeant
parent f0375e59fa
commit d7595b0dd0
9 changed files with 850 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
ixwebsocket/IXWebSocket.cpp
View File

@ -31,14 +31,18 @@ namespace ix
{
OnTrafficTrackerCallback WebSocket::_onTrafficTrackerCallback = nullptr;
const int WebSocket::kDefaultHandShakeTimeoutSecs(60);
const int WebSocket::kDefaultHeartBeatPeriod(-1);
const int WebSocket::kDefaultPingIntervalSecs(-1);
const int WebSocket::kDefaultPingTimeoutSecs(-1);
const bool WebSocket::kDefaultEnablePong(true);
WebSocket::WebSocket() :
_onMessageCallback(OnMessageCallback()),
_stop(false),
_automaticReconnection(true),
_handshakeTimeoutSecs(kDefaultHandShakeTimeoutSecs),
_heartBeatPeriod(kDefaultHeartBeatPeriod)
_enablePong(kDefaultEnablePong),
_pingIntervalSecs(kDefaultPingIntervalSecs),
_pingTimeoutSecs(kDefaultPingTimeoutSecs)
{
_ws.setOnCloseCallback(
[this](uint16_t code, const std::string& reason, size_t wireSize)
@ -79,18 +83,54 @@ namespace ix
return _perMessageDeflateOptions;
}
void WebSocket::setHeartBeatPeriod(int heartBeatPeriod)
void WebSocket::setHeartBeatPeriod(int heartBeatPeriodSecs)
{
std::lock_guard<std::mutex> lock(_configMutex);
_heartBeatPeriod = heartBeatPeriod;
_pingIntervalSecs = heartBeatPeriodSecs;
}
int WebSocket::getHeartBeatPeriod() const
{
std::lock_guard<std::mutex> lock(_configMutex);
return _heartBeatPeriod;
return _pingIntervalSecs;
}
void WebSocket::setPingInterval(int pingIntervalSecs)
{
std::lock_guard<std::mutex> lock(_configMutex);
_pingIntervalSecs = pingIntervalSecs;
}
int WebSocket::getPingInterval() const
{
std::lock_guard<std::mutex> lock(_configMutex);
return _pingIntervalSecs;
}
void WebSocket::setPingTimeout(int pingTimeoutSecs)
{
std::lock_guard<std::mutex> lock(_configMutex);
_pingTimeoutSecs = pingTimeoutSecs;
}
int WebSocket::getPingTimeout() const
{
std::lock_guard<std::mutex> lock(_configMutex);
return _pingTimeoutSecs;
}
void WebSocket::enablePong()
{
std::lock_guard<std::mutex> lock(_configMutex);
_enablePong = true;
}
void WebSocket::disablePong()
{
std::lock_guard<std::mutex> lock(_configMutex);
_enablePong = false;
}
void WebSocket::start()
{
if (_thread.joinable()) return; // we've already been started
@ -125,7 +165,9 @@ namespace ix
{
std::lock_guard<std::mutex> lock(_configMutex);
_ws.configure(_perMessageDeflateOptions,
_heartBeatPeriod);
_enablePong,
_pingIntervalSecs,
_pingTimeoutSecs);
}
WebSocketInitResult status = _ws.connectToUrl(_url, timeoutSecs);
@ -145,7 +187,10 @@ namespace ix
{
{
std::lock_guard<std::mutex> lock(_configMutex);
_ws.configure(_perMessageDeflateOptions, _heartBeatPeriod);
_ws.configure(_perMessageDeflateOptions,
_enablePong,
_pingIntervalSecs,
_pingTimeoutSecs);
}
WebSocketInitResult status = _ws.connectToSocket(fd, timeoutSecs);

22
ixwebsocket/IXWebSocket.h
View File

@ -89,8 +89,12 @@ namespace ix
void setUrl(const std::string& url);
void setPerMessageDeflateOptions(const WebSocketPerMessageDeflateOptions& perMessageDeflateOptions);
void setHandshakeTimeout(int handshakeTimeoutSecs);
void setHeartBeatPeriod(int heartBeatPeriod);
void setHeartBeatPeriod(int heartBeatPeriodSecs);
void setPingInterval(int pingIntervalSecs); // alias of setHeartBeatPeriod
void setPingTimeout(int pingTimeoutSecs);
void enablePong();
void disablePong();
// Run asynchronously, by calling start and stop.
void start();
void stop();
@ -114,6 +118,8 @@ namespace ix
const std::string& getUrl() const;
const WebSocketPerMessageDeflateOptions& getPerMessageDeflateOptions() const;
int getHeartBeatPeriod() const;
int getPingInterval() const;
int getPingTimeout() const;
size_t bufferedAmount() const;
void enableAutomaticReconnection();
@ -152,9 +158,15 @@ namespace ix
std::atomic<int> _handshakeTimeoutSecs;
static const int kDefaultHandShakeTimeoutSecs;
// Optional Heartbeat
int _heartBeatPeriod;
static const int kDefaultHeartBeatPeriod;
// enable or disable PONG frame response to received PING frame
bool _enablePong;
static const bool kDefaultEnablePong;
// Optional ping and ping timeout
int _pingIntervalSecs;
int _pingTimeoutSecs;
static const int kDefaultPingIntervalSecs;
static const int kDefaultPingTimeoutSecs;
friend class WebSocketServer;
};

19
ixwebsocket/IXWebSocketServer.cpp
View File

@ -17,13 +17,15 @@
namespace ix
{
const int WebSocketServer::kDefaultHandShakeTimeoutSecs(3); // 3 seconds
const bool WebSocketServer::kDefaultEnablePong(true);
WebSocketServer::WebSocketServer(int port,
const std::string& host,
int backlog,
size_t maxConnections,
int handshakeTimeoutSecs) : SocketServer(port, host, backlog, maxConnections),
_handshakeTimeoutSecs(handshakeTimeoutSecs)
_handshakeTimeoutSecs(handshakeTimeoutSecs),
_enablePong(kDefaultEnablePong)
{
}
@ -44,6 +46,16 @@ namespace ix
SocketServer::stop();
}
void WebSocketServer::enablePong()
{
_enablePong = true;
}
void WebSocketServer::disablePong()
{
_enablePong = false;
}
void WebSocketServer::setOnConnectionCallback(const OnConnectionCallback& callback)
{
_onConnectionCallback = callback;
@ -58,6 +70,11 @@ namespace ix
webSocket->disableAutomaticReconnection();
if (_enablePong)
webSocket->enablePong();
else
webSocket->disablePong();
// Add this client to our client set
{
std::lock_guard<std::mutex> lock(_clientsMutex);

5
ixwebsocket/IXWebSocketServer.h
View File

@ -33,6 +33,9 @@ namespace ix
virtual ~WebSocketServer();
virtual void stop() final;
void enablePong();
void disablePong();
void setOnConnectionCallback(const OnConnectionCallback& callback);
// Get all the connected clients
@ -41,6 +44,7 @@ namespace ix
private:
// Member variables
int _handshakeTimeoutSecs;
bool _enablePong;
OnConnectionCallback _onConnectionCallback;
@ -48,6 +52,7 @@ namespace ix
std::set<std::shared_ptr<WebSocket>> _clients;
const static int kDefaultHandShakeTimeoutSecs;
const static bool kDefaultEnablePong;
// Methods
virtual void handleConnection(int fd,

116
ixwebsocket/IXWebSocketTransport.cpp
View File

@ -51,10 +51,23 @@
#include <thread>
int greatestCommonDivisor (int a, int b) {
while (b != 0)
{
int t = b;
b = a % b;
a = t;
}
return a;
}
namespace ix
{
const std::string WebSocketTransport::kHeartBeatPingMessage("ixwebsocket::heartbeat");
const int WebSocketTransport::kDefaultHeartBeatPeriod(-1);
const std::string WebSocketTransport::kPingMessage("ixwebsocket::heartbeat");
const int WebSocketTransport::kDefaultPingIntervalSecs(-1);
const int WebSocketTransport::kDefaultPingTimeoutSecs(-1);
const bool WebSocketTransport::kDefaultEnablePong(true);
constexpr size_t WebSocketTransport::kChunkSize;
WebSocketTransport::WebSocketTransport() :
@ -64,8 +77,12 @@ namespace ix
_closeWireSize(0),
_enablePerMessageDeflate(false),
_requestInitCancellation(false),
_heartBeatPeriod(kDefaultHeartBeatPeriod),
_lastSendTimePoint(std::chrono::steady_clock::now())
_enablePong(kDefaultEnablePong),
_pingIntervalSecs(kDefaultPingIntervalSecs),
_pingTimeoutSecs(kDefaultPingTimeoutSecs),
_pingIntervalOrTimeoutGCDSecs(-1),
_lastSendPingTimePoint(std::chrono::steady_clock::now()),
_lastReceivePongTimePoint(std::chrono::steady_clock::now())
{
_readbuf.resize(kChunkSize);
}
@ -76,11 +93,21 @@ namespace ix
}
void WebSocketTransport::configure(const WebSocketPerMessageDeflateOptions& perMessageDeflateOptions,
int heartBeatPeriod)
bool enablePong,
int pingIntervalSecs, int pingTimeoutSecs)
{
_perMessageDeflateOptions = perMessageDeflateOptions;
_enablePerMessageDeflate = _perMessageDeflateOptions.enabled();
_heartBeatPeriod = heartBeatPeriod;
_enablePong = enablePong;
_pingIntervalSecs = pingIntervalSecs;
_pingTimeoutSecs = pingTimeoutSecs;
if (pingIntervalSecs > 0 && pingTimeoutSecs > 0)
_pingIntervalOrTimeoutGCDSecs = greatestCommonDivisor(pingIntervalSecs, pingTimeoutSecs);
else if (_pingTimeoutSecs > 0)
_pingIntervalOrTimeoutGCDSecs = pingTimeoutSecs;
else
_pingIntervalOrTimeoutGCDSecs = pingIntervalSecs;
}
// Client
@ -176,13 +203,25 @@ namespace ix
_onCloseCallback = onCloseCallback;
}
// Only consider send time points for that computation.
// The receive time points is taken into account in Socket::poll (second parameter).
bool WebSocketTransport::heartBeatPeriodExceeded()
// Only consider send PING time points for that computation.
bool WebSocketTransport::pingIntervalExceeded()
{
std::lock_guard<std::mutex> lock(_lastSendTimePointMutex);
if (_pingIntervalSecs <= 0)
return false;
std::lock_guard<std::mutex> lock(_lastSendPingTimePointMutex);
auto now = std::chrono::steady_clock::now();
return now - _lastSendTimePoint > std::chrono::seconds(_heartBeatPeriod);
return now - _lastSendPingTimePoint > std::chrono::seconds(_pingIntervalSecs);
}
bool WebSocketTransport::pingTimeoutExceeded()
{
if (_pingTimeoutSecs <= 0)
return false;
std::lock_guard<std::mutex> lock(_lastReceivePongTimePointMutex);
auto now = std::chrono::steady_clock::now();
return now - _lastReceivePongTimePoint > std::chrono::seconds(_pingTimeoutSecs);
}
void WebSocketTransport::poll()
@ -190,19 +229,27 @@ namespace ix
_socket->poll(
[this](PollResultType pollResult)
{
// If (1) heartbeat is enabled, and (2) no data was received or
// send for a duration exceeding our heart-beat period, send a
// ping to the server.
if (pollResult == PollResultType::Timeout &&
heartBeatPeriodExceeded())
if (_readyState == OPEN)
{
std::stringstream ss;
ss << kHeartBeatPingMessage << "::" << _heartBeatPeriod << "s";
sendPing(ss.str());
// if (1) ping timeout is enabled and (2) duration since last received ping response (PONG)
// exceeds the maximum delay, then close the connection
if (pingTimeoutExceeded())
{
close(1011, "Ping timeout");
}
// If (1) ping is enabled and no ping has been sent for a duration
// exceeding our ping interval, send a ping to the server.
else if (pingIntervalExceeded())
{
std::stringstream ss;
ss << kPingMessage << "::" << _pingIntervalSecs << "s";
sendPing(ss.str());
}
}
// Make sure we send all the buffered data
// there can be a lot of it for large messages.
else if (pollResult == PollResultType::SendRequest)
if (pollResult == PollResultType::SendRequest)
{
while (!isSendBufferEmpty() && !_requestInitCancellation)
{
@ -264,7 +311,7 @@ namespace ix
_socket->close();
}
},
_heartBeatPeriod);
_pingIntervalOrTimeoutGCDSecs);
}
bool WebSocketTransport::isSendBufferEmpty() const
@ -444,12 +491,16 @@ namespace ix
else if (ws.opcode == wsheader_type::PING)
{
unmaskReceiveBuffer(ws);
std::string pingData(_rxbuf.begin()+ws.header_size,
_rxbuf.begin()+ws.header_size + (size_t) ws.N);
// Reply back right away
bool compress = false;
sendData(wsheader_type::PONG, pingData, compress);
if (_enablePong)
{
// Reply back right away
bool compress = false;
sendData(wsheader_type::PONG, pingData, compress);
}
emitMessage(PING, pingData, ws, onMessageCallback);
}
@ -459,6 +510,9 @@ namespace ix
std::string pongData(_rxbuf.begin()+ws.header_size,
_rxbuf.begin()+ws.header_size + (size_t) ws.N);
std::lock_guard<std::mutex> lck(_lastReceivePongTimePointMutex);
_lastReceivePongTimePoint = std::chrono::steady_clock::now();
emitMessage(PONG, pongData, ws, onMessageCallback);
}
else if (ws.opcode == wsheader_type::CLOSE)
@ -724,7 +778,16 @@ namespace ix
WebSocketSendInfo WebSocketTransport::sendPing(const std::string& message)
{
bool compress = false;
return sendData(wsheader_type::PING, message, compress);
WebSocketSendInfo info = sendData(wsheader_type::PING, message, compress);
if(info.success)
{
std::lock_guard<std::mutex> lck(_lastSendPingTimePointMutex);
_lastSendPingTimePoint = std::chrono::steady_clock::now();
}
return info;
}
WebSocketSendInfo WebSocketTransport::sendBinary(
@ -770,9 +833,6 @@ namespace ix
_txbuf.erase(_txbuf.begin(), _txbuf.begin() + ret);
}
}
std::lock_guard<std::mutex> lck(_lastSendTimePointMutex);
_lastSendTimePoint = std::chrono::steady_clock::now();
}
void WebSocketTransport::close(uint16_t code, const std::string& reason, size_t closeWireSize)

29
ixwebsocket/IXWebSocketTransport.h
View File

@ -68,7 +68,8 @@ namespace ix
~WebSocketTransport();
void configure(const WebSocketPerMessageDeflateOptions& perMessageDeflateOptions,
int heartBeatPeriod);
bool enablePong,
int pingIntervalSecs, int pingTimeoutSecs);
WebSocketInitResult connectToUrl(const std::string& url, // Client
int timeoutSecs);
@ -158,15 +159,25 @@ namespace ix
// Used to cancel dns lookup + socket connect + http upgrade
std::atomic<bool> _requestInitCancellation;
// Optional Heartbeat
int _heartBeatPeriod;
static const int kDefaultHeartBeatPeriod;
const static std::string kHeartBeatPingMessage;
mutable std::mutex _lastSendTimePointMutex;
std::chrono::time_point<std::chrono::steady_clock> _lastSendTimePoint;
// enable auto response to ping
bool _enablePong;
static const bool kDefaultEnablePong;
// No data was send through the socket for longer than the heartbeat period
bool heartBeatPeriodExceeded();
// Optional ping and ping timeout
int _pingIntervalSecs;
int _pingTimeoutSecs;
int _pingIntervalOrTimeoutGCDSecs; // if both ping interval and timeout are set (> 0), then use GCD of these value to wait for the lowest time
static const int kDefaultPingIntervalSecs;
static const int kDefaultPingTimeoutSecs;
const static std::string kPingMessage;
mutable std::mutex _lastSendPingTimePointMutex;
mutable std::mutex _lastReceivePongTimePointMutex;
std::chrono::time_point<std::chrono::steady_clock> _lastSendPingTimePoint;
std::chrono::time_point<std::chrono::steady_clock> _lastReceivePongTimePoint;
bool pingIntervalExceeded();
// No PONG data was received through the socket for longer than ping timeout delay
bool pingTimeoutExceeded();
void sendOnSocket();
WebSocketSendInfo sendData(wsheader_type::opcode_type type,