client.cpp 6.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192
  1. //
  2. // client.cpp
  3. // ~~~~~~~~~~
  4. //
  5. // Copyright (c) 2003-2019 Christopher M. Kohlhoff (chris at kohlhoff dot com)
  6. //
  7. // Distributed under the Boost Software License, Version 1.0. (See accompanying
  8. // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
  9. //
  10. #include <boost/asio.hpp>
  11. #include <boost/lambda/lambda.hpp>
  12. #include <boost/lambda/bind.hpp>
  13. #include <boost/lambda/if.hpp>
  14. #include <boost/shared_ptr.hpp>
  15. #include <algorithm>
  16. #include <cstdlib>
  17. #include <exception>
  18. #include <iostream>
  19. #include <string>
  20. #include "protocol.hpp"
  21. using namespace boost;
  22. using boost::asio::ip::tcp;
  23. using boost::asio::ip::udp;
  24. int main(int argc, char* argv[])
  25. {
  26. try
  27. {
  28. if (argc != 3)
  29. {
  30. std::cerr << "Usage: client <host> <port>\n";
  31. return 1;
  32. }
  33. using namespace std; // For atoi.
  34. std::string host_name = argv[1];
  35. std::string port = argv[2];
  36. boost::asio::io_context io_context;
  37. // Determine the location of the server.
  38. tcp::resolver resolver(io_context);
  39. tcp::endpoint remote_endpoint = *resolver.resolve(host_name, port).begin();
  40. // Establish the control connection to the server.
  41. tcp::socket control_socket(io_context);
  42. control_socket.connect(remote_endpoint);
  43. // Create a datagram socket to receive data from the server.
  44. boost::shared_ptr<udp::socket> data_socket(
  45. new udp::socket(io_context, udp::endpoint(udp::v4(), 0)));
  46. // Determine what port we will receive data on.
  47. udp::endpoint data_endpoint = data_socket->local_endpoint();
  48. // Ask the server to start sending us data.
  49. control_request start = control_request::start(data_endpoint.port());
  50. boost::asio::write(control_socket, start.to_buffers());
  51. unsigned long last_frame_number = 0;
  52. for (;;)
  53. {
  54. // Receive 50 messages on the current data socket.
  55. for (int i = 0; i < 50; ++i)
  56. {
  57. // Receive a frame from the server.
  58. frame f;
  59. data_socket->receive(f.to_buffers(), 0);
  60. if (f.number() > last_frame_number)
  61. {
  62. last_frame_number = f.number();
  63. std::cout << "\n" << f.payload();
  64. }
  65. }
  66. // Time to switch to a new socket. To ensure seamless handover we will
  67. // continue to receive packets using the old socket until data arrives on
  68. // the new one.
  69. std::cout << " Starting renegotiation";
  70. // Create the new data socket.
  71. boost::shared_ptr<udp::socket> new_data_socket(
  72. new udp::socket(io_context, udp::endpoint(udp::v4(), 0)));
  73. // Determine the new port we will use to receive data.
  74. udp::endpoint new_data_endpoint = new_data_socket->local_endpoint();
  75. // Ask the server to switch over to the new port.
  76. control_request change = control_request::change(
  77. data_endpoint.port(), new_data_endpoint.port());
  78. boost::system::error_code control_result;
  79. boost::asio::async_write(control_socket, change.to_buffers(),
  80. (
  81. lambda::var(control_result) = lambda::_1
  82. ));
  83. // Try to receive a frame from the server on the new data socket. If we
  84. // successfully receive a frame on this new data socket we can consider
  85. // the renegotation complete. In that case we will close the old data
  86. // socket, which will cause any outstanding receive operation on it to be
  87. // cancelled.
  88. frame f1;
  89. boost::system::error_code new_data_socket_result;
  90. new_data_socket->async_receive(f1.to_buffers(),
  91. (
  92. // Note: lambda::_1 is the first argument to the callback handler,
  93. // which in this case is the error code for the operation.
  94. lambda::var(new_data_socket_result) = lambda::_1,
  95. lambda::if_(!lambda::_1)
  96. [
  97. // We have successfully received a frame on the new data socket,
  98. // so we can close the old data socket. This will cancel any
  99. // outstanding receive operation on the old data socket.
  100. lambda::var(data_socket) = boost::shared_ptr<udp::socket>()
  101. ]
  102. ));
  103. // This loop will continue until we have successfully completed the
  104. // renegotiation (i.e. received a frame on the new data socket), or some
  105. // unrecoverable error occurs.
  106. bool done = false;
  107. while (!done)
  108. {
  109. // Even though we're performing a renegotation, we want to continue
  110. // receiving data as smoothly as possible. Therefore we will continue to
  111. // try to receive a frame from the server on the old data socket. If we
  112. // receive a frame on this socket we will interrupt the io_context,
  113. // print the frame, and resume waiting for the other operations to
  114. // complete.
  115. frame f2;
  116. done = true; // Let's be optimistic.
  117. if (data_socket) // Might have been closed by new_data_socket's handler.
  118. {
  119. data_socket->async_receive(f2.to_buffers(), 0,
  120. (
  121. lambda::if_(!lambda::_1)
  122. [
  123. // We have successfully received a frame on the old data
  124. // socket. Stop the io_context so that we can print it.
  125. lambda::bind(&boost::asio::io_context::stop, &io_context),
  126. lambda::var(done) = false
  127. ]
  128. ));
  129. }
  130. // Run the operations in parallel. This will block until all operations
  131. // have finished, or until the io_context is interrupted. (No threads!)
  132. io_context.restart();
  133. io_context.run();
  134. // If the io_context.run() was interrupted then we have received a frame
  135. // on the old data socket. We need to keep waiting for the renegotation
  136. // operations to complete.
  137. if (!done)
  138. {
  139. if (f2.number() > last_frame_number)
  140. {
  141. last_frame_number = f2.number();
  142. std::cout << "\n" << f2.payload();
  143. }
  144. }
  145. }
  146. // Since the loop has finished, we have either successfully completed
  147. // the renegotation, or an error has occurred. First we'll check for
  148. // errors.
  149. if (control_result)
  150. throw boost::system::system_error(control_result);
  151. if (new_data_socket_result)
  152. throw boost::system::system_error(new_data_socket_result);
  153. // If we get here it means we have successfully started receiving data on
  154. // the new data socket. This new data socket will be used from now on
  155. // (until the next time we renegotiate).
  156. std::cout << " Renegotiation complete";
  157. data_socket = new_data_socket;
  158. data_endpoint = new_data_endpoint;
  159. if (f1.number() > last_frame_number)
  160. {
  161. last_frame_number = f1.number();
  162. std::cout << "\n" << f1.payload();
  163. }
  164. }
  165. }
  166. catch (std::exception& e)
  167. {
  168. std::cerr << "Exception: " << e.what() << std::endl;
  169. }
  170. return 0;
  171. }