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

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