TCPClient boost :: asio :: io_service post not firing

Question

我遇到了boost :: asio :: io_service.post（）沒有調用我的方法處理程序的問題。 我有一個簡單的客戶端和服務器C++應用程序都使用TCPClient類中的相同的代碼。客戶端工作正常，但使用接受填充的類的實例不起作用。

我已經把整個項目放到了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方法

任何額外的信息，請讓我知道。

在此先感謝

Answer 1

有一些問題，他們中的一些，我可以看到：

• 既然你沒有io_service::worker您io_service.run()在沒有活躍的處理程序將停止。

• 你TCPClient::Write試圖post()套接字寫了一份工作，但它提到過std::string，所以當你的TCPClient::DoWrite將被調用，您的數據已經可以銷燬。

有一些基本的C++和boost :: asio使用問題，所以我認爲它的價值從更簡單的實現開始。

Answer 2

的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(); -------------------------------' 
    } 
}