Google :: protobuf + boost :: asio失败 [英] Google::protobuf + boost::asio failure

问题描述

我研究了现有示例:

1. 使用boost :: asio发送Protobuf消息
2. 使用boost :: asio :: read_async 读取Protobuf对象
3. Google协议缓冲区:C ++的parseDelimitedFrom和writeDelimitedTo
4. 是否有C ++ Java中的协议缓冲区定界I/O函数的等效项?
5. 使用boost :: asio发送Protobuf消息

1. Sending Protobuf Messages with boost::asio
2. Reading Protobuf objects using boost::asio::read_async
3. Google Protocol Buffers: parseDelimitedFrom and writeDelimitedTo for C++
4. Are there C++ equivalents for the Protocol Buffers delimited I/O functions in Java?
5. Sending Protobuf Messages with boost::asio

但是我仍然不知道如何使用Boost :: asio API传递Google Protobuf消息.特别是，我对以下问题没有清楚的了解:

but I still can not figure out how to pass Google Protobuf messages using the Boost::asio API. In particular I have no clear understanding of the following problems:

1. boost :: asio :: streambuf与google :: protobuf :: io对象之间的交互(以及最后一个对象的应用必要性)
2. 正确执行消息流(由于C ++ API中缺少writeDelimitedTo和parseDelimitedFrom方法)

以下是我基于boost :: asio v.1.39 ssl_client的实现，来自

Here is my implementation based on boost::asio v. 1.39 ssl_client from examples.

    class client
{
public:
  client(boost::asio::io_service& io_service, boost::asio::ssl::context& context,
      boost::asio::ip::tcp::resolver::iterator endpoint_iterator)
    : socket_(io_service, context),
        request_stream(&b),
        raw_output(&request_stream),
        coded_output(&raw_output)
  {
    ... 
  }

  void handle_connect(const boost::system::error_code& error,
      boost::asio::ip::tcp::resolver::iterator endpoint_iterator)
  {
    ...
  }

  //Debugging function
  void print_buffers_condition(const char *step)
  {
      std::cout << "\nBuffer conditions after " << step << std::endl;
      std::cout << "boost::asio::streambuf\t\tb: " << b.size() << std::endl;
      std::cout << "google::protobuf::io::OstreamOutputStream raw_output: " << raw_output.ByteCount() << std::endl;
      std::cout << "google::protobuf::io::CodedOutputStream coded_output: " << coded_output.ByteCount() << std::endl;
      std::cout << std::endl;
  }

  //Sending test message after SSL Handshake
  void handle_handshake(const boost::system::error_code& error)
  {
      std::cout << "-----------------------------SENDING-----------------------------" << std::endl;
    print_buffers_condition("handle handshake");
    if (!error)
    {
        SearchRequest msg;
        msg.set_query("qwerty");
        msg.set_code(12345);

        std::cout << "Debugged" << std::endl;
        msg.PrintDebugString();


        //Writing the length of the message before and serializing                 
                    print_buffers_condition("before serialising");
        coded_output.WriteVarint32(msg.ByteSize());
        if (!msg.SerializeToCodedStream(&coded_output))
        {
            std::cout << "serailizing error" << std::endl;
        }
        else
        {
            std::cout << "serializing success" << std::endl;
        }

        //Sending
        buffers_condition("before async write");
        boost::asio::async_write(socket_,
                                 b,
                                 boost::bind(&client::handle_write, this, boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred));
        buffers_condition("after async write");
    }
    else
    {
      std::cout << "Handshake failed: " << error << "\n";
    }
  }

  void handle_write(const boost::system::error_code& error,
      size_t bytes_transferred)
  {
    std::cout << " bytes_trransferred: " << bytes_transferred << std::endl;
    if (!error)
    {
        std::cout << "No error" << std::endl;
        ...
    }
    else
    {
      std::cout << "Write failed: " << error << "\n";
    }
  }

