我有一个 TCP 客户端，在 tcp 套接字上调用常规连接可以正常工作。但是，对 async_connect 的调用永远不会触发处理程序。他们都使用几乎相同的代码。区别只是调用 connect 与 async_connect。
标题
#ifndef TCPCLIENT_H
#define TCPCLIENT_H
#include <boost/bind.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/enable_shared_from_this.hpp>
#include <boost/asio.hpp>
#include <boost/thread/thread.hpp>
#include <boost/thread/mutex.hpp>
using boost::asio::ip::tcp;
//How about an interface for outputting errors
class BoostTCPDebugOutputInterface
{
public:
virtual void outputError(std::string& error)=0;
virtual void outputDebug(std::string& debug) = 0;
virtual void outputWarning(std::string& warning) = 0;
virtual void outputInfo(std::string& info) = 0;
};
class ShutdownCallback
{
public:
virtual void shutdown() = 0;
};
class BoostTCPConnection
: public boost::enable_shared_from_this<BoostTCPConnection>
{
public:
typedef boost::shared_ptr<BoostTCPConnection> Connection_ptr;
static Connection_ptr create(boost::asio::io_service& io_service, BoostTCPDebugOutputInterface* output, ShutdownCallback* shutdowner)
{
return Connection_ptr(new BoostTCPConnection(io_service, output, shutdowner));
}
virtual ~BoostTCPConnection(void);
void start();
bool isRunning();
void setRunning(bool running);
void enqueueMessageToSend(boost::shared_ptr<std::string>& message);
boost::asio::ip::tcp::socket& socket()
{
return socket_;
}
void getQueueSize(unsigned long& tcpToSend, unsigned long& tcpRead);
void getMessages(std::list< boost::shared_ptr<std::string> >& messages);
protected:
BoostTCPConnection(boost::asio::io_service& io_service, BoostTCPDebugOutputInterface* output, ShutdownCallback* shutdowner) : socket_(io_service), errorOutput_(output), shutdowner_(shutdowner) {}
void send();
void handleWrite(const boost::system::error_code& error,size_t bytes_transferred);
void handleReceive(const boost::system::error_code& error,size_t bytes_transferred);
void readMessage();
void handleReceive();
std::string hostIP_;
unsigned int hostPort_;
boost::asio::io_service io_service_;
boost::asio::ip::tcp::socket socket_;
boost::mutex runningMutex_;
bool running_;
boost::mutex readMutex_;
std::list< boost::shared_ptr<std::string> > receivedMsgs_;
boost::mutex sendMutex_;
std::list< boost::shared_ptr<std::string> > sendMsgs_;
BoostTCPDebugOutputInterface* errorOutput_;
ShutdownCallback* shutdowner_;
static const size_t HEADERSIZE = 4;
};
class TCPClient
{
public:
TCPClient();
virtual ~TCPClient();
bool start(std::string& hostIP, unsigned int& hostPort, BoostTCPDebugOutputInterface* output, ShutdownCallback* shutdowner);
void stop();
void getQueueSize(unsigned long& tcpToSend, unsigned long& tcpRead);
void send(boost::shared_ptr<std::string>& message);
void getMessages(std::list< boost::shared_ptr<std::string> >& messages);
protected:
void threadAction();
void handleConnect(const boost::system::error_code& error);
BoostTCPConnection::Connection_ptr connection_;
boost::thread_group worker_threads_;
boost::asio::io_service io_service_;
BoostTCPDebugOutputInterface* errorOutput_;
ShutdownCallback* shutdowner_;
};
#endif
CPP 文件
#include "TCPClient.h"
#include <iostream>
BoostTCPConnection::~BoostTCPConnection(void)
{
}
void BoostTCPConnection::start()
{
setRunning(true);
while (isRunning())
{
bool readData(false);
bool wroteData(false);
if (!socket_.is_open())
{
std::string info("BoostTCPConnection::start() socket is no longer open. ");
errorOutput_->outputError(info);
shutdowner_->shutdown();
//Stop this NOW!!!
}
//Check if there are 4 bytes for packet size
//If there are read the size and then do an read to get the packet
//The handler function should put the packet on a queue.
boost::asio::socket_base::bytes_readable command(true);
socket_.io_control(command);
std::size_t bytes_readable = command.get();
if ( bytes_readable >= HEADERSIZE )
{
readMessage();
readData=true;
}
size_t sendSize(0);
{
boost::mutex::scoped_lock(sendMutex_);
sendSize = sendMsgs_.size();
}
if ( sendSize > 0)
{
send();
}
if ( !readData && !wroteData )
boost::this_thread::sleep(boost::posix_time::milliseconds(5));
}
}
void BoostTCPConnection::readMessage()
{
size_t messageSize(0);
char temp[4]="";
std::vector<char> header(4);
boost::system::error_code ec;
//Read the header which is the size
size_t read=boost::asio::read(socket_, boost::asio::buffer(header), ec);
if (ec)
{
std::string info("BoostTCPConnection::readMessage errorcode ");
info+=ec.message();
errorOutput_->outputError(info);
shutdowner_->shutdown();
//TODO Signal the GUI to stop
return;
}
memcpy((void*)(&messageSize),(void*)header.data(),4);
std::vector<char> rcvBuffer(messageSize);
read=boost::asio::read(socket_, boost::asio::buffer(rcvBuffer),ec);
if (ec)
{
std::string info("BoostTCPConnection::readMessage errorcode ");
info+=ec.message();
errorOutput_->outputError(info);
shutdowner_->shutdown();
//TODO Signal the GUI to stop
return;
}
rcvBuffer.push_back('\0');
std::string test(rcvBuffer.begin(),rcvBuffer.end());
boost::shared_ptr<std::string> message(new std::string(rcvBuffer.begin(),rcvBuffer.end()));
receivedMsgs_.push_back(message);
}
void BoostTCPConnection::getMessages(std::list< boost::shared_ptr<std::string> >& messages)
{
if (messages.size() > 0)
messages.clear();
{
boost::mutex::scoped_lock lock(readMutex_);
receivedMsgs_.swap(messages);
}
}
void BoostTCPConnection::handleReceive(const boost::system::error_code& error, size_t bytes_transferred)
{
if (error)
{
std::ostringstream oss;
oss<< "BoostTCPConnection::handleReceive got an error Code of "<<error.value()<<" and message "<<error.message()<<" bytes_transferred = "<<bytes_transferred<<std::endl;
errorOutput_->outputError(oss.str());
shutdowner_->shutdown();
return;
}
}
bool BoostTCPConnection::isRunning()
{
boost::mutex::scoped_lock lock(runningMutex_);
return running_;
}
void BoostTCPConnection::setRunning(bool running)
{
boost::mutex::scoped_lock lock(runningMutex_);
running_=running;
}
void BoostTCPConnection::enqueueMessageToSend(boost::shared_ptr<std::string>& message)
{
boost::mutex::scoped_lock lock(sendMutex_);
sendMsgs_.push_back(message);
}
void BoostTCPConnection::getQueueSize(unsigned long& tcpToSend, unsigned long& tcpRead)
{
{
boost::mutex::scoped_lock lock(sendMutex_);
tcpToSend=sendMsgs_.size();
}
{
boost::mutex::scoped_lock lock(readMutex_);
tcpRead=receivedMsgs_.size();
}
}
void BoostTCPConnection::send()
{
if (sendMsgs_.empty())
return;
boost::shared_ptr<std::string> message;
{
message=sendMsgs_.front();
sendMsgs_.pop_front();
}
char temp[4];
size_t messageSize=message->size();
memcpy(temp,&messageSize, 4);
message->insert(0,temp, 4);
boost::asio::async_write(socket_, boost::asio::buffer(*message),
boost::bind(&BoostTCPConnection::handleWrite, shared_from_this(),
boost::asio::placeholders::error,
boost::asio::placeholders::bytes_transferred));
}
void BoostTCPConnection::handleWrite(const boost::system::error_code& error,size_t bytes_transferred)
{
//Success
if (error.value() == 0)
return;
std::ostringstream oss;
oss<< "BoostTCPConnection::handleWrite got an error Code of "<<error.value()<<" and message "<<error.message()<<" with bytes_transferred = "<<bytes_transferred<<std::endl;
errorOutput_->outputError(oss.str());
shutdowner_->shutdown();
}
//***************************************************
// TCPClient
//***************************************************
TCPClient::TCPClient()
{
}
TCPClient::~TCPClient()
{
}
void TCPClient::threadAction()
{
io_service_.run();
}
bool TCPClient::start(std::string& hostIP, unsigned int& hostPort, BoostTCPDebugOutputInterface* output, ShutdownCallback* shutdowner)
{
bool bResult(false);
errorOutput_=output;
connection_ = BoostTCPConnection::create(io_service_, output, shutdowner);
shutdowner_=shutdowner;
//Use multiple threads to do my bidding
for( int x = 0; x < 3; ++x )
{
worker_threads_.create_thread(
boost::bind( &TCPClient::threadAction, this )
);
}
boost::system::error_code ec;
try
{
boost::asio::ip::tcp::endpoint ep( boost::asio::ip::address_v4::from_string(hostIP), hostPort);
if (connection_)
{
connection_->socket().async_connect(ep,
boost::bind(&TCPClient::handleConnect,
this, boost::asio::placeholders::error));
//Synchronous code that works fine
//boost::system::error_code ec;
//connection_->socket().connect(ep, ec);
/*if (!ec)
{
worker_threads_.create_thread(
boost::bind( &TCPClient::handleConnect, this, ec ));
bResult=true;
}
else
{
std::ostringstream oss;
oss<< "BoostTCPConnection::start has an error "<<ec.message()<<std::endl;
errorOutput_->outputError(oss.str());
}*/
}
}
catch (std::exception& e)
{
std::ostringstream oss;
oss<< "BoostTCPConnection::start received the exception "<<e.what()<<std::endl;
errorOutput_->outputError(oss.str());
}
return bResult;
}
void TCPClient::handleConnect(const boost::system::error_code& error)
{
if (error )
{
std::ostringstream oss;
oss<< "BoostTCPConnection::handleConnect received the error "<<error.message()<<std::endl;
errorOutput_->outputError(oss.str());
shutdowner_->shutdown();
return;
}
if (connection_)
connection_->start();
}
void TCPClient::stop()
{
if (connection_)
connection_->setRunning(false);
io_service_.stop();
worker_threads_.join_all();
}
void TCPClient::getQueueSize(unsigned long& tcpToSend, unsigned long& tcpRead)
{
if (connection_)
connection_->getQueueSize(tcpToSend, tcpRead);
else
{
tcpToSend=0;
tcpRead=0;
}
}
void TCPClient::send(boost::shared_ptr<std::string>& message)
{
if (connection_)
connection_->enqueueMessageToSend(message);
}
void TCPClient::getMessages(std::list< boost::shared_ptr<std::string> >& messages)
{
if (connection_)
connection_->getMessages(messages);
}