TCPClient boost :: asio :: io_service post not firing

我遇到了boost :: asio :: io_service.post（）没有调用我的方法处理程序的问题。我有一个简单的客户端和服务器C++应用程序都使用TCPClient类中的相同的代码。客户端工作正常，但使用接受填充的类的实例不起作用。TCPClient boost :: asio :: io_service post not firing

我已经把整个项目放到了here，但我已经把相关的代码放在下面。

在的TcpClient :: Write方法这行

io_service.post(boost::bind(&TCPClient::DoWrite, this, msg));

被调用，但处理程序（的TcpClient :: DoWrite）不会被调用服务器端。

我知道IO_Service正在运行，因为我的同步TCPClient中的async_reads正常工作。

这是我的TcpClient类

.HPP文件

class TCPClient 
    : public boost::enable_shared_from_this<TCPClient> 
{ 
    public: 
     typedef boost::shared_ptr<TCPClient> pointer; 

    private: 
     boost::asio::io_service io_service; 

     bool m_IsConnected; 
     bool m_HeartbeatEnabled; 

     boost::asio::ip::tcp::socket m_Socket; 
     boost::asio::ip::tcp::endpoint m_Endpoint; 

     boost::asio::steady_timer m_HeartBeatTimer; 
     boost::asio::streambuf m_Buffer; 
     std::string m_Delimiter; 
     std::deque<std::string> m_Messages; 
     bool m_HeartBeatEnabled; 
     int m_HeartBeatTime; 

    private: 
     void HandleConnect(const boost::system::error_code& error); 
     void DoHeartBeat(const boost::system::error_code& error); 
     void DoWrite(const std::string &msg); 
     void HandleWrite(const boost::system::error_code& error); 
     void HandleRead(const boost::system::error_code& error); 

    public: 
     TCPClient(boost::asio::io_service &io_service); 
     TCPClient(bool enableHeartbeat); 
     ~TCPClient(); 
     void Close(); 
     void ConnectToServer(boost::asio::ip::tcp::endpoint& endpoint); 
     void ConnectToServer(const std::string &ip, const std::string &protocol); 
     void ConnectToServer(const std::string &ip, unsigned short port); 
     void Write(const std::string &msg); 
     void StartRead(); 
     void SetHeartBeatTime(int time); 
     boost::asio::ip::tcp::socket& Socket(); 
     boost::asio::io_service& Service(); 
     static pointer Create(boost::asio::io_service& io_service); 

    public: 
     // signals 
     boost::signals2::signal<void(const boost::asio::ip::tcp::endpoint&)> sConnected; 
     boost::signals2::signal<void(const boost::asio::ip::tcp::endpoint&)> sDisconnected;  
     boost::signals2::signal<void(const std::string&)>      sMessage; 
};

.cpp文件

using boost::asio::ip::tcp; 

TCPClient::pointer TCPClient::Create(boost::asio::io_service& io) 
{ 
    return pointer(new TCPClient(io)); 
} 

TCPClient::TCPClient(boost::asio::io_service& io) 
    : m_IsConnected(true), m_Socket(io), m_HeartBeatTimer(io), m_Delimiter(), m_HeartBeatTime(10) 
{ 
    m_Delimiter = "\n"; 
    m_HeartbeatEnabled = false; 
    // start heartbeat timer (optional) 
    if(m_HeartBeatEnabled) 
    { 
     m_HeartBeatTimer.expires_from_now(boost::chrono::seconds(m_HeartBeatTime)); 
     m_HeartBeatTimer.async_wait(boost::bind(&TCPClient::DoHeartBeat, this, boost::asio::placeholders::error)); 
    } 
} 

TCPClient::TCPClient(bool enableHeartBeat) 
    : m_IsConnected(false), m_Socket(io_service), m_HeartBeatTimer(io_service), m_Delimiter(), m_HeartBeatTime(10) 
{ 
    m_Delimiter = "\n"; 
    m_HeartbeatEnabled = enableHeartBeat; 
} 

TCPClient::TCPClient::~TCPClient() 
{ 
} 

