c++ - async_send 数据未发送

Question

[免责声明] 我是 boost 新手。

研究 boost::asio 并尝试创建一个具有以下功能的简单异步 TCP 服务器:

1. 监听端口 13 上的连接
2. 连接后，接收数据
3. 如果收到的数据 == 时间，则返回当前日期时间，否则返回预定义的字符串(“请求其他内容”)

问题:虽然，我接受连接并接收数据，但在使用 async_send 传输数据时，虽然我没有收到任何错误并且 bytes_transferred 的值是正确的，但我在客户端收到空数据。

如果我尝试从 handle_accept(而不是 handle_read)中传输数据，这工作正常。

实现:我研究了发现的 boost asio 教程 here :实例化一个 tcp_server 对象，它基本上启动接受器并开始监听。如下图:

int main()
{
    try
    {
        boost::asio::io_service io_service;
        tcp_server server(io_service);
        io_service.run();
    }
    catch (std::exception& e)
    {
        std::cerr << e.what() << std::endl;
    }

    return 0;
}

在 tcp_server 中:

class tcp_server
{
public:
    tcp_server(boost::asio::io_service& io_service)
        : acceptor_(io_service, tcp::endpoint(tcp::v4(), 13))
    {
        start_accept();
    }

private:
    void start_accept()
    {
        using std::cout;
        tcp_connection::pointer new_connection =
            tcp_connection::create(acceptor_.get_io_service());

        acceptor_.async_accept(new_connection->socket(),
            boost::bind(&tcp_server::handle_accept, this, new_connection,
                boost::asio::placeholders::error));
        cout << "Done";
    }
    ...
}

接受连接后，我将按如下所示进行处理:

void handle_accept(tcp_connection::pointer new_connection,
        const boost::system::error_code& error)
{
    if (!error)
    {
        new_connection->start();
    }

    start_accept();
}

下面是tcp_connection::start()方法:

void start()
{

    boost::asio::async_read(socket_, boost::asio::buffer(inputBuffer_),
        boost::bind(&tcp_connection::handle_read, shared_from_this(),
            boost::asio::placeholders::error,
            boost::asio::placeholders::bytes_transferred));

    /* the snippet below works here - but not in handle_read 
    outputBuffer_ = make_daytime_string();

    boost::asio::async_write(socket_, boost::asio::buffer(outputBuffer_),
        boost::bind(&tcp_connection::handle_write, shared_from_this(),
            boost::asio::placeholders::error,
            boost::asio::placeholders::bytes_transferred));*/
}

在handle_read中:

void handle_read(const boost::system::error_code& error, size_t bytes_transferred)
{
    outputBuffer_ = make_daytime_string();
    if (strcmp(inputBuffer_, "time"))
    {
        /*this does not work - correct bytes_transferred but nothing shown on receiving end */
        boost::asio::async_write(socket_, boost::asio::buffer(outputBuffer_),
            boost::bind(&tcp_connection::handle_write, shared_from_this(),
                boost::asio::placeholders::error,
                boost::asio::placeholders::bytes_transferred)); 
    }
    else
    {
        outputBuffer_ = "Something else was requested";//, 128);
        boost::asio::async_write(socket_, boost::asio::buffer(outputBuffer_),
            boost::bind(&tcp_connection::handle_write, shared_from_this(),
                boost::asio::placeholders::error,
                boost::asio::placeholders::bytes_transferred));
    }
}

handle_write 如下所示:

void handle_write(const boost::system::error_code& error,
    size_t bytes_transferred)
{
    if (!error)
    {
        std::cout << "Bytes transferred: " << bytes_transferred;
        std::cout << "Message sent: " << outputBuffer_;
    }
    else
    {
        std::cout << "Error in writing: " << error.message();
    }
}

请注意以下关于 handle_write 的内容(这真的很奇怪):

• 没有错误
• bytes_transferred 变量具有正确的值
• outputBuffer_ 具有正确的值(在 handle_read 中设置)

然而，客户端收到的包 ( Packet Sender ) 是空的(就数据而言)。

完整代码分享here .

Answer 1

完整的测试程序(c++14)。请注意响应接收时异步缓冲的处理 - 可能已经在进行发送。

#include <boost/asio.hpp>
#include <thread>
#include <future>
#include <vector>
#include <array>
#include <memory>
#include <mutex>
#include <condition_variable>
#include <iterator>
#include <iostream>

