c++ - 使用 BOOST ASIO 缓冲器

Question

我正在按照 http://www.boost.org/doc/libs/1_53_0/doc/html/boost_asio/example/serialization/connection.hpp 中的示例进行操作.我已经修改了 connection.hpp header 以插入更多称为 ynchronous_read 和 asynchronous_write 的操作。代码附在此处...

    //
// connection.hpp
// ~~~~~~~~~~~~~~
//
// Copyright (c) 2003-2012 Christopher M. Kohlhoff (chris at kohlhoff dot com)
//
// Distributed under the Boost Software License, Version 1.0. (See accompanying
// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
//

#ifndef SERIALIZATION_CONNECTION_HPP
#define SERIALIZATION_CONNECTION_HPP

#include <boost/asio.hpp>
#include <boost/archive/text_iarchive.hpp>
#include <boost/archive/text_oarchive.hpp>
#include <boost/bind.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/tuple/tuple.hpp>
#include <iomanip>
#include <string>
#include <sstream>
#include <vector>

namespace s11n_example {

/// The connection class provides serialization primitives on top of a socket.
/**
 * Each message sent using this class consists of:
 * @li An 8-byte header containing the length of the serialized data in
 * hexadecimal.
 * @li The serialized data.
 */
class connection
{
public:
  /// Constructor.
  connection(boost::asio::io_service& io_service)
    : socket_(io_service)
  {
  }

  /// Get the underlying socket. Used for making a connection or for accepting
  /// an incoming connection.
  boost::asio::ip::tcp::socket& socket()
  {
    return socket_;
  }

  /// Asynchronously write a data structure to the socket.
  template <typename T, typename Handler>
  void async_write(const T& t, Handler handler)
  {
    // Serialize the data first so we know how large it is.
    std::ostringstream archive_stream;
    boost::archive::text_oarchive archive(archive_stream);
    archive << t;
    outbound_data_ = archive_stream.str();

    // Format the header.
    std::ostringstream header_stream;
    header_stream << std::setw(header_length)
      << std::hex << outbound_data_.size();
    if (!header_stream || header_stream.str().size() != header_length)
    {
      // Something went wrong, inform the caller.
      boost::system::error_code error(boost::asio::error::invalid_argument);
      socket_.get_io_service().post(boost::bind(handler, error));
      return;
    }
    outbound_header_ = header_stream.str();

    // Write the serialized data to the socket. We use "gather-write" to send
    // both the header and the data in a single write operation.
    std::vector<boost::asio::const_buffer> buffers;
    buffers.push_back(boost::asio::buffer(outbound_header_));
    buffers.push_back(boost::asio::buffer(outbound_data_));
    boost::asio::async_write(socket_, buffers, handler);
  }

  ///Asynchronously write a string to the socket.
  template <typename T, typename Handler>
  void asyncronous_write(const T& t, Handler handler)
  {
    std::ostringstream archive_stream;
    boost::archive::text_oarchive archive(archive_stream);
    archive << t;
    outbound_str_ = archive_stream.str();
    //boost::asio::buffer buffer1(outbound_str_);
    boost::asio::async_write(socket_,boost::asio::buffer(outbound_str_), handler);
  }
  /// Asynchronously read a string from the socket.
  template <typename T, typename Handler>
  void asyncronous_read(T& t, Handler handler)
  {
    // Issue a read operation to read exactly the number of bytes in a header.
    void (connection::*f)(
        const boost::system::error_code&,
        T&, boost::tuple<Handler>)
      = &connection::handle_read_str<T, Handler>;
    boost::asio::async_read(socket_, boost::asio::buffer(inbound_str_),
        boost::bind(f,
          this, boost::asio::placeholders::error, boost::ref(t),
          boost::make_tuple(handler)));
  }

  /// Handle a completed read of message data.
  template <typename T, typename Handler>
  void handle_read_str(const boost::system::error_code& e,
      T& t, boost::tuple<Handler> handler)
  {
    if (e)
    {
      boost::get<0>(handler)(e);
    }
    else
    {
      // Extract the data structure from the data just received.
      try
      {
        std::string archive_data(&inbound_str_[0], inbound_str_.size());
        std::istringstream archive_stream(archive_data);
        boost::archive::text_iarchive archive(archive_stream);
        archive >> t;
      }
      catch (std::exception& e)
      {
        // Unable to decode data.
        boost::system::error_code error(boost::asio::error::invalid_argument);
        boost::get<0>(handler)(error);
        return;
      }

      // Inform caller that data has been received ok.
      boost::get<0>(handler)(e);
    }
  }



