NetworkConnection.cpp 6.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264
  1. /*
  2. * NetworkConnection.cpp, part of VCMI engine
  3. *
  4. * Authors: listed in file AUTHORS in main folder
  5. *
  6. * License: GNU General Public License v2.0 or later
  7. * Full text of license available in license.txt file, in main folder
  8. *
  9. */
  10. #include "StdInc.h"
  11. #include "NetworkConnection.h"
  12. VCMI_LIB_NAMESPACE_BEGIN
  13. NetworkConnection::NetworkConnection(INetworkConnectionListener & listener, const std::shared_ptr<NetworkSocket> & socket, const std::shared_ptr<NetworkContext> & context)
  14. : socket(socket)
  15. , timer(std::make_shared<NetworkTimer>(*context))
  16. , listener(listener)
  17. {
  18. socket->set_option(boost::asio::ip::tcp::no_delay(true));
  19. // iOS throws exception on attempt to set buffer size
  20. constexpr auto bufferSize = 4 * 1024 * 1024;
  21. try
  22. {
  23. socket->set_option(boost::asio::socket_base::send_buffer_size{bufferSize});
  24. }
  25. catch(const boost::system::system_error & e)
  26. {
  27. logNetwork->error("error setting 'send buffer size' socket option: %s", e.what());
  28. }
  29. try
  30. {
  31. socket->set_option(boost::asio::socket_base::receive_buffer_size{bufferSize});
  32. }
  33. catch(const boost::system::system_error & e)
  34. {
  35. logNetwork->error("error setting 'receive buffer size' socket option: %s", e.what());
  36. }
  37. }
  38. void NetworkConnection::start()
  39. {
  40. heartbeat();
  41. startReceiving();
  42. }
  43. void NetworkConnection::startReceiving()
  44. {
  45. boost::asio::async_read(*socket,
  46. readBuffer,
  47. boost::asio::transfer_exactly(messageHeaderSize),
  48. [self = shared_from_this()](const auto & ec, const auto & endpoint) { self->onHeaderReceived(ec); });
  49. }
  50. void NetworkConnection::heartbeat()
  51. {
  52. constexpr auto heartbeatInterval = std::chrono::seconds(10);
  53. timer->expires_after(heartbeatInterval);
  54. timer->async_wait( [self = weak_from_this()](const auto & ec)
  55. {
  56. if (ec)
  57. return;
  58. auto locked = self.lock();
  59. if (!locked)
  60. return;
  61. locked->sendPacket({});
  62. locked->heartbeat();
  63. });
  64. }
  65. void NetworkConnection::onHeaderReceived(const boost::system::error_code & ecHeader)
  66. {
  67. if (ecHeader)
  68. {
  69. onError(ecHeader.message());
  70. return;
  71. }
  72. if (readBuffer.size() < messageHeaderSize)
  73. throw std::runtime_error("Failed to read header!");
  74. uint32_t messageSize;
  75. readBuffer.sgetn(reinterpret_cast<char *>(&messageSize), sizeof(messageSize));
  76. if (messageSize > messageMaxSize)
  77. {
  78. onError("Invalid packet size!");
  79. return;
  80. }
  81. if (messageSize == 0)
  82. {
  83. //heartbeat package with no payload - wait for next packet
  84. startReceiving();
  85. return;
  86. }
  87. boost::asio::async_read(*socket,
  88. readBuffer,
  89. boost::asio::transfer_exactly(messageSize),
  90. [self = shared_from_this(), messageSize](const auto & ecPayload, const auto & endpoint) { self->onPacketReceived(ecPayload, messageSize); });
  91. }
  92. void NetworkConnection::onPacketReceived(const boost::system::error_code & ec, uint32_t expectedPacketSize)
  93. {
  94. if (ec)
  95. {
  96. onError(ec.message());
  97. return;
  98. }
  99. if (readBuffer.size() < expectedPacketSize)
  100. {
  101. // FIXME: figure out what causes this. This should not be possible without error set
  102. std::string errorMessage = "Failed to read packet! " + std::to_string(readBuffer.size()) + " bytes read, but " + std::to_string(expectedPacketSize) + " bytes expected!";
  103. onError(errorMessage);
  104. }
  105. std::vector<std::byte> message(expectedPacketSize);
  106. readBuffer.sgetn(reinterpret_cast<char *>(message.data()), expectedPacketSize);
  107. listener.onPacketReceived(shared_from_this(), message);
  108. startReceiving();
  109. }
  110. void NetworkConnection::setAsyncWritesEnabled(bool on)
  111. {
  112. asyncWritesEnabled = on;
  113. }
  114. void NetworkConnection::sendPacket(const std::vector<std::byte> & message)
  115. {
  116. std::lock_guard lock(writeMutex);
  117. std::vector<std::byte> headerVector(sizeof(uint32_t));
  118. uint32_t messageSize = message.size();
  119. std::memcpy(headerVector.data(), &messageSize, sizeof(uint32_t));
  120. // At the moment, vcmilobby *requires* async writes in order to handle multiple connections with different speeds and at optimal performance
  121. // However server (and potentially - client) can not handle this mode and may shutdown either socket or entire asio service too early, before all writes are performed
  122. if (asyncWritesEnabled)
  123. {
  124. bool messageQueueEmpty = dataToSend.empty();
  125. dataToSend.push_back(headerVector);
  126. if (!message.empty())
  127. dataToSend.push_back(message);
  128. if (messageQueueEmpty)
  129. doSendData();
  130. //else - data sending loop is still active and still sending previous messages
  131. }
  132. else
  133. {
  134. boost::system::error_code ec;
  135. boost::asio::write(*socket, boost::asio::buffer(headerVector), ec );
  136. if (!message.empty())
  137. boost::asio::write(*socket, boost::asio::buffer(message), ec );
  138. }
  139. }
  140. void NetworkConnection::doSendData()
  141. {
  142. if (dataToSend.empty())
  143. throw std::runtime_error("Attempting to sent data but there is no data to send!");
  144. boost::asio::async_write(*socket, boost::asio::buffer(dataToSend.front()), [self = shared_from_this()](const auto & error, const auto & )
  145. {
  146. self->onDataSent(error);
  147. });
  148. }
  149. void NetworkConnection::onDataSent(const boost::system::error_code & ec)
  150. {
  151. std::lock_guard lock(writeMutex);
  152. dataToSend.pop_front();
  153. if (ec)
  154. {
  155. onError(ec.message());
  156. return;
  157. }
  158. if (!dataToSend.empty())
  159. doSendData();
  160. }
  161. void NetworkConnection::onError(const std::string & message)
  162. {
  163. listener.onDisconnected(shared_from_this(), message);
  164. close();
  165. }
  166. void NetworkConnection::close()
  167. {
  168. boost::system::error_code ec;
  169. socket->close(ec);
  170. #if BOOST_VERSION >= 108700
  171. timer->cancel();
  172. #else
  173. timer->cancel(ec);
  174. #endif
  175. //NOTE: ignoring error code, intended
  176. }
  177. InternalConnection::InternalConnection(INetworkConnectionListener & listener, const std::shared_ptr<NetworkContext> & context)
  178. : io(context)
  179. , listener(listener)
  180. {
  181. }
  182. void InternalConnection::receivePacket(const std::vector<std::byte> & message)
  183. {
  184. boost::asio::post(*io, [self = std::static_pointer_cast<InternalConnection>(shared_from_this()), message](){
  185. if (self->connectionActive)
  186. self->listener.onPacketReceived(self, message);
  187. });
  188. }
  189. void InternalConnection::disconnect()
  190. {
  191. boost::asio::post(*io, [self = std::static_pointer_cast<InternalConnection>(shared_from_this())](){
  192. self->listener.onDisconnected(self, "Internal connection has been terminated");
  193. self->otherSideWeak.reset();
  194. self->connectionActive = false;
  195. });
  196. }
  197. void InternalConnection::connectTo(std::shared_ptr<IInternalConnection> connection)
  198. {
  199. otherSideWeak = connection;
  200. connectionActive = true;
  201. }
  202. void InternalConnection::sendPacket(const std::vector<std::byte> & message)
  203. {
  204. auto otherSide = otherSideWeak.lock();
  205. if (otherSide)
  206. otherSide->receivePacket(message);
  207. }
  208. void InternalConnection::setAsyncWritesEnabled(bool on)
  209. {
  210. // no-op
  211. }
  212. void InternalConnection::close()
  213. {
  214. auto otherSide = otherSideWeak.lock();
  215. if (otherSide)
  216. otherSide->disconnect();
  217. otherSideWeak.reset();
  218. connectionActive = false;
  219. }
  220. VCMI_LIB_NAMESPACE_END