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

(websocket) WebSocketMessagePtr is a unique_ptr instead of a shared_ptr

Browse Source
This commit is contained in:
Benjamin Sergeant
2020-04-13 21:56:01 -07:00
parent 37a054723a
commit 432f0570f4
8 changed files with 11 additions and 314 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
CMakeLists.txt
View File

@@ -44,7 +44,6 @@ set( IXWEBSOCKET_SOURCES
ixwebsocket/IXWebSocketCloseConstants.cpp ixwebsocket/IXWebSocketCloseConstants.cpp
ixwebsocket/IXWebSocketHandshake.cpp ixwebsocket/IXWebSocketHandshake.cpp
ixwebsocket/IXWebSocketHttpHeaders.cpp ixwebsocket/IXWebSocketHttpHeaders.cpp
ixwebsocket/IXWebSocketMessageQueue.cpp
ixwebsocket/IXWebSocketPerMessageDeflate.cpp ixwebsocket/IXWebSocketPerMessageDeflate.cpp
ixwebsocket/IXWebSocketPerMessageDeflateCodec.cpp ixwebsocket/IXWebSocketPerMessageDeflateCodec.cpp
ixwebsocket/IXWebSocketPerMessageDeflateOptions.cpp ixwebsocket/IXWebSocketPerMessageDeflateOptions.cpp
@@ -84,7 +83,6 @@ set( IXWEBSOCKET_HEADERS
ixwebsocket/IXWebSocketHttpHeaders.h ixwebsocket/IXWebSocketHttpHeaders.h
ixwebsocket/IXWebSocketInitResult.h ixwebsocket/IXWebSocketInitResult.h
ixwebsocket/IXWebSocketMessage.h ixwebsocket/IXWebSocketMessage.h
ixwebsocket/IXWebSocketMessageQueue.h
ixwebsocket/IXWebSocketMessageType.h ixwebsocket/IXWebSocketMessageType.h
ixwebsocket/IXWebSocketOpenInfo.h ixwebsocket/IXWebSocketOpenInfo.h
ixwebsocket/IXWebSocketPerMessageDeflate.h ixwebsocket/IXWebSocketPerMessageDeflate.h

4
docs/CHANGELOG.md
View File

@@ -1,6 +1,10 @@
# Changelog # Changelog
All changes to this project will be documented in this file. All changes to this project will be documented in this file.
## [9.2.5] - 2020-04-13
(websocket) WebSocketMessagePtr is a unique_ptr instead of a shared_ptr
## [9.2.4] - 2020-04-13 ## [9.2.4] - 2020-04-13
(websocket) use persistent member variable as temp variables to encode/decode zlib messages in order to reduce transient allocations (websocket) use persistent member variable as temp variables to encode/decode zlib messages in order to reduce transient allocations

10
ixwebsocket/IXWebSocket.cpp
View File

@@ -34,7 +34,7 @@ namespace ix
_ws.setOnCloseCallback( _ws.setOnCloseCallback(
[this](uint16_t code, const std::string& reason, size_t wireSize, bool remote) { [this](uint16_t code, const std::string& reason, size_t wireSize, bool remote) {
_onMessageCallback( _onMessageCallback(
std::make_shared<WebSocketMessage>(WebSocketMessageType::Close, std::make_unique<WebSocketMessage>(WebSocketMessageType::Close,
"", "",
wireSize, wireSize,
WebSocketErrorInfo(), WebSocketErrorInfo(),
@@ -193,7 +193,7 @@ namespace ix
return status; return status;
} }
_onMessageCallback(std::make_shared<WebSocketMessage>( _onMessageCallback(std::make_unique<WebSocketMessage>(
WebSocketMessageType::Open, WebSocketMessageType::Open,
"", "",
0, 0,
@@ -225,7 +225,7 @@ namespace ix
} }
_onMessageCallback( _onMessageCallback(
std::make_shared<WebSocketMessage>(WebSocketMessageType::Open, std::make_unique<WebSocketMessage>(WebSocketMessageType::Open,
"", "",
0, 0,
WebSocketErrorInfo(), WebSocketErrorInfo(),
@@ -310,7 +310,7 @@ namespace ix
connectErr.reason = status.errorStr; connectErr.reason = status.errorStr;
connectErr.http_status = status.http_status; connectErr.http_status = status.http_status;
_onMessageCallback(std::make_shared<WebSocketMessage>(WebSocketMessageType::Error, _onMessageCallback(std::make_unique<WebSocketMessage>(WebSocketMessageType::Error,
"", "",
0, 0,
connectErr, connectErr,
@@ -386,7 +386,7 @@ namespace ix
bool binary = messageKind == WebSocketTransport::MessageKind::MSG_BINARY; bool binary = messageKind == WebSocketTransport::MessageKind::MSG_BINARY;
_onMessageCallback(std::make_shared<WebSocketMessage>(webSocketMessageType, _onMessageCallback(std::make_unique<WebSocketMessage>(webSocketMessageType,
msg, msg,
wireSize, wireSize,
webSocketErrorInfo, webSocketErrorInfo,

2
ixwebsocket/IXWebSocketMessage.h
View File

@@ -45,5 +45,5 @@ namespace ix
} }
}; };
using WebSocketMessagePtr = std::shared_ptr<WebSocketMessage>; using WebSocketMessagePtr = std::unique_ptr<WebSocketMessage>;
} // namespace ix } // namespace ix

86
ixwebsocket/IXWebSocketMessageQueue.cpp
View File

@@ -1,86 +0,0 @@
/*
* IXWebSocketMessageQueue.cpp
* Author: Korchynskyi Dmytro
* Copyright (c) 2017-2019 Machine Zone, Inc. All rights reserved.
*/
#include "IXWebSocketMessageQueue.h"
namespace ix
{
WebSocketMessageQueue::WebSocketMessageQueue(WebSocket* websocket)
{
bindWebsocket(websocket);
}
WebSocketMessageQueue::~WebSocketMessageQueue()
{
if (!_messages.empty())
{
// not handled all messages
}
bindWebsocket(nullptr);
}
void WebSocketMessageQueue::bindWebsocket(WebSocket* websocket)
{
if (_websocket == websocket) return;
// unbind old
if (_websocket)
{
// set dummy callback just to avoid crash
_websocket->setOnMessageCallback([](const WebSocketMessagePtr&) {});
}
_websocket = websocket;
// bind new
if (_websocket)
{
_websocket->setOnMessageCallback([this](const WebSocketMessagePtr& msg) {
std::lock_guard<std::mutex> lock(_messagesMutex);
_messages.emplace_back(std::move(msg));
});
}
}
void WebSocketMessageQueue::setOnMessageCallback(const OnMessageCallback& callback)
{
_onMessageUserCallback = callback;
}
void WebSocketMessageQueue::setOnMessageCallback(OnMessageCallback&& callback)
{
_onMessageUserCallback = std::move(callback);
}
WebSocketMessagePtr WebSocketMessageQueue::popMessage()
{
WebSocketMessagePtr message;
std::lock_guard<std::mutex> lock(_messagesMutex);
if (!_messages.empty())
{
message = std::move(_messages.front());
_messages.pop_front();
}
return message;
}
void WebSocketMessageQueue::poll(int count)
{
if (!_onMessageUserCallback) return;
WebSocketMessagePtr message;
while (count > 0 && (message = popMessage()))
{
_onMessageUserCallback(message);
--count;
}
}
} // namespace ix

41
ixwebsocket/IXWebSocketMessageQueue.h
View File

@@ -1,41 +0,0 @@
/*
* IXWebSocketMessageQueue.h
* Author: Korchynskyi Dmytro
* Copyright (c) 2017-2019 Machine Zone, Inc. All rights reserved.
*/
#pragma once
#include "IXWebSocket.h"
#include <list>
#include <memory>
#include <thread>
namespace ix
{
//
// A helper class to dispatch websocket message callbacks in your thread.
//
class WebSocketMessageQueue
{
public:
WebSocketMessageQueue(WebSocket* websocket = nullptr);
~WebSocketMessageQueue();
void bindWebsocket(WebSocket* websocket);
void setOnMessageCallback(const OnMessageCallback& callback);
void setOnMessageCallback(OnMessageCallback&& callback);
void poll(int count = 512);
protected:
WebSocketMessagePtr popMessage();
private:
WebSocket* _websocket = nullptr;
OnMessageCallback _onMessageUserCallback;
std::mutex _messagesMutex;
std::list<WebSocketMessagePtr> _messages;
};
} // namespace ix

2
ixwebsocket/IXWebSocketVersion.h
View File

@@ -6,4 +6,4 @@
#pragma once #pragma once
#define IX_WEBSOCKET_VERSION "9.2.4" #define IX_WEBSOCKET_VERSION "9.2.5"

178
test/IXWebSocketMessageQTest.cpp
View File

@@ -1,178 +0,0 @@
/*
* IXWebSocketMessageQTest.cpp
* Author: Korchynskyi Dmytro
* Copyright (c) 2019 Machine Zone. All rights reserved.
*/
#include "IXTest.h"
#include "catch.hpp"
#include <ixwebsocket/IXWebSocket.h>
#include <ixwebsocket/IXWebSocketMessageQueue.h>
#include <ixwebsocket/IXWebSocketServer.h>
#include <thread>
using namespace ix;
namespace
{
bool startServer(ix::WebSocketServer& server)
{
server.setOnConnectionCallback([&server](std::shared_ptr<ix::WebSocket> webSocket,
std::shared_ptr<ConnectionState> connectionState) {
webSocket->setOnMessageCallback(
[connectionState, &server](const WebSocketMessagePtr& msg) {
if (msg->type == ix::WebSocketMessageType::Open)
{
TLogger() << "New connection";
connectionState->computeId();
TLogger() << "id: " << connectionState->getId();
TLogger() << "Uri: " << msg->openInfo.uri;
TLogger() << "Headers:";
for (auto&& it : msg->openInfo.headers)
{
TLogger() << it.first << ": " << it.second;
}
}
else if (msg->type == ix::WebSocketMessageType::Close)
{
TLogger() << "Closed connection";
}
else if (msg->type == ix::WebSocketMessageType::Message)
{
TLogger() << "Message received: " << msg->str;
for (auto&& client : server.getClients())
{
client->send(msg->str);
}
}
});
});
auto res = server.listen();
if (!res.first)
{
TLogger() << res.second;
return false;
}
server.start();
return true;
}
class MsgQTestClient
{
public:
MsgQTestClient()
{
msgQ.bindWebsocket(&ws);
msgQ.setOnMessageCallback([this](const WebSocketMessagePtr& msg) {
REQUIRE(mainThreadId == std::this_thread::get_id());
std::stringstream ss;
if (msg->type == WebSocketMessageType::Open)
{
log("client connected");
sendNextMessage();
}
else if (msg->type == WebSocketMessageType::Close)
{
log("client disconnected");
}
else if (msg->type == WebSocketMessageType::Error)
{
ss << "Error ! " << msg->errorInfo.reason;
log(ss.str());
testDone = true;
}
else if (msg->type == WebSocketMessageType::Pong)
{
ss << "Received pong message " << msg->str;
log(ss.str());
}
else if (msg->type == WebSocketMessageType::Ping)
{
ss << "Received ping message " << msg->str;
log(ss.str());
}
else if (msg->type == WebSocketMessageType::Message)
{
REQUIRE(msg->str.compare("Hey dude!") == 0);
++receivedCount;
ss << "Received message " << msg->str;
log(ss.str());
sendNextMessage();
}
else
{
ss << "Invalid WebSocketMessageType";
log(ss.str());
testDone = true;
}
});
}
void sendNextMessage()
{
if (receivedCount >= 3)
{
testDone = true;
succeeded = true;
}
else
{
auto info = ws.sendText("Hey dude!");
if (info.success)
log("sent message");
else
log("send failed");
}
}
void run(const std::string& url)
{
mainThreadId = std::this_thread::get_id();
testDone = false;
receivedCount = 0;
ws.setUrl(url);
ws.start();
while (!testDone)
{
msgQ.poll();
msleep(50);
}
}
bool isSucceeded() const
{
return succeeded;
}
private:
WebSocket ws;
WebSocketMessageQueue msgQ;
bool testDone = false;
uint32_t receivedCount = 0;
std::thread::id mainThreadId;
bool succeeded = false;
};
} // namespace
TEST_CASE("Websocket_message_queue", "[websocket_message_q]")
{
SECTION("Send several messages")
{
int port = getFreePort();
WebSocketServer server(port);
REQUIRE(startServer(server));
MsgQTestClient testClient;
testClient.run("ws://127.0.0.1:" + std::to_string(port));
REQUIRE(testClient.isSucceeded());
server.stop();
}
}