C++自定义封装socket操作业务类完整实例

#ifndef SOCKETCONNECTOR_H
#define SOCKETCONNECTOR_H
#include "global.h"
using namespace std;
class SocketConnector
{
public:
  typedef enum {
    ENormal,
    EOther,
  } SocketState;
public:
  static SocketConnector * getInstance();
  inline SocketState state(){ return m_state; }
  inline void setState(SocketState _state){  m_state = _state;  }
  inline bool isConnected() { return m_isConnected;  }
  inline void setConnected(bool state) { m_isConnected = state; }
  void start();
  inline void setServerIP(string ip){  m_server_ip = ip;}
  inline void setServerPort(int port){ m_server_port = port; }
  int connect_sockfd();
  int onSendMessage(string & message);
private:
  SocketConnector();
  void onConnectToServer(string & ip,int port);
  static void * onReportMessage(void * p);
  static void * onReadMessage(void * p);
  static void * onWriteMessage(void * p);
private:
  SocketState m_state;
  bool m_isConnected;
  int m_sockFd;
  string m_server_ip;
  int m_server_port;
  pthread_t m_report_tid;
  pthread_t m_read_tid;
  pthread_t m_write_tid;
};
#endif // SOCKETCONNECTOR_H

#include "global.h"
#include "socketconnector.h"
#include "cmessagecenter.h"
#include "cmip_requestparser.h"
#include "csettings.h"
#include "datadef.h"
#include "cstringutils.h"
using namespace std;
static SocketConnector * g_instance = NULL;
/**************************************************************************************************
*  Single Instance.
***************************************************************************************************/
SocketConnector * SocketConnector::getInstance()
{
  if (g_instance == NULL)
  {
    g_instance = new SocketConnector();
  }
  return g_instance;
}
/**************************************************************************************************
*  Consturoctor
***************************************************************************************************/
SocketConnector::SocketConnector()
{
  m_isConnected = false;
  m_state = ENormal;
}
/**************************************************************************************************
*  Connect to Server By Blocking Method.
***************************************************************************************************/
void SocketConnector::onConnectToServer(string & ip,int port){
  cout << __FUNCTION__ << "connecting::[" << ip << " , " << port << "]" << endl;
  struct timeval send_timeout;
  send_timeout.tv_sec = 5;
  send_timeout.tv_usec = 0;
  int keepalive = 1;
  int keepidle = 10;
  int keepinterval = 5;
  int keepcount = 3;
  int value = 0;
  socklen_t len = sizeof(int);
  static struct sockaddr_in server_addr;
  memset(&server_addr, 0, sizeof(server_addr));
  server_addr.sin_family = AF_INET;
  server_addr.sin_port = htons(port);
  server_addr.sin_addr.s_addr = inet_addr(ip.c_str());
  do
  {
    m_sockFd = socket(AF_INET, SOCK_STREAM, 0);
    if ( -1 == m_sockFd )
    {
      sleep(1);
      continue;
    }
  }while(-1 == m_sockFd);
  if(setsockopt(m_sockFd, SOL_SOCKET, SO_SNDTIMEO, &send_timeout, sizeof(send_timeout)) == -1)
  {
    printf("setsockopt SO_SNDTIMEO fail\n");
  }
  if(setsockopt(m_sockFd, SOL_SOCKET, SO_KEEPALIVE, (void *)&keepalive , sizeof(keepalive )) == -1)
  {
    printf("setsockopt SO_KEEPALIVE fail\n");
  }
  if(setsockopt(m_sockFd, SOL_TCP, TCP_KEEPIDLE, (void*)&keepidle , sizeof(keepidle )) == -1)
  {
    printf("setsockopt TCP_KEEPIDLE fail\n");
  }
  if(setsockopt(m_sockFd, SOL_TCP, TCP_KEEPINTVL, (void *)&keepinterval , sizeof(keepinterval )) == -1)
  {
    printf("setsockopt TCP_KEEPINTVL fail\n");
  }
  if(setsockopt(m_sockFd, SOL_TCP, TCP_KEEPCNT, (void *)&keepcount , sizeof(keepcount )) == -1)
  {
    printf("setsockopt TCP_KEEPCNT fail\n");
  }
  getsockopt(m_sockFd, SOL_TCP, TCP_KEEPINTVL, (void *)&value, &len);
  cout << __FUNCTION__ << "sockFd KeepIntval::[" << value << endl;
  while (!m_isConnected)
  {
    if(connect(m_sockFd, (struct sockaddr *)&server_addr, sizeof(struct sockaddr)) == 0)
    {
      m_isConnected = true;
      break;
    }
    else
    {
      if ( ECONNREFUSED == errno)
      {
        m_isConnected = false;
        sleep(1);
        printf("Reconnect To Server:%s Port:%d\n", m_server_ip.c_str(), m_server_port);
      }
      else
      {
        m_isConnected = false;
        perror("connected() error()");
        exit(-1);
      }
    }
  }
}
/**************************************************************************************************
*  Create Report Thread;
*  Create Read Thread;
*  Create Write Thread;
*  MainThread wait the subThreads exits;
***************************************************************************************************/
void SocketConnector::start()
{
  m_sockFd = connect_sockfd();
  cout << __FUNCTION__ << "Will Create Report|Read|Write Thread." << endl;
  pthread_create(&m_report_tid,NULL, onReportMessage, this);  /* report to cmdmodule*/
  pthread_create(&m_read_tid, NULL, onReadMessage, this);  /* read from cmdmodule*/
  pthread_create(&m_write_tid, NULL, onWriteMessage, this);  /* reply to cmdmodule*/
  pthread_join(m_read_tid,NULL);
  pthread_join(m_write_tid,NULL);
  pthread_join(m_report_tid,NULL);
}
/**************************************************************************************************
*  Used to Get connected socket fd.
*  if connected, return directly.
*  if not connected,try to create connect fd.
***************************************************************************************************/
int SocketConnector::connect_sockfd()
{
  if ( m_isConnected == true)
  {
    cout << __FUNCTION__ << "::Socket is Already Connected." << endl;
    return m_sockFd;
  }
  cout << __FUNCTION__ << "::Will Try to Connect to Server." << endl;
  onConnectToServer(m_server_ip, m_server_port);
  return m_sockFd;
}
/**************************************************************************************************
*  Report Status to CmdModule Thread.
*  every 2s ,report one message to cmdwifi.
***************************************************************************************************/
void * SocketConnector::onReportMessage(void * p)
{
  SocketConnector * connector = (SocketConnector *)(p);
  if ( NULL == p)
  {
    cout << __FUNCTION__ << "onSelectSocket() Error: param [connector] is NULL" << endl;
    return NULL;
  }
  string content;
  int devType = atoi(CSettings::getInstance()->getKuType().c_str());
  int report_interval = atoi(CSettings::getInstance()->getKuReportinterval().c_str());
  string position = CSettings::getInstance()->getKuPosition();
  string local_ip = CSettings::getInstance()->getKuAgentip();
  cout << endl;
  cout << "###################################" << endl;
  cout << "Local-IP::" << local_ip << endl;
  cout << "Ku-CMA-Pos::" << position << endl;
  cout << "Ku-CMA-Type::" << devType << endl;
  cout << "###################################" << endl;
  cout << endl;
  while(true)
  {
    int state = connector->state();
    content = "<status>" + CStringUtils::toString(state) + "</status>";
    content += "<type>" + CStringUtils::toString(devType) + "</type>";
    content += "<site>" + position + "</site>";
    content += "<ip>" + local_ip + "</ip>";
    Response resp(STATUS_REPORT_CMD,0,string(content));
    CMessageCenter::getInstance()->addReply(resp);
    sleep(report_interval);
  }
}
/**************************************************************************************************
*  Read Message from Connection.
*  Then Send Message to MessageCenter Queue.
***************************************************************************************************/
void * SocketConnector::onReadMessage(void * p)
{
  SocketConnector * connector = (SocketConnector *)(p);
  if ( NULL == p)
  {
    cout << __FUNCTION__ << "onSelectSocket() Error: param [connector] is NULL" << endl;
    return NULL;
  }
  int sockFd = connector->connect_sockfd();
  fd_set fds;
  struct timeval timeout={0,0};
  const int BUFFER_LEN = 4*1024;
  static char buffer[BUFFER_LEN]={0};
  while(true)
  {
    FD_ZERO(&fds);
    FD_SET(sockFd,&fds);
    int ret = select(sockFd + 1,&fds,NULL,NULL,&timeout);
    switch (ret) {
      case -1:/*Error process*/
      {
        perror("select()");
        if ( EBADF == errno)
        {
          close(sockFd);
          connector->setConnected(false);
          sleep(1);
          sockFd = connector->connect_sockfd();
          continue;
        }
        if ( EINTR == errno || ENOMEM == errno)
        {
          sleep(1);
          continue;
        }
      }break;
      case 0:
      {
        //cout << "select() timeout! " << endl;
      }break;
      default:
      {
        if(FD_ISSET(sockFd,&fds))
        {
          memset(buffer, 0, BUFFER_LEN);
          int nRead = read(sockFd, buffer, BUFFER_LEN);
          cout << ">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>" << endl;
          cout << "From Server Recevied Data::" << string(buffer) << endl;
          cout << "From Server Recevied Length::" << nRead << endl;
          cout << ">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>" << endl;
          CRequestParser parser;
          Request req;
          int ret = parser.parseToMessage(buffer,&req);
          if (0 != ret)
          {
            cout << __FUNCTION__ << "Request Format is invalid" << endl;
            continue;
          }
          req.print();
          CMessageCenter::getInstance()->addRequest(req);
        }
      }break;
    }
  }
}
/**************************************************************************************************
*  Write Message to Connection.
*  Then Send Message to MessageCenter Queue.
***************************************************************************************************/
void * SocketConnector::onWriteMessage(void * p)
{
  SocketConnector * connector = (SocketConnector *)(p);
  if ( NULL == p)
  {
    cout << __FUNCTION__ << "onSelectSocket() Error: param [connector] is NULL" << endl;
    return NULL;
  }
  while (true)
  {
      Response msg;
      CMessageCenter::getInstance()->getReplyMsg(msg);
      string data = CMessageEncoder(msg).encode();
      connector->onSendMessage(data);
  }
}
/**************************************************************************************************
*  Send Message By Socket.
***************************************************************************************************/
int SocketConnector::onSendMessage(string & strSend)
{
  if (atoi(CSettings::getInstance()->getDebugMode().c_str()) == 1)
  {
    cout << __FUNCTION__ << "Send To Cmdwifi Data::" << endl;
    cout << strSend << endl;
  }
  int sock = m_sockFd;
  char *pData = &strSend[0];
  int nLen = static_cast<int>(strSend.size());
  int nTotal = nLen;
  int i = 0;
  while(1)
  {
    int nTmp = send(sock, &pData[i], nTotal, 0);
    if (nTmp <= 0)
    {
      close(sock);
      return -1;
    }
    nTotal -= nTmp;
    i += nTmp;
    if (nTotal <= 0)
    {
      break;
    }
  }
  return 0;
}