namespace asio = boost::asio;

asio::io_service        server_service;
asio::io_service::work  server_work{server_service};

bool listening = false;
std::condition_variable cv_listening;
std::mutex              management_mutex;

auto const shared_query = asio::ip::tcp::resolver::query(asio::ip::tcp::v4(), "localhost", "8082");

void client()
try
{
    asio::io_service      client_service;
    asio::ip::tcp::socket socket(client_service);

    auto lock = std::unique_lock<std::mutex>(management_mutex);
    cv_listening.wait(lock, [] { return listening; });
    lock.unlock();

    asio::ip::tcp::resolver resolver(client_service);
    asio::connect(socket, resolver.resolve(shared_query));
    auto s = std::string("time\ntime\ntime\n");
    asio::write(socket, asio::buffer(s));
    socket.shutdown(asio::ip::tcp::socket::shutdown_send);

    asio::streambuf sb;
    boost::system::error_code sink;
    asio::read(socket, sb, sink);
    std::cout << std::addressof(sb);
    socket.close();
    server_service.stop();
}
catch(const boost::system::system_error& se)
{
    std::cerr << "client: " << se.code().message() << std::endl;
}

struct connection
    : std::enable_shared_from_this<connection>
{
    connection(asio::io_service& ios)
        : strand_(ios)
    {

    }

    void run()
    {
        asio::async_read_until(socket_, buffer_, "\n",
                               strand_.wrap([self = shared_from_this()](auto const&ec, auto size)
        {
            if (size == 0 )
            {
                // error condition
                boost::system::error_code sink;
                self->socket_.shutdown(asio::ip::tcp::socket::shutdown_receive, sink);
            }
            else {
                self->buffer_.commit(size);
                std::istream is(std::addressof(self->buffer_));
                std::string str;
                while (std::getline(is, str))
                {
                    if (str == "time") {
                        self->queue_send("eight o clock");
                    }
                }
                self->run();
            }
        }));
    }

    void queue_send(std::string s)
    {
        assert(strand_.running_in_this_thread());
        s += '\n';
        send_buffers_pending_.push_back(std::move(s));
        nudge_send();
    }

    void nudge_send()
    {
        assert(strand_.running_in_this_thread());
        if (send_buffers_sending_.empty() and not send_buffers_pending_.empty())
        {
            std::swap(send_buffers_pending_, send_buffers_sending_);
            std::vector<asio::const_buffers_1> send_buffers;
            send_buffers.reserve(send_buffers_sending_.size());
            std::transform(send_buffers_sending_.begin(), send_buffers_sending_.end(),
                           std::back_inserter(send_buffers),
            [](auto&& str) {
                return asio::buffer(str);
            });
            asio::async_write(socket_, send_buffers,
                              strand_.wrap([self = shared_from_this()](auto const& ec, auto size)
            {
                // should check for errors here...
                self->send_buffers_sending_.clear();
                self->nudge_send();
            }));
        }
    }

    asio::io_service::strand strand_;
    asio::ip::tcp::socket    socket_{strand_.get_io_service()};
    asio::streambuf          buffer_;

    std::vector<std::string> send_buffers_pending_;
    std::vector<std::string> send_buffers_sending_;
};

void begin_accepting(asio::ip::tcp::acceptor& acceptor)
{
    auto candidate = std::make_shared<connection>(acceptor.get_io_service());
    acceptor.async_accept(candidate->socket_, [candidate, &acceptor](auto const& ec)
    {
        if (not ec) {
            candidate->run();
            begin_accepting(acceptor);
        }
    });
}

void server()
try
{
    asio::ip::tcp::acceptor acceptor(server_service);
    asio::ip::tcp::resolver resolver(server_service);

    auto first = resolver.resolve(shared_query);
    acceptor.open(first->endpoint().protocol());
    acceptor.bind(first->endpoint());

    acceptor.listen();

    begin_accepting(acceptor);

    auto lock = std::unique_lock<std::mutex>(management_mutex);
    listening = true;
    lock.unlock();
    cv_listening.notify_all();


    server_service.run();

}
catch(const boost::system::system_error& se)
{
    std::cerr << "server: " << se.code().message() << std::endl;
}

int main()
{

    using future_type = std::future<void>;

    auto stuff = std::array<future_type, 2> {{std::async(std::launch::async, client),
                                                 std::async(std::launch::async, server)}};

    for (auto& f : stuff) f.wait();

}