#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 failn");
}
if(setsockopt(m_sockFd, SOL_SOCKET, SO_KEEPALIVE, (void *)&keepalive , sizeof(keepalive )) == -1) {
printf("setsockopt SO_KEEPALIVE failn");
}
if(setsockopt(m_sockFd, SOL_TCP, TCP_KEEPIDLE, (void*)&keepidle , sizeof(keepidle )) == -1) {
printf("setsockopt TCP_KEEPIDLE failn");
}
if(setsockopt(m_sockFd, SOL_TCP, TCP_KEEPINTVL, (void *)&keepinterval , sizeof(keepinterval )) == -1) {
printf("setsockopt TCP_KEEPINTVL failn");
}
if(setsockopt(m_sockFd, SOL_TCP, TCP_KEEPCNT, (void *)&keepcount , sizeof(keepcount )) == -1) {
printf("setsockopt TCP_KEEPCNT failn");
}
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:%dn", 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;
}