  /// Asynchronously read a data structure from the socket.
  template <typename T, typename Handler>
  void async_read(T& t, Handler handler)
  {
    // Issue a read operation to read exactly the number of bytes in a header.
    void (connection::*f)(
        const boost::system::error_code&,
        T&, boost::tuple<Handler>)
      = &connection::handle_read_header<T, Handler>;
    boost::asio::async_read(socket_, boost::asio::buffer(inbound_header_),
        boost::bind(f,
          this, boost::asio::placeholders::error, boost::ref(t),
          boost::make_tuple(handler)));
  }

  /// Handle a completed read of a message header. The handler is passed using
  /// a tuple since boost::bind seems to have trouble binding a function object
  /// created using boost::bind as a parameter.
  template <typename T, typename Handler>
  void handle_read_header(const boost::system::error_code& e,
      T& t, boost::tuple<Handler> handler)
  {
    if (e)
    {
      boost::get<0>(handler)(e);
    }
    else
    {
      // Determine the length of the serialized data.
      std::istringstream is(std::string(inbound_header_, header_length));
      std::size_t inbound_data_size = 0;
      if (!(is >> std::hex >> inbound_data_size))
      {
        // Header doesn't seem to be valid. Inform the caller.
        boost::system::error_code error(boost::asio::error::invalid_argument);
        boost::get<0>(handler)(error);
        return;
      }

      // Start an asynchronous call to receive the data.
      inbound_data_.resize(inbound_data_size);
      void (connection::*f)(
          const boost::system::error_code&,
          T&, boost::tuple<Handler>)
        = &connection::handle_read_data<T, Handler>;
      boost::asio::async_read(socket_, boost::asio::buffer(inbound_data_),
        boost::bind(f, this,
          boost::asio::placeholders::error, boost::ref(t), handler));
    }
  }

  /// Handle a completed read of message data.
  template <typename T, typename Handler>
  void handle_read_data(const boost::system::error_code& e,
      T& t, boost::tuple<Handler> handler)
  {
    if (e)
    {
      boost::get<0>(handler)(e);
    }
    else
    {
      // Extract the data structure from the data just received.
      try
      {
        std::string archive_data(&inbound_data_[0], inbound_data_.size());
        std::istringstream archive_stream(archive_data);
        boost::archive::text_iarchive archive(archive_stream);
        archive >> t;
      }
      catch (std::exception& e)
      {
        // Unable to decode data.
        boost::system::error_code error(boost::asio::error::invalid_argument);
        boost::get<0>(handler)(error);
        return;
      }

      // Inform caller that data has been received ok.
      boost::get<0>(handler)(e);
    }
  }

private:
  /// The underlying socket.
  boost::asio::ip::tcp::socket socket_;

  /// The size of a fixed length header.
  enum { header_length = 8 };

  /// Holds an outbound header.
  std::string outbound_header_;

  /// Holds the outbound data.
  std::string outbound_data_;
  std::string outbound_str_;

  /// Holds an inbound header.
  char inbound_header_[header_length];

  /// Holds the inbound data.
  std::vector<char> inbound_data_;
  std::vector<char>inbound_str_;
  //char* inbound_str_=new char[1024];
};

typedef boost::shared_ptr<connection> connection_ptr;

} // namespace s11n_example

#endif // SERIALIZATION_CONNECTION_HPP

服务端和客户端代码是---

#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/lexical_cast.hpp>
#include <iostream>
#include <fstream>
#include <vector>
#include "connection.hpp" // Must come before boost/serialization headers.
#include <boost/serialization/vector.hpp>
#include "structsample.hpp"


namespace s11n_example {

/// Serves stock quote information to any client that connects to it.
class server
{
public:
  /// Constructor opens the acceptor and starts waiting for the first incoming
  /// connection.