  void handle_read(const boost::system::error_code& error,
      size_t bytes_transferred)
  {
    ...
  }

private:
  boost::asio::ssl::stream<boost::asio::ip::tcp::socket> socket_;
  boost::asio::streambuf b;
  std::ostream request_stream;
  google::protobuf::io::OstreamOutputStream raw_output;
  google::protobuf::io::CodedOutputStream coded_output;
};

此代码是可操作的，因此在创建消息后，我们将使用void handle_write(const boost::system::error_code& error, size_t bytes_transferred)函数.打印bytes_transferred_值将返回0:服务器(基于

This code is operational, so after creating the message we fall into the void handle_write(const boost::system::error_code& error, size_t bytes_transferred) function. Printing the bytes_transferred_ value returns 0: server (implemented on the base of examples too) recieves nothing.

调试功能void print_buffers_condition(const char *step)的使用暗示了消息在通过不同的缓冲对象的堆栈传输过程中的消息丢失:

The usage of the debugging function void print_buffers_condition(const char *step) hints at loss of message during its transmission through a stack of different buffering objects:

    $ ./client 127.0.0.1 5000
-----------------------------SENDING-----------------------------

Buffer conditions after handle handshake
boost::asio::streambuf      b: 0
google::protobuf::io::OstreamOutputStream raw_output: 8192
google::protobuf::io::CodedOutputStream coded_output: 0

Debugged: 
query: "qwerty"
code: 12345

Buffer conditions after before serialization
boost::asio::streambuf      b: 0
google::protobuf::io::OstreamOutputStream raw_output: 8192
google::protobuf::io::CodedOutputStream coded_output: 0

serializing success

Buffer conditions after before async write
boost::asio::streambuf      b: 0
google::protobuf::io::OstreamOutputStream raw_output: 8192
google::protobuf::io::CodedOutputStream coded_output: 13


Buffer conditions after after async write
boost::asio::streambuf      b: 0
google::protobuf::io::OstreamOutputStream raw_output: 8192
google::protobuf::io::CodedOutputStream coded_output: 13

 bytes_trransferred: 0

我不知道如何以适当的方式做到这一点. 操作系统是RHEL 6.4. 谢谢你.

I have no idea how to do it in a proper way. OS is RHEL 6.4. Thank you.

推荐答案

我不熟悉asio，但在我看来，问题在于您没有刷新缓冲区.数据卡在CodedOutputStream中，永远找不到进入asio的路径.

I'm not familiar with asio, but it looks to me like the problem is that you aren't flushing your buffers. The data is stuck in CodedOutputStream and never finds its way into asio.

CodedOutputStream应该分配在堆栈上，这样，一旦您写完消息，它就会被销毁.析构函数将刷新缓冲区.请注意，CodedOutputStream的分配很便宜，因此将其放在堆栈上没有性能问题(实际上，这样做可能会更好).

CodedOutputStream should be allocated on the stack, such that it is destroyed as soon as you're done writing the message. The destructor will flush the buffer. Note that CodedOutputStream is cheap to allocate so there's no performance problem with putting it on the stack (in fact, it's probably better that way).

OstreamOutputStream可以类似地在堆栈上分配，但是它堆分配了您可能要重用的缓冲区.如果选择重用同一对象，请确保在CodedOutputStream被销毁后调用Flush()刷新缓冲区.

OstreamOutputStream can similarly be allocated on the stack, but it heap-allocates a buffer which you might want to reuse. If you choose to reuse the same object, make sure to call Flush() to flush the buffer after the CodedOutputStream is destroyed.

顺便说一句，OstreamOutputStream并不是特别有效，因为它必须在ostream已经执行的操作之上做自己的缓冲层.您可能需要序列化为字符串(str = message.SerializeAsString()或message.SerializeToString(&str))，然后将其直接写入套接字(如果asio允许这样做)，因为它可能避免冗余副本.

Incidentally, OstreamOutputStream is not particularly efficient, because it has to do its own layer of buffering on top of what ostream is already doing. You may want to serialize to a string (str = message.SerializeAsString() or message.SerializeToString(&str)) and then write that directly to the socket (if asio allows this), as it will probably avoid a redundant copy.

这篇关于Google :: protobuf + boost :: asio失败的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！