copilot/infinipaint
1
0
Fork 0
Code Issues Pull requests Projects Releases 1 Packages Wiki Activity Actions

Added multi command block

Browse source
This commit is contained in:
Yousef Khadadeh 2026-06-02 21:57:53 +03:00
parent 6889eeff60
commit 279a9317a4
14 changed files with 173 additions and 106 deletions
Download patch file Download diff file Expand all files Collapse all files

63
include/Helpers/Networking/NetClient.cpp
View file

@ -21,6 +21,8 @@
#include "NetServer.hpp"
#include <chrono>
#include <Helpers/Logger.hpp>
#include <cereal/types/vector.hpp>
#include <cereal/types/string.hpp>
NetClient::NetClient(const std::string& serverFullID) {
receiveQueue = std::make_shared<ReceiveQueue>();
@ -111,9 +113,14 @@ void NetClient::parse_received_messages() {
decode_fragmented_message(inArchive, spfm, NetLibrary::FRAGMENT_MESSAGE_STRIDE, [&](cereal::PortableBinaryInputArchive& completeArchive) {
MessageCommandType completeCommandID;
completeArchive(completeCommandID);
recvCallbacks[completeCommandID](completeArchive);
if(completeCommandID == 1)
parse_multi_command_id(completeArchive);
else
recvCallbacks[completeCommandID](completeArchive);
});
}
else if(commandID == 1)
parse_multi_command_id(inArchive);
else
recvCallbacks[commandID](inArchive);
@ -121,23 +128,39 @@ void NetClient::parse_received_messages() {
}
}
void NetClient::parse_multi_command_id(cereal::PortableBinaryInputArchive& a) {
std::vector<std::string> commands;
a(commands);
for(const std::string& str : commands) {
ByteMemStream strm((char*)str.data(), str.size());
cereal::PortableBinaryInputArchive inArchive(strm);
MessageCommandType commandID;
inArchive(commandID);
recvCallbacks[commandID](inArchive);
}
}
void NetClient::send_string_stream_to_server(const std::string& channel, const std::shared_ptr<std::stringstream>& ss) {
if(isDisconnected)
return;
auto& messageQueue = messageQueues[channel];
if(channel == UNRELIABLE_COMMAND_CHANNEL) {
if(ss->view().length() <= NetLibrary::MAX_UNRELIABLE_MESSAGE_SIZE) // Drop unreliable messages that are too big
messageQueue.emplace(NetLibrary::calc_order_for_queued_message(channel, nextMessageOrderToSend), ss);
}
if(!multiCommandData.expired() && channel == multiCommandData.lock()->channelToSendTo)
multiCommandData.lock()->commands.emplace_back(ss->str());
else {
std::vector<std::shared_ptr<std::stringstream>> fragmentedMessage = fragment_message(ss->view(), NetLibrary::FRAGMENT_MESSAGE_STRIDE);
if(fragmentedMessage.empty())
messageQueue.emplace(NetLibrary::calc_order_for_queued_message(channel, nextMessageOrderToSend), ss);
auto& messageQueue = messageQueues[channel];
if(channel == UNRELIABLE_COMMAND_CHANNEL) {
if(ss->view().length() <= NetLibrary::MAX_UNRELIABLE_MESSAGE_SIZE) // Drop unreliable messages that are too big
messageQueue.emplace(NetLibrary::calc_order_for_queued_message(channel, nextMessageOrderToSend), ss);
}
else {
for(auto& ss2 : fragmentedMessage)
messageQueue.emplace(NetLibrary::calc_order_for_queued_message(channel, nextMessageOrderToSend), ss2);
std::vector<std::shared_ptr<std::stringstream>> fragmentedMessage = fragment_message(ss->view(), NetLibrary::FRAGMENT_MESSAGE_STRIDE);
if(fragmentedMessage.empty())
messageQueue.emplace(NetLibrary::calc_order_for_queued_message(channel, nextMessageOrderToSend), ss);
else {
for(auto& ss2 : fragmentedMessage)
messageQueue.emplace(NetLibrary::calc_order_for_queued_message(channel, nextMessageOrderToSend), ss2);
}
}
}
}
@ -194,6 +217,22 @@ void NetClient::send_queued_messages() {
}
}
void NetClient::send_multi_command_to_server(const std::string& channel, const std::function<void()>& captureSendBlock) {
if(!is_disconnected() && multiCommandData.expired() && !channel.empty()) { // Check for multiUpdateData.expired() to make sure we aren't overriding a previous call to send_multi_update_messsage
std::shared_ptr<MultiCommandData> d = std::make_shared<MultiCommandData>();
d->channelToSendTo = channel;
multiCommandData = d; // Using weak pointer here ensures that object will be freed
captureSendBlock();
if(!d->commands.empty()) {
std::vector<std::string> movedVec = std::move(d->commands);
d = nullptr;
send_items_to_server(channel, (MessageCommandType)1, movedVec);
}
}
else
captureSendBlock();
}
bool NetClient::is_disconnected() const {
return isDisconnected;
}

