ceph/src/boost/libs/asio/example/cpp03/porthopper/client.cpp

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