void TCPClient::Close() 
{ 
    io_service.stop(); 
    m_Socket.close(); 
} 

boost::asio::ip::tcp::socket& TCPClient::Socket() 
{ 
    return m_Socket; 
} 

boost::asio::io_service& TCPClient::Service() 
{ 
    return io_service; 
} 

void TCPClient::ConnectToServer(const std::string &ip, unsigned short port) 
{ 
    try { 
     boost::asio::ip::tcp::endpoint endpoint(boost::asio::ip::address::from_string(ip), port); 
     ConnectToServer(endpoint); 
    } 
    catch(const std::exception &e) { 
     std::cout << "Error: " << e.what() << std::endl; 
    } 
} 

void TCPClient::ConnectToServer(const std::string &url, const std::string &protocol) 
{ 
    // You can also explicitly pass a port, like "8080" 
    boost::asio::ip::tcp::resolver::query query(url, protocol); 
    boost::asio::ip::tcp::resolver resolver(io_service); 
    try { 
     boost::asio::ip::tcp::resolver::iterator destination = resolver.resolve(query); 
     boost::asio::ip::tcp::endpoint endpoint; 
     while (destination != boost::asio::ip::tcp::resolver::iterator()) 
      endpoint = *destination++; 

     ConnectToServer(endpoint); 
    } 
    catch(const std::exception &e) { 
     std::cout << "Error: " << e.what() << std::endl; 
    } 
} 

void TCPClient::ConnectToServer(boost::asio::ip::tcp::endpoint& endpoint) 
{ 
    m_Endpoint = endpoint; 

    std::cout << "Trying to connect to port " << endpoint << std::endl; 

    // try to connect, then call handle_connect 
    m_Socket.async_connect(m_Endpoint, 
     boost::bind(&TCPClient::HandleConnect, this, boost::asio::placeholders::error)); 

    //start processing messages 
    io_service.run(); 
} 

void TCPClient::Write(const std::string &msg) 
{ 
    if(!m_IsConnected) return; 
    std::cout << "write: " << msg << std::endl; 
    // safe way to request the client to write a message 
    io_service.post(boost::bind(&TCPClient::DoWrite, this, msg)); 
} 

void TCPClient::StartRead() 
{ 
    if(!m_IsConnected) return; 

    // wait for a message to arrive, then call handle_read 
    boost::asio::async_read_until(m_Socket, m_Buffer, m_Delimiter, 
      boost::bind(&TCPClient::HandleRead, this, boost::asio::placeholders::error)); 
} 


void TCPClient::HandleRead(const boost::system::error_code& error) 
{ 
    if (!error) 
    { 
     std::string msg; 
     std::istream is(&m_Buffer); 
     std::getline(is, msg); 

     if(msg.empty()) return; 

     //cout << "Server message:" << msg << std::endl; 

     // TODO: you could do some message processing here, like breaking it up 
     //  into smaller parts, rejecting unknown messages or handling the message protocol 

     // create signal to notify listeners 
     sMessage(msg); 

     // restart heartbeat timer (optional) 
     if(m_HeartBeatEnabled) 
     { 
      m_HeartBeatTimer.expires_from_now(boost::chrono::seconds(m_HeartBeatTime)); 
      m_HeartBeatTimer.async_wait(boost::bind(&TCPClient::DoHeartBeat, this, boost::asio::placeholders::error)); 
     } 

     // wait for the next message 
     StartRead(); 
    } 
    else 
    { 
     // try to reconnect if external host disconnects 
     if(error.value() != 0) { 
      m_IsConnected = false; 

      // let listeners know 
      sDisconnected(m_Endpoint); 

      // cancel timers 
      m_HeartBeatTimer.cancel(); 
     } 
     //else 
      //do_close(); 
    } 
} 