  server(boost::asio::io_service& io_service, unsigned short port)
    : acceptor_(io_service,
        boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), port))
  {
    // Create the data to be sent to each client.
     std::string str;
     std::ifstream in;
     stock s;
     in.open("aces.clf");
     if(!in)
     {
     std::cout<<"Error in Opening a file"<<std::endl;
     exit(1);
     }
     for(int i=0;i<13;i++)
     {
      getline(in,str);
     }
     while(!in.eof())
     {
        getline(in,str);
        s.data.append(str);
        getline(in,str);
        s.data.append(str);
        s.d_size=s.data.size();
        stocks_.push_back(s);
        s.data.clear();
     }

    // Start an accept operation for a new connection.
    connection_ptr new_conn(new connection(acceptor_.get_io_service()));
    acceptor_.async_accept(new_conn->socket(),
        boost::bind(&server::handle_accept, this,
          boost::asio::placeholders::error, new_conn));
  }

  /// Handle completion of a accept operation.
  void handle_accept(const boost::system::error_code& e, connection_ptr conn)
  {
   std::string buf1;
    if (!e)
    {

      // Successfully accepted a new connection. Send the list of stocks to the
      // client. The connection::async_write() function will automatically
      // serialize the data structure for us.
    for(int i=0;i<5;i++)
     {

      conn->async_write(stocks_[i],
          boost::bind(&server::handle_write, this,
            boost::asio::placeholders::error, conn));

      usleep(500000);
      conn->async_read(buf1,
          boost::bind(&server::handle_read, this,
            boost::asio::placeholders::error, conn));
       std::cout<<buf1<<"\n"<<std::endl;
    }

      // Start an accept operation for a new connection.
      connection_ptr new_conn(new connection(acceptor_.get_io_service()));
      acceptor_.async_accept(new_conn->socket(),
          boost::bind(&server::handle_accept, this,
            boost::asio::placeholders::error, new_conn));
    }
    else
    {
      // An error occurred. Log it and return. Since we are not starting a new
      // accept operation the io_service will run out of work to do and the
      // server will exit.
      std::cerr << e.message() << std::endl;
    }
  }

  /// Handle completion of a write operation.
  void handle_write(const boost::system::error_code& e, connection_ptr conn)
  {

      //conn->async_read(buf1,
          //boost::bind(&server::handle_read, this,
           // boost::asio::placeholders::error, conn));
     // std::cout<<buf1<<"\n"<<std::endl;

    }

///handle completion of read operation.
void handle_read(const boost::system::error_code& e,connection_ptr conn)
{
 //if(!e)
    //{
          //std::cout<<"\n"<<stocks_1.data<<std::endl;
          //stocks_1.data.clear();
    //}
         //{
           //if (buf1.compare("data received and processed")==0)
            //{
                          // buf1.clear();
                            //std::cout<<buf1<<"\n"<<std::endl;

            //}
           //else
            //{
                //std::cout<<"Error occurred"<<std::endl;
                //exit(1);
           // }
    //}
}

private:
  /// The acceptor object used to accept incoming socket connections.
  boost::asio::ip::tcp::acceptor acceptor_;

  /// The data to be sent to each client.
  std::vector<stock> stocks_;
};

} // namespace s11n_example

int main(int argc, char* argv[])
{
  try
  {
    // Check command line arguments.
    if (argc != 2)
    {
      std::cerr << "Usage: server <port>" << std::endl;
      return 1;
    }
    unsigned short port = boost::lexical_cast<unsigned short>(argv[1]);
    boost::asio::io_service io_service;
    s11n_example::server server(io_service, port);
    io_service.run();
  }
  catch (std::exception& e)
  {
    std::cerr << e.what() << std::endl;
  }

  return 0;
}






#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <iostream>
#include <fstream>
#include <vector>
#include "connection.hpp" // Must come before boost/serialization headers.
#include <boost/serialization/vector.hpp>
#include "structsample.hpp"

struct stock1
  {
    std::string data;
    int d_size;
   template <typename Archive>
  void serialize(Archive& ar, const unsigned int version)
  {
    ar & data;
    ar & d_size;

  }
} stocks_1;
namespace s11n_example {

/// Downloads stock quote information from a server.
class client
{
public:
  /// Constructor starts the asynchronous connect operation.
  client(boost::asio::io_service& io_service,
      const std::string& host, const std::string& service)
    : connection_(io_service)
  {
    // Resolve the host name into an IP address.
    boost::asio::ip::tcp::resolver resolver(io_service);
    boost::asio::ip::tcp::resolver::query query(host, service);
    boost::asio::ip::tcp::resolver::iterator endpoint_iterator =
      resolver.resolve(query);
    boost::asio::ip::tcp::endpoint endpoint = *endpoint_iterator;

// Start an asynchronous connect operation.
    connection_.socket().async_connect(endpoint,
        boost::bind(&client::handle_connect, this,
          boost::asio::placeholders::error, ++endpoint_iterator));
  }