8
include/Helpers/Networking/NetClient.hpp
View file

@ -46,6 +46,7 @@ class NetClient : public std::enable_shared_from_this<NetClient> {
send_string_stream_to_server(channel, ss);
}
void send_string_stream_to_server(const std::string& channel, const std::shared_ptr<std::stringstream>& ss);
void send_multi_command_to_server(const std::string& channel, const std::function<void()>& captureSendBlock);
void update();
void add_recv_callback(MessageCommandType commandID, const NetClientRecvCallback& callback);
bool is_disconnected() const;
@ -58,6 +59,7 @@ class NetClient : public std::enable_shared_from_this<NetClient> {
void parse_received_messages();
void send_queued_messages();
void send_str_as_bytes(std::shared_ptr<rtc::DataChannel> channel, const std::string& str);
void parse_multi_command_id(cereal::PortableBinaryInputArchive& inArchive);
struct OutgoingMessage {
MessageOrder order;
@ -90,6 +92,12 @@ class NetClient : public std::enable_shared_from_this<NetClient> {
std::string serverFullID;
bool fullyRegistered = false;
struct MultiCommandData {
std::string channelToSendTo;
std::vector<std::string> commands;
};
std::weak_ptr<MultiCommandData> multiCommandData;
std::chrono::steady_clock::time_point lastMessageTime;
friend class NetLibrary;

69
include/Helpers/Networking/NetServer.cpp
View file

@ -24,7 +24,9 @@
#include <Helpers/Random.hpp>
#include <Helpers/Logger.hpp>
#include "NetClient.hpp"
#include "cereal/archives/portable_binary.hpp"
#include <cereal/archives/portable_binary.hpp>
#include <cereal/types/vector.hpp>
#include <cereal/types/string.hpp>
NetServer::NetServer(const std::string& serverLocalID) {
localID = serverLocalID;
@ -124,6 +126,9 @@ void NetServer::client_connected(std::shared_ptr<rtc::PeerConnection> connection
}
void NetServer::send_string_stream_to_client(const std::shared_ptr<ClientData>& client, const std::string& channel, const std::shared_ptr<std::stringstream>& ss) {
if(!multiCommandData.expired() && channel == multiCommandData.lock()->channelToSendTo)
throw std::runtime_error("[NetServer::send_string_stream_to_client] Can't call when in send_multi_command block using same channel");
if(!client)
return;
@ -144,18 +149,33 @@ void NetServer::send_string_stream_to_client(const std::shared_ptr<ClientData>&
}
void NetServer::send_string_stream_to_all_clients(const std::string& channel, const std::shared_ptr<std::stringstream>& ss) {
send_string_stream_to_client_if([&](const std::shared_ptr<ClientData>& c) {
return true;
}, channel, ss);
if(!multiCommandData.expired() && channel == multiCommandData.lock()->channelToSendTo)
multiCommandData.lock()->commands.emplace_back(ss->str());
else {
send_string_stream_to_client_if([&](const std::shared_ptr<ClientData>& c) {
return true;
}, channel, ss);
}
}
void NetServer::send_string_stream_to_all_clients_except(const std::shared_ptr<ClientData>& client, const std::string& channel, const std::shared_ptr<std::stringstream>& ss) {
send_string_stream_to_client_if([&](const std::shared_ptr<ClientData>& c) {
return c != client;
}, channel, ss);
if(!multiCommandData.expired() && channel == multiCommandData.lock()->channelToSendTo) {
if(!client)
multiCommandData.lock()->commands.emplace_back(ss->str());
else
throw std::runtime_error("[NetServer::send_string_stream_to_all_clients_except] Can't call when in send_multi_command block using same channel");
}
else {
send_string_stream_to_client_if([&](const std::shared_ptr<ClientData>& c) {
return c != client;
}, channel, ss);
}
}
void NetServer::send_string_stream_to_client_if(std::function<bool(const std::shared_ptr<ClientData>&)> clientChecker, const std::string& channel, const std::shared_ptr<std::stringstream>& ss) {
if(!multiCommandData.expired() && channel == multiCommandData.lock()->channelToSendTo)
throw std::runtime_error("[NetServer::send_string_stream_to_client_if] Can't call when in send_multi_command block using same channel");
if(channel == UNRELIABLE_COMMAND_CHANNEL) {
if(ss->view().length() <= NetLibrary::MAX_UNRELIABLE_MESSAGE_SIZE) { // Drop unreliable messages that are too big
for(auto& client : clients) {
@ -183,6 +203,22 @@ void NetServer::send_string_stream_to_client_if(std::function<bool(const std::sh
}
}
void NetServer::send_multi_command_to_all_clients(const std::string& channel, const std::function<void()>& captureSendBlock) {
if(!is_disconnected() && multiCommandData.expired() && !channel.empty()) { // Check for multiUpdateData.expired() to make sure we aren't overriding a previous call to send_multi_update_messsage
std::shared_ptr<MultiCommandData> d = std::make_shared<MultiCommandData>();
d->channelToSendTo = channel;
multiCommandData = d; // Using weak pointer here ensures that object will be freed
captureSendBlock();
if(!d->commands.empty()) {
std::vector<std::string> movedVec = std::move(d->commands);
d = nullptr;
send_items_to_all_clients(channel, (MessageCommandType)1, movedVec);
}
}
else
captureSendBlock();
}
void NetServer::ClientData::send_queued_messages(NetServer& server) {
for(auto& [channelName, messageQueue] : messageQueues) {
bool addMessageOrder = NetLibrary::is_ordered_channel(channelName);
@ -258,9 +294,14 @@ void NetServer::ClientData::parse_received_messages(NetServer& server) {
decode_fragmented_message(inArchive, spfm, NetLibrary::FRAGMENT_MESSAGE_STRIDE, [&](cereal::PortableBinaryInputArchive& completeArchive) {
MessageCommandType completeCommandID;
completeArchive(completeCommandID);
server.recvCallbacks[completeCommandID](shared_from_this(), completeArchive);
if(completeCommandID == 1)
parse_multi_command_id(server, completeArchive);
else
server.recvCallbacks[completeCommandID](shared_from_this(), completeArchive);
});
}
else if(commandID == 1)
parse_multi_command_id(server, inArchive);
else
server.recvCallbacks[commandID](shared_from_this(), inArchive);
@ -268,6 +309,18 @@ void NetServer::ClientData::parse_received_messages(NetServer& server) {
}
}
void NetServer::ClientData::parse_multi_command_id(NetServer& server, cereal::PortableBinaryInputArchive& a) {
std::vector<std::string> commands;
a(commands);
for(const std::string& str : commands) {
ByteMemStream strm((char*)str.data(), str.size());
cereal::PortableBinaryInputArchive inArchive(strm);
MessageCommandType commandID;
inArchive(commandID);
server.recvCallbacks[commandID](shared_from_this(), inArchive);
}
}
NetLibrary::DownloadProgress NetServer::ClientData::get_progress_into_fragmented_message(const std::string& channel) const {
auto it = pfm.find(channel);
if(it == pfm.end())

8
include/Helpers/Networking/NetServer.hpp
View file

@ -60,6 +60,7 @@ class NetServer : public std::enable_shared_from_this<NetServer> {
void send_queued_messages(NetServer& server);
void parse_received_messages(NetServer& server);
void parse_multi_command_id(NetServer& server, cereal::PortableBinaryInputArchive& a);
void send_str_as_bytes(std::shared_ptr<rtc::DataChannel> channel, const std::string& str);
NetLibrary::DownloadProgress get_progress_into_fragmented_message(const std::string& channel) const;
@ -98,6 +99,7 @@ class NetServer : public std::enable_shared_from_this<NetServer> {
void send_string_stream_to_all_clients(const std::string& channel, const std::shared_ptr<std::stringstream>& ss);
void send_string_stream_to_all_clients_except(const std::shared_ptr<ClientData>& client, const std::string& channel, const std::shared_ptr<std::stringstream>& ss);
void send_string_stream_to_client_if(std::function<bool(const std::shared_ptr<ClientData>&)> clientChecker, const std::string& channel, const std::shared_ptr<std::stringstream>& ss);
void send_multi_command_to_all_clients(const std::string& channel, const std::function<void()>& captureSendBlock);
bool is_disconnected() const;
@ -127,6 +129,12 @@ class NetServer : public std::enable_shared_from_this<NetServer> {
NetClient* directConnectClient = nullptr;
std::shared_ptr<ClientData> directConnectClientData;
struct MultiCommandData {
std::string channelToSendTo;
std::vector<std::string> commands;
};
std::weak_ptr<MultiCommandData> multiCommandData;
friend class NetLibrary;
friend class NetClient;
};