void TCPClient::HandleWrite(const boost::system::error_code& error) 
{ 
    if(!error) 
    { 
     // write next message 
     m_Messages.pop_front(); 
     if (!m_Messages.empty()) 
     { 
      std::cout << "Client message:" << m_Messages.front() << std::endl; 

      boost::asio::async_write(m_Socket, 
       boost::asio::buffer(m_Messages.front()), 
       boost::bind(&TCPClient::HandleWrite, this, boost::asio::placeholders::error)); 
     } 
     if(m_HeartBeatEnabled) 
     { 
      // restart heartbeat timer (optional) 
      m_HeartBeatTimer.expires_from_now(boost::chrono::seconds(m_HeartBeatTime)); 
      m_HeartBeatTimer.async_wait(boost::bind(&TCPClient::DoHeartBeat, this, boost::asio::placeholders::error)); 
     } 
    } 
    else 
    { 
     std::cout << "HandleWrite Error: " << error << std::endl; 
    } 
} 

void TCPClient::DoWrite(const std::string &msg) 
{ 
    if(!m_IsConnected) return; 

    bool write_in_progress = !m_Messages.empty(); 
    m_Messages.push_back(msg + m_Delimiter); 

    if (!write_in_progress) 
    { 
     std::cout << "Client message2: " << m_Messages.front() << std::endl; 

     boost::asio::async_write(m_Socket, 
      boost::asio::buffer(m_Messages.front()), 
      boost::bind(&TCPClient::HandleWrite, this, boost::asio::placeholders::error)); 
    } 
    else 
    { 
     std::cout << "DoWrite write_in_progress: " << msg << std::endl; 
    } 
} 

void TCPClient::HandleConnect(const boost::system::error_code& error) 
{ 
    if (!error) { 
     // we are connected! 
     m_IsConnected = true; 

     // let listeners know 
     sConnected(m_Endpoint); 

     // start heartbeat timer (optional) 
     if(m_HeartBeatEnabled) 
     { 
      m_HeartBeatTimer.expires_from_now(boost::chrono::seconds(m_HeartBeatTime)); 
      m_HeartBeatTimer.async_wait(boost::bind(&TCPClient::DoHeartBeat, this, boost::asio::placeholders::error)); 
     } 

     // await the first message 
     StartRead(); 
    } 
    else { 
     // there was an error :(
     m_IsConnected = false; 

     std::cout << "Server error:" << error.message() << std::endl; 
    } 
} 

void TCPClient::DoHeartBeat(const boost::system::error_code& error) 
{ 
    // here you can regularly send a message to the server to keep the connection alive, 
    // I usualy send a PING and then the server replies with a PONG 
    if(!error) Write("PING"); 
} 

void TCPClient::SetHeartBeatTime(int time) 
{ 
    m_HeartBeatTime = time; 
    m_HeartBeatEnabled = true; 
    m_HeartBeatTimer.expires_from_now(boost::chrono::seconds(m_HeartBeatTime)); 
    m_HeartBeatTimer.async_wait(boost::bind(&TCPClient::DoHeartBeat, this, boost::asio::placeholders::error)); 
}

使用我TCPSERVER接受连接

.HPP文件

class TCPServer 
{ 
    private: 
     boost::asio::io_service io_service; 
     boost::asio::ip::tcp::acceptor m_acceptor; 

    public: 
     TCPServer(int port); 
     ~TCPServer(); 
     void Close(); 
     void StartAccept(); 

    private: 
     void HandleAccept(TCPClient::pointer new_connection, const boost::system::error_code& error); 

    public: 
     boost::signals2::signal<void(const TCPClient::pointer&)> sig_NewClient; 
};

.cpp文件

TCPServer::TCPServer(int port) 
    : m_acceptor(io_service, boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), port)) 
{ 
} 

TCPServer::TCPServer::~TCPServer() 
{ 
} 

void TCPServer::Close() 
{ 
    m_acceptor.close(); 
    io_service.stop(); 
} 

void TCPServer::StartAccept() 
{ 
    TCPClient::pointer new_connection = TCPClient::Create(io_service); 

    m_acceptor.async_accept(new_connection->Socket(), 
     boost::bind(&TCPServer::HandleAccept, this, new_connection, boost::asio::placeholders::error)); 

    io_service.run(); 

    std::cout << "Run ended for server " << std::endl; 
} 