  /// Handle completion of a connect operation.
  void handle_connect(const boost::system::error_code& e,
      boost::asio::ip::tcp::resolver::iterator endpoint_iterator)
  {
    if (!e)
    {
      // Successfully established connection. Start operation to read the list
      // of stocks. The connection::async_read() function will automatically
      // decode the data that is read from the underlying socket.
      connection_.async_read(stocks_1,
          boost::bind(&client::handle_read, this,
            boost::asio::placeholders::error));

    }
    else if (endpoint_iterator != boost::asio::ip::tcp::resolver::iterator())
    {
      // Try the next endpoint.
      connection_.socket().close();
      boost::asio::ip::tcp::endpoint endpoint = *endpoint_iterator;
      connection_.socket().async_connect(endpoint,
          boost::bind(&client::handle_connect, this,
            boost::asio::placeholders::error, ++endpoint_iterator));
    }

 else
    {
      // An error occurred. Log it and return. Since we are not starting a new
      // operation the io_service will run out of work to do and the client will
      // exit.
      std::cerr << e.message() << std::endl;
    }
  }

  /// Handle completion of a read operation.
  void handle_read(const boost::system::error_code& e)
  {
   std::string buf2;
    if (!e)
    {

        std::cout << "  data: " << stocks_1.data << "\n";
        std::cout << "  size: " << stocks_1.d_size << "\n";

        usleep(500000);
      if(!e)
      {
       //stocks_1.data="data received and processed";
       //tocks_1.d_size=stocks_1.data.size();
       connection_.async_write(buf2,
          boost::bind(&client::handle_write, this,
            boost::asio::placeholders::error));
       std::cout<<buf2<<std::endl;
        //buf2.clear();
       // connection_.async_write(stocks_1,
          //boost::bind(&client::handle_write, this,
            //boost::asio::placeholders::error));
       //stocks_1.data.clear();
      }
    }
      else
    {
      // An error occurred.
      std::cerr << e.message() << std::endl;
    }


// Since we are not starting a new operation the io_service will run out of
    // work to do and the client will exit.

  }

void handle_write(const boost::system::error_code& e)
{
  if(!e)
{
          connection_.async_read(stocks_1,
          boost::bind(&client::handle_read, this,
            boost::asio::placeholders::error));
}
else
    //std::cout<<"Error is there"<<std::endl;
        exit(1);
}
private:
  /// The connection to the server.
  connection connection_;

  /// The data received from the server.
  std::vector<stock> stocks_;


};

} // namespace s11n_example

int main(int argc, char* argv[])
{
  try
  {
    // Check command line arguments.
    if (argc != 3)
    {
      std::cerr << "Usage: client <host> <port>" << std::endl;
      return 1;
    }

    boost::asio::io_service io_service;
    s11n_example::client client(io_service, argv[1], argv[2]);
    io_service.run();
  }
  catch (std::exception& e)
  {
    std::cerr << e.what() << std::endl;
  }

  return 0;
}

结构样本.hpp----

#ifndef SERIALIZATION_STOCK_HPP
#define SERIALIZATION_STOCK_HPP

#include <string>

namespace s11n_example {

/// Structure to hold information about a single stock.
struct stock
{
 public:

  std::string data;
  int d_size;

  template <typename Archive>
  void serialize(Archive& ar, const unsigned int version)
  {
    ar & data;
    ar & d_size;

  }
};

} // namespace s11n_example

#endif // SERIALIZATION_STOCK_HPP

此处数据从服务器到达客户端，但客户端的响应未发回。

Answer 1

ASIO 真正传输的是数据字节。字符串不适合存储任意数据(可能包含空字符，如果包含在字符串中会导致混淆结果)。

要使用boost::asio传输字符串，最好使用基于std::vector或std::array的缓冲区，然后分析此缓冲区中的数据以获取您的字符串。