.h
/************************************************************** * Filename: TcpClient.h * Copyright: Shanghai X Co., Ltd. * * Description: TcpClient头文件. * * @author: w * @version 10/28/2016 @Reviser Initial Version **************************************************************/ #ifndef _TCPCLIENT_ #define _TCPCLIENT_ #include <string> #include <ace/Svc_Handler.h> #include <ace/Connector.h> #include <ace/SOCK_Connector.h> #include <ace/Task.h> using namespace std; //连接状态改变时回调 typedef void (__stdcall *pfnConnectChangeCallBack)(bool); //接收到数据时回调 typedef void (__stdcall *pfnReceiveCallBack)(char*, const int); class CTcpClient : public ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_MT_SYNCH> { public: // 是否退出的标识 long m_lStop; public: // 是否允许重连 bool m_nReconnect; // 通信超时ms int m_nCommunicateTimeOut; public: //typedef ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_MT_SYNCH> Base; CTcpClient(); virtual ~CTcpClient(); public: /** * 设置连接参数. * * @param -[in] char* szHost: [主链接地址]; * @param -[in] char* szBackup: [备连接地址]; * @param -[in] int nRemotePort: [目标端口号]; * @param -[in] int nLocalPort: [本地端口号]; * @return int. * @version 10/28/2016 w Initial Version */ virtual long SetConnectParam(char* szHost, char* szBackup, int nRemotePort, int nLocalPort = 0); /** * 首次连接. * * @return int. * @version 10/28/2016 w Initial Version */ virtual int Connect(); /** * 断开连接. * * @return int. * @version 10/28/2016 w Initial Version */ virtual int Reconnect(); /** * 断开连接. * * @return int. * @version 10/28/2016 w Initial Version */ virtual int Disconnect(); /** * 发送数据. * * @param -[in,out] char* szSend: [数据] * @param -[in] char* lSendSize: [大小] * @param -[in] int nCommunicateTimeOut: [超时ms] * @return int. * @version 10/28/2016 w Initial Version */ virtual long Send(const char* szSend, long lSendSize, int nCommunicateTimeOut = COMMUNICATE_TIMEOUT); /** * 接收数据. * * @param -[in,out] char* szReceive: [数据] * @param -[in] long lReceiveSize: [大小] * @param -[in] int nCommunicateTimeOut: [超时ms] * @version 10/28/2016 w Initial Version */ virtual long Receive(char* szReceive, long lReceiveSize, int nCommunicateTimeOut = COMMUNICATE_TIMEOUT); // virtual bool IsConnected() { return m_nIsConnected; } /** * 设置连接改变回调函数. * * @version 10/28/2016 w Initial Version */ void SetOnConnectChangeCallBack(pfnConnectChangeCallBack func); /** * 设置数据接收回调函数. * * @version 10/28/2016 w Initial Version */ void SetOnReceiveCallBack(pfnReceiveCallBack func); public: /** * 建立连接时被调用. * * @param -[in,out] char* param: [参数] * @return int. * @version 10/28/2016 w Initial Version */ int open(void* param = 0); /** * 当有输入时该函数被调用. * * @param -[in] ACE_HANDLE: [参数] * @return int. * @version 10/28/2016 w Initial Version */ int handle_input(ACE_HANDLE handle = ACE_INVALID_HANDLE); /** * 当有输出时该函数被调用. * * @param -[in] ACE_HANDLE handle: [参数] * @return int. * @version 10/28/2016 w Initial Version */ virtual int handle_output(ACE_HANDLE handle = ACE_INVALID_HANDLE); /** * 当SockHandler从ACE_Reactor中移除时该函数被调用. * * @param -[in] ACE_HANDLE handle: [参数] * @param -[in] ACE_HANDLE closeMask: [参数] * @return int * @version 10/28/2016 w Initial Version */ int handle_close(ACE_HANDLE handle, ACE_Reactor_Mask closeMask); /** * 任务的主流程. * 1.激活事件 * * @return int. * @version 10/10/2016 w Initial Version */ int svc(); protected: /** * 触发连接改变回调函数. * * @param -[in] bool nIsConnected: [是否已连接] * * @version 10/28/2016 w Initial Version */ void OnConnectChange(bool nIsConnected); /** * 触发数据接收回调函数. * * @param -[in,out] char* pszReceive: [接收的数据区] * @param -[in] const int nReceiveSize: [数据大小] * @version 10/28/2016 w Initial Version */ void OnReceive(char* pszReceive, const int nReceiveSize); protected: // 是否已经连接 bool m_nIsConnected; // 当前连接IP地址 string m_strConnectIPAddress; // 主线连接IP地址 string m_strHostIPAddress; // 备线连接IP地址 string m_strBackupIPAddress; // 远程连接端口号 unsigned short m_nRemotePort; // 本地连接端口号 unsigned short m_nLocalPort; pfnConnectChangeCallBack m_pfnOnConnectChange; pfnReceiveCallBack m_pfnOnReceive; protected: // 最后一次连接时间 time_t m_tmLastConnect; private: /** * 关闭Socket. * * @return int. * @version 10/28/2016 w Initial Version */ int CloseSocket(); }; //typedef ACE_Connector<CTcpClient, ACE_SOCK_CONNECTOR> CONNECTOR; #endif // !_TCPCLIENT_
.cpp
/************************************************************** * Filename: TcpClient.cpp * Copyright: Shanghai X Co., Ltd. * * Description: TcpClient源文件. * * @author: w * @version 10/28/2016 @Reviser Initial Version **************************************************************/ #include "TcpClient.h" #include <iostream> #include <string> #include <ace/ACE.h> #include <ace/OS_NS_sys_socket.h> #include <ace/OS_NS_strings.h> using namespace std; //ctor CTcpClient::CTcpClient() { m_lStop = true; m_nReconnect = true; m_nIsConnected = false; m_pfnOnConnectChange = NULL; m_pfnOnReceive = NULL; m_strConnectIPAddress = ""; m_strHostIPAddress = ""; m_strBackupIPAddress = ""; m_nRemotePort = 0; m_nLocalPort = 0; m_tmLastConnect = 0; } //dctor CTcpClient::~CTcpClient() { m_lStop = true; wait(); // close(); m_pfnOnConnectChange = NULL; m_pfnOnReceive = NULL; } long CTcpClient::SetConnectParam(char* szHost, char* szBackup, int nRemotePort, int nLocalPort) { m_strConnectIPAddress = m_strHostIPAddress = szHost; m_strBackupIPAddress = szBackup; m_nRemotePort = nRemotePort; m_nLocalPort = nLocalPort; return 0; } void CTcpClient::SetOnConnectChangeCallBack(pfnConnectChangeCallBack func) { this->m_pfnOnConnectChange = func; } void CTcpClient::SetOnReceiveCallBack(pfnReceiveCallBack func) { this->m_pfnOnReceive = func; } void CTcpClient::OnConnectChange(bool nIsConnected) { m_nIsConnected = nIsConnected; if(!m_nIsConnected) Log(LOGLEVEL_ERROR, "Disconnect from(%s:%d).", m_strConnectIPAddress.c_str(), m_nRemotePort); else Log(LOGLEVEL_NOTICE, "Connected to(%s:%d).", m_strConnectIPAddress.c_str(), m_nRemotePort); // if(m_pfnOnConnectChange) m_pfnOnConnectChange(m_nIsConnected); } void CTcpClient::OnReceive(char* pszReceive, const int nReceiveSize) { ACE_Message_Block *pFrame = new ACE_Message_Block(nReceiveSize); memcpy(pFrame->wr_ptr(), pszReceive, nReceiveSize); pFrame->wr_ptr(nReceiveSize); this->putq(pFrame); /*if(m_pfnOnReceive) m_pfnOnReceive(pszReceive, nReceiveSize);*/ delete[] pszReceive; } int CTcpClient::Disconnect() { CloseSocket(); OnConnectChange(false); return 0; } int CTcpClient::CloseSocket() { //ACE_OS::shutdown(get_handle(), ACE_SHUTDOWN_BOTH); //int nRet = ACE_OS::closesocket(m_sockHandler.get_handle()); this->peer().close(); set_handle(ACE_INVALID_HANDLE); return 0; } int CTcpClient::Reconnect() { //已连接 if (IsConnected()) return 0; //未设置重连机制 if(!m_nReconnect) { Log(LOGLEVEL_INFO, "Reconnect is disabled."); return -1; } //小于超时时间3s不能重连 time_t tmNow; time(&tmNow); if(abs(tmNow - m_tmLastConnect) <= CONNECTION_TIMEOUT) return -1; //清理Socket CloseSocket(); return Connect(); } int CTcpClient::Connect() { //与服务器建立连接 CTcpClient *pSockHandler = this; //创建连接器 ACE_Connector<CTcpClient, ACE_SOCK_CONNECTOR> connector; //设置默认连接超时 ACE_Time_Value connTimeOut(CONNECTION_TIMEOUT); ACE_Synch_Options synch_option(ACE_Synch_Options::USE_TIMEOUT, connTimeOut); //远程端点 ACE_INET_Addr remoteEP(m_nRemotePort, m_strConnectIPAddress.c_str()); Log(LOGLEVEL_INFO, "Connecting to(%s:%d) ...", remoteEP.get_host_addr(), remoteEP.get_port_number()); //更新当前连接时间戳 time(&m_tmLastConnect); int nRet = 0; if (m_nLocalPort > 0) { //绑定本地固定端口号 ACE_INET_Addr localEP(m_nLocalPort); nRet = connector.connect(pSockHandler, remoteEP, synch_option, localEP); } else { //绑定本地随机端口号 nRet = connector.connect(pSockHandler, remoteEP, synch_option); } //连接失败 if(nRet == -1) { //轮询切换连接主备服务器(存在) if(!m_strBackupIPAddress.empty() && m_strBackupIPAddress.compare(m_strHostIPAddress) != 0) { m_strConnectIPAddress = m_strConnectIPAddress.compare(m_strHostIPAddress) == 0 ? m_strBackupIPAddress : m_strHostIPAddress; } OnConnectChange(false); return -1; } //启动接收事件(OneTime) if(!nRet && m_lStop) { m_lStop = false; this->activate(THR_NEW_LWP | THR_JOINABLE |THR_INHERIT_SCHED); } OnConnectChange(true); return 0; } int CTcpClient::svc() { //接收 while(!m_lStop) { ACE_Time_Value tvSleep; tvSleep.msec(TASK_NAP_TIME_VALUE); ACE_OS::sleep(tvSleep); ACE_Time_Value tvWaite(0, TASK_NAP_TIME_VALUE); //BLOCKED this->reactor()->handle_events(&tvWaite); } return 0; } int CTcpClient::open(void* param) { return this->reactor()->register_handler(this, ACE_Event_Handler::READ_MASK | ACE_Event_Handler::WRITE_MASK); //if (Base::open(param) == -1) //{ // Log(LOGLEVEL_ERROR, "open() Failied."); // return -1; //} //return 0; } int CTcpClient::handle_input(ACE_HANDLE) { char *szBuffer = new char[DEFAULT_BUFFER_SIZE]; //接收数据 ssize_t length = this->peer().recv(szBuffer, DEFAULT_BUFFER_SIZE); //连接断开接收失败 if(length <= 0) { delete[] szBuffer; return -1;//implicit call handle_close() clear up } OnReceive(szBuffer, length); return 0; } int CTcpClient::handle_close(ACE_HANDLE handle, ACE_Reactor_Mask closeMask) { int nRet = ACE_Event_Handler::handle_close(handle, closeMask); Disconnect(); return nRet; } int CTcpClient::handle_output(ACE_HANDLE handle /* = ACE_INVALID_HANDLE */) { //调用一次 return 0; } long CTcpClient::Receive(char* szReceive, long lReceiveSize, int nCommunicateTimeOut) { //Confirmed //implicit call handle_close() clear up if(!IsConnected()) return -1; ACE_Time_Value tvTimeout(0, nCommunicateTimeOut); //return this->peer().recv((void *)szReceive, lReceiveSize, &tvTimeout); return this->peer().recv_n((void *)szReceive, lReceiveSize, &tvTimeout); } long CTcpClient::Send(const char* szSend, long lSendSize, int nCommunicateTimeOut) { //Uncertainty //implicit call handle_close() clear up if(!IsConnected()) return -1; ACE_Time_Value tvTimeout(0, nCommunicateTimeOut); ssize_t length = this->peer().send_n(szSend, lSendSize, &tvTimeout); return length; }