void TCPServer::HandleAccept(TCPClient::pointer new_connection, const boost::system::error_code& error) 
{ 
    if (!error) 
    { 
     sig_NewClient(new_connection); 
     StartAccept(); 
    } 
}

我很新的提高，不要做与C++（正常的C＃，JAVA等）的工作太多了，所以我认为我失去了一些东西根本，但我不能找到问题。

须藤流
服务器
创建TCPSERVER
服务器 - StartAccept（）
在新的连接呼叫StartRead上产生的TcpClient实例
时收到ELLO写欧莱
时收到PING写PONG

Cl ient
连接到服务器
发送ELLO
发送Ping每10秒

客户接收并写入网络精细 服务器接受罚款，但在写永远不会使其DoWrite或HandleWrite方法

任何额外的信息，请让我知道。

在此先感谢

来源

2014-10-10 Lepon

有一些问题，他们中的一些，我可以看到：

既然你没有io_service::worker您io_service.run()在没有活跃的处理程序将停止。
你TCPClient::Write试图post()套接字写了一份工作，但它提到过std::string，所以当你的TCPClient::DoWrite将被调用，您的数据已经可以销毁。

有一些基本的C++和boost :: asio使用问题，所以我认为它的价值从更简单的实现开始。

来源

2014-10-11 11:10:32 PSIAlt

为写链的缓冲寿命看起来还给我。在'TCPClient :: Write（）'中，引用将被值复制到'boost :: bind（）'函子中。 'TCPClient :: DoWrite（）'CompletionHandler引用复制引用，并将其复制到deque中，使用缓冲区的deque元素启动'async_write（）'操作。底层内存然后在'async_write（）'完成处理程序（'TCPClient :: HandleWrite（）'）中被释放。 – 2014-10-11 16:52:58

有我的理解，因为我有一个async_accept（）之前调用io_service.run（）和句柄再次接受calles async_accept io_service应该始终有工作要做？看着绑定的doco它说：“绑定的参数被复制并由内部返回的函数对象保存。”这意味着即使我的原始对象超出了范围，并且销毁绑定中使用的值将是一个副本，因此仍然有效？ – Lepon 2014-10-12 00:30:49

的TCPServer接受时它当前服务于相同io_service对象上的事件循环中的线程内调用io_service.run()调用链违反一个io_service要求导致未定义的行为。该documentation状态：

的run()功能不能从当前调用相同io_service对象上的run()之一，run_one()，poll()或poll_one()一个线程调用。

在TCPServer代码，要求被违反时HandleAccept()完成处理，内io_service.run()由一个线程调用，调用StartAccept()，然后将在同一io_service调用io_service.run()：

   .------------------------------------. 
       V         | 
void TCPServer::StartAccept()      | 
{             | 
    m_acceptor.async_accept(..., &HandleAccept); --. | 
    io_service.run();        | | 
}             | | 
       .---------------------------------' | 
       V         | 
void TCPServer::HandleAccept(...)     | 
{             | 
    if (!error)          | 
    {             | 
    StartAccept(); ---------------------------------' 
    } 
}

要解决这个，不要在完成处理程序中调用io_service.run()。相反，考虑增加一个入口点发起接受呼叫链和运行io_service，但不是异步调用链环的一部分：

void TCPServer::StartAccept() 
{ 
    DoStartAccept(); ---------------------------------. 
    io_service.run();         | 
}             | 
     .---------------------------------------------' 
     |  .------------------------------------. 
     V  V         | 
void TCPServer::DoStartAccept()      | 
{             | 
    m_acceptor.async_accept(..., &HandleAccept); --. | 
}             | | 
       .---------------------------------' | 
       V         | 
void TCPServer::HandleAccept(...)     | 
{             | 
    if (!error)          | 
    {             | 
    DoStartAccept(); -------------------------------' 
    } 
}

来源

2014-10-11 17:23:02

谢谢，我修正了建议的代码。虽然这确实修复了一个错误，但我仍然有写入问题。 – Lepon 2014-10-12 00:24:26

TCPClient boost :: asio :: io_service post not firing

回答

相关问题