diff --git a/NetworkingLib/include/DatagramReceiver.h b/NetworkingLib/include/DatagramReceiver.h
index 43c79d9da187a2d8494f5ab0763ce828ef6fa1c2..634e1c6cf8d870897ddbad1f511b62266bc1259f 100644
--- a/NetworkingLib/include/DatagramReceiver.h
+++ b/NetworkingLib/include/DatagramReceiver.h
@@ -47,7 +47,7 @@ public:
: net(net)
, bindingPort(bindingPort)
, socket(net.getIoService(), Udp::v4())
- , buffer(maxMessageSize)
+ , buffer(maxMessageSize + internal::Frame::HEADER_SIZE)
{}
Message receive(std::string & host, std::uint16_t & port, const time::Duration & timeout)
diff --git a/NetworkingLib/include/DatagramSender.h b/NetworkingLib/include/DatagramSender.h
index 942e1f35e1203c969ec663969be884e4360fd4cd..8a6012eac3550a96ee8386ac7e9e6b56319c372b 100644
--- a/NetworkingLib/include/DatagramSender.h
+++ b/NetworkingLib/include/DatagramSender.h
@@ -28,9 +28,7 @@ private:
public:
using Ptr = std::shared_ptr<DatagramSender>;
- using SendHandler = std::function<void( const
- error::ErrorCode & error
- )>;
+ using SendHandler = std::function<void(const error::ErrorCode & error)>;
using Udp = boost::asio::ip::udp;
using Socket = Udp::socket;
diff --git a/NetworkingLib/include/Error.h b/NetworkingLib/include/Error.h
index c3f5f4a354a58a494c416bc6c164ecebd60de0a3..062d175252219bf93cefb6b4f2f8382f1d4628b5 100644
--- a/NetworkingLib/include/Error.h
+++ b/NetworkingLib/include/Error.h
@@ -38,7 +38,7 @@ protected:
ErrorCode code;
};
-class FailedOperation : Error
+class FailedOperation : public Error
{
public:
FailedOperation()
@@ -46,7 +46,7 @@ public:
{}
};
-class Aborted : Error
+class Aborted : public Error
{
public:
Aborted()
@@ -54,7 +54,7 @@ public:
{}
};
-class Busy : Error
+class Busy : public Error
{
public:
Busy()
diff --git a/NetworkingLib/include/Frame.h b/NetworkingLib/include/Frame.h
index 18bca9d2cca36f146d5ce8257a3833dd96072565..26c840b19cab98525e0c03bc4448d08da214aa9d 100644
--- a/NetworkingLib/include/Frame.h
+++ b/NetworkingLib/include/Frame.h
@@ -17,6 +17,8 @@ namespace internal
class Frame
{
public:
+ static constexpr std::size_t HEADER_SIZE = 4;
+
Frame(const std::uint8_t * data, std::uint32_t numDataBytes)
: data(data), numDataBytes(numDataBytes)
{
diff --git a/NetworkingLib/include/Message.h b/NetworkingLib/include/Message.h
index 2887fd628cfab2dfd07bebc2356b633c3df53f1a..90fa9826b58b9a35b8d7eae87d4590c33b0c28ba 100644
--- a/NetworkingLib/include/Message.h
+++ b/NetworkingLib/include/Message.h
@@ -72,7 +72,7 @@ void asyncSend(Networking & net,
boost::asio::ip::tcp::socket & socket,
const Message & message,
const time::Duration & timeout,
- SendHandler handler)
+ const SendHandler & handler)
{
auto data = std::make_shared<std::string>(Encoder<Message>{}(message));
networking::stream::asyncWrite(
@@ -100,7 +100,7 @@ void asyncReceive(Networking & net,
boost::asio::ip::tcp::socket & socket,
boost::asio::streambuf & buffer,
const time::Duration & timeout,
- ReceiveHandler<Message> handler)
+ const ReceiveHandler<Message> & handler)
{
networking::stream::asyncRead(
net, socket, buffer, timeout,
@@ -129,7 +129,7 @@ void asyncSendDatagramTo(Networking & net,
const std::string & host,
std::uint16_t port,
const time::Duration & timeout,
- SendToHandler handler)
+ const SendToHandler & handler)
{
auto data = std::make_shared<std::string>(Encoder<Message>{}(message));
networking::socket::asyncSendTo(
@@ -155,7 +155,7 @@ void asyncReceiveDatagramFrom(Networking & net,
boost::asio::ip::udp::socket & socket,
std::vector<char> & buffer,
const time::Duration & timeout,
- ReceiveFromHandler<Message> handler)
+ const ReceiveFromHandler<Message> & handler)
{
networking::socket::asyncReceiveFrom(
net, socket, buffer, timeout,
diff --git a/NetworkingLib/include/ServiceClient.h b/NetworkingLib/include/ServiceClient.h
index 8bb8b9538e0931d637eb6da782b9b35c9793a94d..9d8fb641a8e3e119e5425b93ff667d51f0a531a2 100644
--- a/NetworkingLib/include/ServiceClient.h
+++ b/NetworkingLib/include/ServiceClient.h
@@ -41,14 +41,15 @@ public:
using CallHandler = std::function<void(const error::ErrorCode & error, ResponseMessage & response)>;
- static Ptr create(Networking & net)
+ static Ptr create(Networking & net, std::size_t maxMessageSize = 512)
{
- return std::make_shared<Client<Service>>(PrivateTag{}, net);
+ return std::make_shared<Client<Service>>(PrivateTag{}, net, maxMessageSize);
}
- Client(PrivateTag, Networking & net)
+ Client(PrivateTag, Networking & net, std::size_t maxMessageSize = 512)
: net(net)
, socket(net.getIoService())
+ , maxMessageSize(maxMessageSize)
{}
ResponseMessage call(const std::string & host,
@@ -71,7 +72,7 @@ public:
message::send(net, socket, request, timeout);
updateTimeout(timeout, startTime);
// Receive the response.
- boost::asio::streambuf buffer;
+ boost::asio::streambuf buffer{maxMessageSize + internal::Frame::HEADER_SIZE};
return message::receive<ResponseMessage>(net, socket, buffer, timeout);
}
@@ -120,8 +121,7 @@ public:
// Receive the response.
networking::message::asyncReceive<ResponseMessage>(
state->self->net, state->self->socket, state->buffer, state->timeout,
- [state](auto const & error,
- auto & response)
+ [state](auto const & error, auto & response)
{
state->handler(error, response);
});
@@ -142,6 +142,7 @@ public:
private:
networking::Networking & net;
Socket socket;
+ const std::size_t maxMessageSize;
static void updateTimeout(time::Duration & timeout, time::TimePoint & startTime)
{
@@ -169,6 +170,7 @@ private:
, handler(handler)
, timeout(timeout)
, startTime(startTime)
+ , buffer(self->maxMessageSize + internal::Frame::HEADER_SIZE)
, closer(self->socket)
{}
@@ -178,7 +180,7 @@ private:
time::Duration timeout;
time::TimePoint startTime;
boost::asio::streambuf buffer;
- closeable::Closer<Socket> closer;
+ closeable::Closer <Socket> closer;
};
};
diff --git a/NetworkingLib/include/ServiceServer.h b/NetworkingLib/include/ServiceServer.h
index 9ac34e469236a26d0fe2f181f02ac8f3f4fdcde1..c11d8cb7cb4e3ce1cb93185d885533093c50c52d 100644
--- a/NetworkingLib/include/ServiceServer.h
+++ b/NetworkingLib/include/ServiceServer.h
@@ -37,18 +37,20 @@ public:
std::function<ResponseMessage(const Endpoint & clientEndpoint,
RequestMessage & requestMessage)>;
- static Ptr create(Networking & net, uint16_t bindingPort)
+ static Ptr create(Networking & net, uint16_t bindingPort, std::size_t maxMessageSize = 512)
{
- return std::make_shared<Server<Service>>(PrivateTag{}, net, bindingPort);
+ return std::make_shared<Server<Service>>(PrivateTag{}, net, bindingPort, maxMessageSize);
}
// Should not be used outside.
Server(PrivateTag,
Networking & net,
- uint16_t bindingPort)
+ uint16_t bindingPort,
+ std::size_t maxMessageSize)
: net(net)
, bindingPort(bindingPort)
, acceptor(net.getIoService())
+ , maxMessageSize(maxMessageSize)
{}
void advertiseService(const RequestReceivedHandler & requestReceivedHandler)
@@ -72,6 +74,7 @@ private:
std::uint16_t bindingPort;
Acceptor acceptor;
std::atomic<bool> running{false};
+ const std::size_t maxMessageSize;
void accept(RequestReceivedHandler handler)
{
@@ -119,13 +122,13 @@ private:
});
}
-private:
struct AsyncState
{
AsyncState(Ptr self, const RequestReceivedHandler & requestReceivedHandler)
: self(self)
, socket(self->net.getIoService())
, requestReceivedHandler(requestReceivedHandler)
+ , buffer(self->maxMessageSize + internal::Frame::HEADER_SIZE)
{}
Ptr self;
diff --git a/NetworkingLib/include/Socket.h b/NetworkingLib/include/Socket.h
index c61e01821f180377cb76ada52700836e9580297e..eb7e172e24603a7368cb7d7f24f935f05466444b 100644
--- a/NetworkingLib/include/Socket.h
+++ b/NetworkingLib/include/Socket.h
@@ -128,17 +128,17 @@ void sendTo(Networking & net,
udp::endpoint endpoint{address::from_string(host), port};
using namespace networking::internal;
- Frame buffer{(const std::uint8_t *) sendData.c_str(), sendData.size()};
+ Frame frame{(const std::uint8_t *) sendData.c_str(), sendData.size()};
auto asyncOperation = [&socket](auto && ... args)
{ socket.async_send_to(std::forward<decltype(args)>(args)...); };
std::tuple<boost::system::error_code, std::size_t> result;
closeable::timedOperation(
- result, net, asyncOperation, socket, timeout, buffer.getBuffers(), endpoint);
+ result, net, asyncOperation, socket, timeout, frame.getBuffers(), endpoint);
auto numBytesTransferred = std::get<1>(result);
- if (numBytesTransferred < buffer.getSize())
+ if (numBytesTransferred < frame.getSize())
throw error::FailedOperation{};
};
@@ -155,18 +155,18 @@ void asyncSendTo(Networking & net,
udp::endpoint endpoint{address::from_string(host), port};
using namespace networking::internal;
- auto buffer = std::make_shared<Frame>((const std::uint8_t *) sendData.c_str(), sendData.size());
+ auto frame = std::make_shared<Frame>((const std::uint8_t *) sendData.c_str(), sendData.size());
auto asyncOperation = [&socket](auto && ... args)
{ socket.async_send_to(std::forward<decltype(args)>(args)...); };
closeable::timedAsyncOperation(
net, asyncOperation, socket, timeout,
- [handler, buffer](const auto & networkingError,
+ [handler, frame](const auto & networkingError,
const auto & boostError,
auto numBytesTransferred)
{
- if (numBytesTransferred < buffer->getSize())
+ if (numBytesTransferred < frame->getSize())
{
handler(error::codes::FAILED_OPERATION);
return;
@@ -174,7 +174,7 @@ void asyncSendTo(Networking & net,
handler(networkingError);
},
- buffer->getBuffers(), endpoint);
+ frame->getBuffers(), endpoint);
};
template<typename DatagramSocket>
diff --git a/NetworkingLib/include/Stream.h b/NetworkingLib/include/Stream.h
index bf8a61df8a286cbf62154498bd7a1e428c72efbc..2088208bf029e8ab1f335a6c169c0cc65892735d 100644
--- a/NetworkingLib/include/Stream.h
+++ b/NetworkingLib/include/Stream.h
@@ -24,10 +24,9 @@ namespace internal
std::string stringFromStreambuf(boost::asio::streambuf & streambuf, std::size_t numBytes);
-void numDataBytesFromBuffer(boost::asio::streambuf & streambuf,
- std::size_t numBytesTransferred,
- std::uint32_t & numDataBytes,
- std::uint32_t & numRemainingBytes);
+std::uint32_t numDataBytesFromBuffer(boost::asio::streambuf & streambuf);
+
+std::size_t getNumRemainingBytes(std::size_t numBytesTransferred, std::uint32_t numDataBytes);
}
@@ -42,16 +41,16 @@ void write(Networking & net,
const time::Duration & timeout)
{
using namespace networking::internal;
- Frame buffer{(const std::uint8_t *) writeData.c_str(), writeData.size()};
+ Frame frame{(const std::uint8_t *) writeData.c_str(), writeData.size()};
auto asyncOperation = [](auto && ... args)
{ boost::asio::async_write(std::forward<decltype(args)>(args)...); };
std::tuple<boost::system::error_code, std::size_t> result;
- closeable::timedOperation(result, net, asyncOperation, stream, timeout, stream, buffer.getBuffers());
+ closeable::timedOperation(result, net, asyncOperation, stream, timeout, stream, frame.getBuffers());
auto numBytesTransferred = std::get<1>(result);
- if (numBytesTransferred < buffer.getSize())
+ if (numBytesTransferred < frame.getSize())
throw error::FailedOperation{};
};
@@ -63,16 +62,16 @@ void asyncWrite(Networking & net,
const WriteHandler & handler)
{
using namespace networking::internal;
- auto buffer = std::make_shared<Frame>((const std::uint8_t *) writeData.c_str(), writeData.size());
+ auto frame = std::make_shared<Frame>((const std::uint8_t *) writeData.c_str(), writeData.size());
auto asyncOperation = [](auto && ... args)
{ boost::asio::async_write(std::forward<decltype(args)>(args)...); };
closeable::timedAsyncOperation(
net, asyncOperation, stream, timeout,
- [handler, buffer](const auto & networkingError, const auto & boostError, auto numBytesTransferred)
+ [handler, frame](const auto & networkingError, const auto & boostError, auto numBytesTransferred)
{
- if (numBytesTransferred < buffer->getSize())
+ if (numBytesTransferred < frame->getSize())
{
handler(error::codes::FAILED_OPERATION);
return;
@@ -80,7 +79,7 @@ void asyncWrite(Networking & net,
handler(networkingError);
},
- stream, buffer->getBuffers());
+ stream, frame->getBuffers());
}
template<typename SyncReadStream>
@@ -89,7 +88,8 @@ std::string read(Networking & net,
boost::asio::streambuf & buffer,
time::Duration timeout)
{
- using namespace internal;
+ using networking::internal::Frame;
+ using namespace networking::stream::internal;
auto startTime = time::now();
@@ -98,21 +98,30 @@ std::string read(Networking & net,
std::tuple<boost::system::error_code, std::size_t> result;
- // We receive 4 bytes first, giving us the number of data bytes.
+ // Receive frame header.
closeable::timedOperation(
- result, net, asyncOperation, stream, timeout, stream, buffer, boost::asio::transfer_at_least(4));
+ result, net, asyncOperation, stream, timeout, stream, buffer, boost::asio::transfer_at_least(Frame::HEADER_SIZE));
auto numBytesTransferred = std::get<1>(result);
- std::uint32_t numDataBytes, numRemainingBytes;
- numDataBytesFromBuffer(buffer, numBytesTransferred, numDataBytes, numRemainingBytes);
- if (numRemainingBytes == 0)
+ if (numBytesTransferred < Frame::HEADER_SIZE)
+ throw error::FailedOperation{};
+
+ auto numDataBytes = numDataBytesFromBuffer(buffer);
+ if (numDataBytes == 0)
+ return std::string{};
+
+ if (getNumRemainingBytes(numBytesTransferred, numDataBytes) == 0)
return stringFromStreambuf(buffer, numDataBytes);
timeout -= time::now() - startTime;
- // Receive the actual data.
+ // Receive actual data.
closeable::timedOperation(
- result, net, asyncOperation, stream, timeout, stream, buffer, boost::asio::transfer_at_least(numRemainingBytes));
+ result, net, asyncOperation, stream, timeout, stream, buffer, boost::asio::transfer_at_least(numDataBytes));
+
+ numBytesTransferred = std::get<1>(result);
+ if (numBytesTransferred < numDataBytes)
+ throw error::FailedOperation{};
return stringFromStreambuf(buffer, numDataBytes);
};
@@ -124,47 +133,70 @@ void asyncRead(Networking & net,
const time::Duration & timeout,
const ReadHandler & handler)
{
- using namespace internal;
+ using networking::internal::Frame;
+ using namespace networking::stream::internal;
auto startTime = time::now();
auto asyncOperation = [](auto && ... args)
{ boost::asio::async_read(std::forward<decltype(args)>(args)...); };
+ // Receive frame header.
closeable::timedAsyncOperation(
net, asyncOperation, stream, timeout,
[&net, &stream, &buffer, timeout, handler, startTime, asyncOperation]
(const auto & networkingError, const auto & boostError, auto numBytesTransferred)
{
+ std::string data{};
+
if (networkingError)
{
- std::string noData{};
- handler(networkingError, noData);
+ handler(networkingError, data);
return;
}
- std::uint32_t numDataBytes, numRemainingBytes;
- numDataBytesFromBuffer(buffer, numBytesTransferred, numDataBytes, numRemainingBytes);
- if (numRemainingBytes == 0)
+ if (numBytesTransferred < Frame::HEADER_SIZE)
{
- auto receiveData = stringFromStreambuf(buffer, numDataBytes);
- handler(networkingError, receiveData);
+ handler(error::codes::FAILED_OPERATION, data);
+ return;
+ }
+
+ auto numDataBytes = numDataBytesFromBuffer(buffer);
+ if (numDataBytes == 0)
+ {
+ handler(networkingError, data);
+ return;
+ }
+
+ if (getNumRemainingBytes(numBytesTransferred, numDataBytes) == 0)
+ {
+ data = stringFromStreambuf(buffer, numDataBytes);
+ handler(networkingError, data);
return;
}
auto timeSpend = time::now() - startTime;
auto newTimeout = timeout - timeSpend;
+ // Receive actual data.
closeable::timedAsyncOperation(
net, asyncOperation, stream, newTimeout,
[&buffer, handler, numDataBytes](const auto & networkingError, const auto & boostError, auto numBytesTransferred)
{
- auto receiveData = stringFromStreambuf(buffer, numDataBytes);
- handler(networkingError, receiveData);
+ std::string data{};
+
+ if (numBytesTransferred < numDataBytes)
+ {
+ handler(error::codes::FAILED_OPERATION, data);
+ return;
+ }
+
+ data = stringFromStreambuf(buffer, numDataBytes);
+ handler(networkingError, data);
},
- stream, buffer, boost::asio::transfer_at_least(numRemainingBytes));
+ stream, buffer, boost::asio::transfer_at_least(numDataBytes));
},
- stream, buffer, boost::asio::transfer_at_least(4));
+ stream, buffer, boost::asio::transfer_at_least(Frame::HEADER_SIZE));
};
}
diff --git a/NetworkingLib/src/Stream.cpp b/NetworkingLib/src/Stream.cpp
index 01b769bfe7ec1fafdc520cfc86f1c5e0ef569782..f4be80af697a91ace2a95f9df3f1375e4f04a12c 100644
--- a/NetworkingLib/src/Stream.cpp
+++ b/NetworkingLib/src/Stream.cpp
@@ -20,17 +20,19 @@ std::string stringFromStreambuf(boost::asio::streambuf & streambuf, std::size_t
return data;
}
-void numDataBytesFromBuffer(boost::asio::streambuf & streambuf,
- std::size_t numBytesTransferred,
- std::uint32_t & numDataBytes,
- std::uint32_t & numRemainingBytes)
+std::uint32_t numDataBytesFromBuffer(boost::asio::streambuf & streambuf)
{
auto numDataBytesStr = stringFromStreambuf(streambuf, 4);
- numDataBytes = utils::bytesToInt<4, std::uint32_t>((const std::uint8_t *) numDataBytesStr.c_str());
+ return utils::bytesToInt<4, std::uint32_t>((const std::uint8_t *) numDataBytesStr.c_str());
+}
- numRemainingBytes = 0;
- if (numDataBytes > numBytesTransferred - 4)
- numRemainingBytes = numDataBytes - numBytesTransferred - 4;
+std::size_t getNumRemainingBytes(std::size_t numBytesTransferred, std::uint32_t numDataBytes)
+{
+ using networking::internal::Frame;
+ std::size_t numRemainingBytes = 0;
+ if (numDataBytes > numBytesTransferred - Frame::HEADER_SIZE)
+ numRemainingBytes = numDataBytes - numBytesTransferred - Frame::HEADER_SIZE;
+ return numRemainingBytes;
}
}
diff --git a/NetworkingLib/test/Main.cpp b/NetworkingLib/test/Main.cpp
index 66fb4c1185dc1614424fbb119ebcd471541d4841..9b58d1a3bac8c0bac488170a0d672b848c7ee8dd 100644
--- a/NetworkingLib/test/Main.cpp
+++ b/NetworkingLib/test/Main.cpp
@@ -8,8 +8,10 @@
int main(int argc, char ** argv)
{
- std::cout << "\ntestService\n\n";
- networking::test::testServices();
+ std::cout << "\ntestSyncService\n\n";
+ networking::test::testSyncServices();
+ std::cout << "\ntestAsyncService\n\n";
+ networking::test::testAsyncServices();
std::cout <<"\ntestTcpClientTimeout\n\n";
networking::test::testTcpClientTimeout();
std::cout <<"\ntestMultipleConnections\n\n";
@@ -30,5 +32,11 @@ int main(int argc, char ** argv)
networking::test::testStringMessageOverDatagram();
std::cout << "\ntestStringMessageOverService\n\n";
networking::test::testStringMessageOverService();
+ std::cout << "\ntestServiceServerMaxMessageSize\n\n";
+ networking::test::testServiceServerMaxMessageSize();
+ std::cout << "\ntestServiceClientMaxMessageSize\n\n";
+ networking::test::testServiceClientMaxMessageSize();
+ std::cout << "\ntestDatagramReceiverMaxMessageSize\n\n";
+ networking::test::testDatagramReceiverMaxMessageSize();
return 0;
}
\ No newline at end of file
diff --git a/NetworkingLib/test/Test.cpp b/NetworkingLib/test/Test.cpp
index ee5f58ea6609c04d8d52a602b5c7755b2a473934..0cb6f2a00d755bdaf75c65be167f0e3006afed2d 100644
--- a/NetworkingLib/test/Test.cpp
+++ b/NetworkingLib/test/Test.cpp
@@ -21,7 +21,7 @@ namespace networking
namespace test
{
-void testServices()
+void testSyncServices()
{
using namespace protocol;
Networking net1;
@@ -43,14 +43,74 @@ void testServices()
sleep(1);
+ int correct = 0;
+
auto client = service::Client<PlatoonService>::create(net2);
for (int i = 0; i < 5; i++)
{
- auto response = client->call("127.0.0.1", 10001, PlatoonMessage::followerRequest(2), 5s);
+ auto response = client->call("127.0.0.1", 10001, PlatoonMessage::followerRequest(2), 1s);
std::cout << "Response from " << (int) response.getVehicleId() << " with type: "
<< (int) response.getMessageType()
<< " and platoonId: " << (int) response.getPlatoonId() << "\n";
+ if (response.getVehicleId() == 1 && response.getPlatoonId() == 42)
+ correct++;
}
+
+ if (correct == 5)
+ std::cout << "SUCCESS!\n";
+}
+
+void testAsyncServices()
+{
+ using namespace protocol;
+ Networking net;
+
+ auto server = service::Server<PlatoonService>::create(net, 10001);
+
+ server->advertiseService(
+ [](const auto & clientEndpoint,
+ auto & requestMessage) -> PlatoonMessage
+ {
+ std::cout << "Request from "
+ << (int) requestMessage.getVehicleId()
+ << " with type: "
+ << (int) requestMessage.getMessageType()
+ << "\n";
+ return PlatoonMessage::acceptResponse(1, 42);
+ });
+
+ sleep(1);
+
+ std::atomic<std::size_t> pending{5};
+ std::atomic<std::size_t> correct{0};
+
+ auto client = service::Client<PlatoonService>::create(net);
+ for (int i = 0; i < 5; i++)
+ {
+ client->asyncCall(
+ "127.0.0.1", 10001, PlatoonMessage::followerRequest(2), 1s,
+ [&pending, &correct](const auto & error, auto & response)
+ {
+ if (error)
+ std::cout << "FAILED!\n";
+ else
+ {
+ std::cout << "Response from " << (int) response.getVehicleId() << " with type: "
+ << (int) response.getMessageType()
+ << " and platoonId: " << (int) response.getPlatoonId() << "\n";
+ if (response.getVehicleId() == 1 && response.getPlatoonId() == 42)
+ correct++;
+ }
+
+ pending--;
+ });
+
+ while (client->isCalling());
+ }
+
+ while (pending > 0);
+ if (correct == 5)
+ std::cout << "SUCCESS!\n";
}
void testTcpClientTimeout()
@@ -118,11 +178,15 @@ void testMultipleConnections()
auto client = service::Client<PlatoonService>::create(net3);
- auto response = client->call("127.0.0.1", 10001, PlatoonMessage::followerRequest(1), 5s);
- std::cout << "Response from " << response.getVehicleId() << std::endl;
+ auto response1 = client->call("127.0.0.1", 10001, PlatoonMessage::followerRequest(1), 5s);
+ std::cout << "Response from " << response1.getVehicleId() << std::endl;
+
+ auto response2 = client->call("127.0.0.1", 10002, PlatoonMessage::followerRequest(2), 5s);
+ std::cout << "Response from " << response2.getVehicleId() << std::endl;
- response = client->call("127.0.0.1", 10002, PlatoonMessage::followerRequest(2), 5s);
- std::cout << "Response from " << response.getVehicleId() << std::endl;
+ if (response1.getVehicleId() == 1 && response1.getPlatoonId() == 42 &&
+ response1.getVehicleId() == 2 && response1.getPlatoonId() == 43)
+ std::cout << "SUCCESS!\n";
}
void testStoppingServiceServer()
@@ -403,6 +467,138 @@ void testStringMessageOverService()
receiverThread.join();
}
+void testServiceServerMaxMessageSize()
+{
+ Networking net;
+
+ std::atomic<bool> running{true};
+ std::atomic<bool> syncCallError{false};
+
+ auto server = service::Server<StringService>::create(net, 10000, 100);
+ server->advertiseService(
+ [](auto && ...)
+ {
+ std::cout << "FAILED! (This should not have been called!\n";
+ return std::string{};
+ });
+
+ sleep(1);
+ auto client = service::Client<StringService>::create(net, 200);
+ try
+ {
+ client->call("127.0.0.1", 10000, std::string(200, 'a'), 1s);
+ }
+ catch (const error::Error &)
+ {
+ syncCallError = true;
+ }
+
+ client->asyncCall(
+ "127.0.0.1", 10000, std::string(200, 'a'), 1s,
+ [&syncCallError, &running](const auto & error, auto & message)
+ {
+ if (error == error::codes::FAILED_OPERATION && syncCallError)
+ std::cout << "SUCCESS!\n";
+
+ running = false;
+ });
+
+ while (running);
+}
+
+void testServiceClientMaxMessageSize()
+{
+ Networking net;
+
+ std::atomic<bool> running{true};
+ std::atomic<std::size_t> serverReceivedCount{0};
+ std::atomic<bool> syncCallError{false};
+
+ auto server = service::Server<StringService>::create(net, 10000, 200);
+ server->advertiseService(
+ [&serverReceivedCount](auto && ...)
+ {
+ serverReceivedCount++;
+ return std::string(200, 'a');
+ });
+
+ sleep(1);
+ auto client = service::Client<StringService>::create(net, 100);
+ try
+ {
+ client->call("127.0.0.1", 10000, std::string{}, 1s);
+ }
+ catch (const error::Error &)
+ {
+ syncCallError = true;
+ }
+
+ client->asyncCall(
+ "127.0.0.1", 10000, std::string(100, 'a'), 1s,
+ [&syncCallError, &running, &serverReceivedCount](const auto & error, auto & message)
+ {
+ if (error == error::codes::FAILED_OPERATION && syncCallError && serverReceivedCount == 2)
+ std::cout << "SUCCESS!\n";
+
+ running = false;
+ });
+
+ while (running);
+}
+
+void testDatagramReceiverMaxMessageSize()
+{
+ Networking net;
+
+ std::atomic<bool> asyncReceiveFailed{false};
+ std::atomic<bool> running{true};
+
+ auto receiver = message::DatagramReceiver<std::string>::create(net, 10000, 100);
+ auto sender = message::DatagramSender<std::string>::create(net);
+
+ receiver->asyncReceive(
+ 1s,
+ [&asyncReceiveFailed, &running](const auto & error, auto && ... args)
+ {
+ if (error)
+ asyncReceiveFailed = true;
+ else
+ std::cout << "FAILED! (This should not be called!\n";
+
+ running = false;
+ });
+
+ sender->send(std::string(200, 'a'), "127.0.0.1", 10000, 1s);
+
+ while (running);
+
+ running = true;
+
+ std::thread receiveThread{
+ [&running, &asyncReceiveFailed, receiver]()
+ {
+ try
+ {
+ std::string host{};
+ std::uint16_t port;
+ receiver->receive(host, port, 3s);
+ }
+ catch (const error::Error &)
+ {
+ if (&asyncReceiveFailed)
+ std::cout << "SUCCESS!\n";
+ }
+
+ running = false;
+ }};
+
+ sleep(1);
+ sender->send(std::string(200, 'a'), "127.0.0.1", 10000, 1s);
+
+ while (running);
+ receiveThread.join();
+}
+
}
}
diff --git a/NetworkingLib/test/Test.h b/NetworkingLib/test/Test.h
index d7b7e9ba2d50602717450f8c93ee9d7517ddd7b9..e519d005bf5d961860349784745a3fc21f48c6a5 100644
--- a/NetworkingLib/test/Test.h
+++ b/NetworkingLib/test/Test.h
@@ -19,7 +19,9 @@ public:
using ResponseMessage = std::string;
};
-void testServices();
+void testSyncServices();
+
+void testAsyncServices();
void testTcpClientTimeout();
@@ -41,6 +43,12 @@ void testStringMessageOverDatagram();
void testStringMessageOverService();
+void testServiceServerMaxMessageSize();
+
+void testServiceClientMaxMessageSize();
+
+void testDatagramReceiverMaxMessageSize();
+
}
}