字节型TCP协议解析

#include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <unistd.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <error.h>
#include <errno.h>
#include <string>
#include <list>

#define DEBUGS

inline char* memstr(void *buf , int size  , void * need , int needsize)
{
char *p=(char*) buf;
int index = 0 ;
while(index < size)
{
if((size-index) < needsize )
{
return NULL;
}
if( memcmp(p,need,needsize) == 0 ) //find the right buffer(need)
{
//  printf("find,index=%d\n",index);
return p;
}
else
{
++p;
index++;
}
}
return NULL;
}

class BaseHandler
{
public:
BaseHandler()
{}
virtual ~BaseHandler()
{}
public:
virtual int hander( void * data ,  int size ) =0;
void config(const char *p )
{
strFileName= p;
}
protected :
std::string strFileName ;
};

class MagReader :public BaseHandler
{
public:
typedef struct _msg_data_
{
void *p;
int size ;
_msg_data_( void * a , int b )
{
p =malloc(b);
memcpy(p , a , b ) ;
size = b ;
}
~_msg_data_()
{
if(p)
free(p);
}
}MSG,*P_MSG;
public:
MagReader()
{}
~MagReader()
{
for( std::list<P_MSG>::iterator it =m_list.begin() ; it!=m_list.end() ;it++ )
{
delete (*it) ;
}
}
public:
int hander( void * data ,  int size )
{
P_MSG buf =  new MSG( data ,size ) ;
m_list.push_back(buf);

#ifdef DEBUGS
char *p_=(char*)data ;
for( int  i = 0 ; i < size ; i++)
{
printf("%c" , p_[i]) ;
}
printf("\n") ;
#endif
}
private:
std::list<P_MSG> m_list ;
};

class FileWriter:public BaseHandler
{
public:
FileWriter():fp(NULL) , bopened(false)
{

}
~FileWriter()
{
if(fp)
{
fclose(fp) ;
fp = NULL ;
bopened = false ;
}
}
public:
int hander( void * data ,  int size  )
{
if(bopened)
{
this->write(data, size ) ;
}
else
{
this->open(strFileName.c_str() ) ;
this->write(data, size ) ;
}
}
public:
const int open(const char * filename)
{
fp=fopen(filename,"w+") ;
bopened=true;
}
const int write( void * buffer  , int buffer_size = 1 )
{
fwrite(buffer , buffer_size , 1 ,fp ) ;
return 0;
}
private:
FILE *fp;
bool bopened;
};

class Factory
{
public:
static BaseHandler* object( int type )
{
switch(type)
{
case 0:
return new MagReader();
break;
case 1:
return new FileWriter();
break;
default:
break;
}
return NULL ;
}
};

typedef int socket_t ;
class FileTransport
{
private:
FileTransport();
~FileTransport();
public:
static FileTransport* instance();
static void release() ;
public:
const int Init( char *Server
4000
IP,  int Serverport ) ;
const int OpenCusor( char *mesh , char*sql) ;
const int Fetch( ) ;
const int Fetch2( ) ;
const int Fetch3( ) ;
void Close() ;
void setfile(const char * p) ;
private:
socket_t fd ;
static FileTransport* m_pFileTransport ;
BaseHandler* handler;
std::string strFileName ;
} ;

FileTransport* FileTransport::m_pFileTransport=NULL;
FileTransport::FileTransport()
{
fd = socket(AF_INET,SOCK_STREAM,0);
handler = NULL ;
}

FileTransport::~FileTransport()
{
this->Close() ;
if(handler)
{
delete handler;
handler=NULL;
}
}
void FileTransport::setfile(const char * p)
{
strFileName=p;
}

FileTransport* FileTransport::instance()
{
if(m_pFileTransport == NULL )
{
m_pFileTransport=new FileTransport();
}
return m_pFileTransport;
}

void FileTransport::release()
{
if(m_pFileTransport)
{
delete m_pFileTransport ;
}
}

//连接服务器
const int FileTransport::Init( char *ServerIP,  int Serverport )
{
struct sockaddr_in address;
address.sin_family = AF_INET;
address.sin_addr.s_addr = inet_addr(ServerIP);
address.sin_port = htons(Serverport);
int uiLen = sizeof(address);
int ret = connect( fd ,(struct sockaddr *)&address,uiLen );
if(ret != 0 )
{
printf("connect failed") ;
return -1;
}
}
void FileTransport::Close()
{
::close(fd);
}
const int FileTransport::OpenCusor( char *mesh , char*sql)
{
char ch[1024] = {0};
char *p = ch ;
strcpy( p, mesh ) ;
p+= strlen(mesh) ;

strcpy( p," " ) ;
p+= 1 ;
strcpy( p,sql ) ;
int idatasize = strlen(ch);
printf("data is [%s],size=%d\n",ch,idatasize ) ;
::send(fd,&idatasize,sizeof(int) , 0 ) ;
::send(fd,ch,idatasize ,0);

char chm ={0};
::recv( fd , &chm , 1 , 0   ) ;
printf("chm=%d\n" , chm ) ;
}

//provide a method to prevent tcp pollution.
//the process hanged up to  wait the timing when the Protocol_Header  is found in the
// recv tcp streams .
// in anthor words , if the process is waked up , means that recv the right character
//we wanted . by the way , continue to handler the next stream is allowed .
#if 1
const int FileTransport::Fetch3( )
{
#define Protocol_Header 0xFDF4
#define Protocol_Tail 0xFDF5
#define ReadIOCacheSize 2
/***不需要字节对齐***/
#pragma pack(1)
struct headers
{
//unsigned int a ;
unsigned short head ;
unsigned char type;
unsigned int length;
};
#pragma pack()

#pragma pack(1)
struct tail
{
unsigned char check;
unsigned short tail;
};
#pragma pack()

struct headers obj ;
memset(&obj,  0x00 ,sizeof(obj) ) ;
char *pHead=(char*)malloc( obj.head)   ;
char *pHead_Walk = pHead ;
char *pHead_End = pHead_Walk+(sizeof(obj.head) -1 ) ;
while(1)
{
unsigned char ch;
::recv(fd,&ch,1,0 ) ;
if( pHead_Walk <   pHead_End )
{
memcpy(pHead_Walk , &ch , 1 )  ;
++pHead_Walk;
}
else// equal
{
memcpy(pHead_Walk , &ch , 1 )  ;
unsigned short s = 0 ;
memcpy(&s , pHead , sizeof(s) ) ;
if( s == Protocol_Header )//find the right character position
{
obj.head = s ;
break;
}
else
{
pHead_Walk=pHead;
}

}
}
if(pHead)
{
free(pHead) ;
}
printf("start get type and length\n");
char *pBuf = (char*)(&obj) ;
pBuf += sizeof(obj.head) ;//skip the head
char *pBuf_Walk = pBuf ;

int iReadHead_Reset = 0 ;
while(iReadHead_Reset<(sizeof(obj) - sizeof(obj.head)  ))
{
char chBuf[ 1024 ] ={0} ;
int size = ::recv(fd,chBuf, ( sizeof(obj) - sizeof(obj.head) ) ,0 )  ;
memcpy(pBuf_Walk ,chBuf ,size  ) ;
pBuf_Walk += size ;
iReadHead_Reset+=size ;
}
printf("type=%d,   length=%d\n" , obj.type, obj.length ) ;

handler = Factory::object(obj.type) ;
if(!handler)
{
printf("tyep[%d] is error\n" , obj.type )  ;
return -1;
}
handler->config(strFileName.c_str());

char buf[ReadIOCacheSize]={0};
int bufreaded = 0 ;
/*查找匹配的正文.每次读取当前剩余的字节数与ReadIOCacheSize的最小值*/
do
{
int sizeEach=(ReadIOCacheSize>( obj.length -bufreaded ) )?( ( obj.length -bufreaded ) ):(ReadIOCacheSize);
memset(buf,0x00 ,sizeof(buf) ) ;
int ret = recv( fd , buf , sizeEach , 0   ) ;
#ifdef WIN32
if(ret == SOCKET_ERROR)
{
return -1;
}
#else
if(ret == EAGAIN)
{
return -1;
}
#endif
bufreaded+=ret;

handler->hander(buf ,ret ) ;
printf("[%d receieved]%s\n",ret,buf) ;
}while(bufreaded < obj.length  ) ;

//检查包的结尾
struct tail otail;
memset( &otail , 0x00 , sizeof(otail) ) ;
char *ptail = (char*)&otail ;
char *otail_Walk  = ptail ;
int inReadTail  =  0  ;
while( inReadTail < sizeof(otail) )
{
char tmp[sizeof(otail) ] = {0} ;
int ret = ::recv(fd,tmp,sizeof(otail),0);
#ifdef WIN32
if(ret == SOCKET_ERROR)
{
return -1;
}
#else
if(ret == EAGAIN)
{
return -1;
}
#endif
memcpy(otail_Walk,tmp ,ret  ) ;
inReadTail+=ret;
otail_Walk+=ret;
}

#if  0
char tmp[1024] = {0} ;
while(1)
{
memset(tmp,0x00 , sizeof(tmp) ) ;
int ret = ::recv(fd,tmp,sizeof(otail),0);
#ifdef WIN32
if(ret == SOCKET_ERROR)
{
return -1;
}
#else
if(ret == EAGAIN)
{
return -1;
}
#endif
if(ret ==  sizeof(otail))
{
break ;
}
else
{
continue  ;
}
}

memcpy(&otail , tmp , sizeof(otail)  ) ;
#endif

printf("tail=%d\n" , otail.tail) ;
if( otail.tail != Protocol_Tail )
{
printf("receive error tail[%d]\n" ,otail.tail ) ;
sleep(10000) ;
return -1;
}

return 0;
//.........
}
#endif

const int FileTransport::Fetch2( )
{
#define Protocol_Header 0xFDF4
#define Protocol_Tail 0xFDF5
#define ReadIOCacheSize 2
/***不需要字节对齐***/
#pragma pack(1)
struct headers
{
//unsigned int a ;
unsigned short head ;
unsigned char type;
unsigned int length;
};
#pr
d507
agma pack()

#pragma pack(1)
struct tail
{
unsigned char check;
unsigned short tail;
};
#pragma pack()

/*查找匹配的头*/
struct headers oheaders;
memset(&oheaders,0x00,sizeof(oheaders) ) ;
printf("sizeof struct headers = %d\n" , sizeof( oheaders ) );
while(1)
{
char ch[1024]={0} ;
int ret = recv(fd,ch,sizeof(oheaders),0);
if(ret == 0 )
{
printf("server has close this session\n" ) ;
return -1;
}
#ifdef WIN32
if(ret == SOCKET_ERROR)
{
return -1;
}
#else
if(ret == EAGAIN)
{
return -1;
}
#endif
printf("recv size=%d\n" , ret) ;
memcpy(&oheaders,ch , sizeof(oheaders));
printf("oheaders.type=%d,oheaders.head=%d,oheaders.length=%d\n",oheaders.type,oheaders.head,oheaders.length);
if(oheaders.head==Protocol_Header)
{
break;
}
else
{
printf("receive error head\n") ;
return -1;
}
}

/***在这里根据type选择创建的类型 0Msg文言 1 文件输出***/
handler = Factory::object(oheaders.type) ;
if(!handler)
{
printf("tyep[%d] is error\n" , oheaders.type )  ;
return -1;
}
handler->config(strFileName.c_str());
char buf[ReadIOCacheSize]={0};
int bufreaded = 0 ;
/*查找匹配的正文.每次读取当前剩余的字节数与ReadIOCacheSize的最小值*/
do
{
int sizeEach=(ReadIOCacheSize>( oheaders.length -bufreaded ) )?( ( oheaders.length -bufreaded ) ):(ReadIOCacheSize);
memset(buf,0x00 ,sizeof(buf) ) ;
int ret = recv( fd , buf , sizeEach , 0   ) ;
#ifdef WIN32
if(ret == SOCKET_ERROR)
{
return -1;
}
#else
if(ret == EAGAIN)
{
return -1;
}
#endif
bufreaded+=ret;

handler->hander(buf ,ret ) ;
//printf("[%d receieved]%s\n",ret,buf) ;
}while(bufreaded < oheaders.length  ) ;

//检查包的结尾
struct tail otail;
memset( &otail , 0x00 , sizeof(otail) ) ;
char tmp[1024] = {0} ;
recv(fd,tmp,sizeof(otail),0);
memcpy(&otail , tmp , sizeof(otail)  ) ;
printf("tail=%d\n" , otail.tail) ;
if( otail.tail != Protocol_Tail )
{
printf("receive error tail\n") ;
return -1;
}
}

int main( int argc ,char ** argv)
{
FileTransport*p = FileTransport::instance();
p->setfile(argv[3]);
p->Init(argv[1] ,atoi(argv[2]) ) ;
p->OpenCusor("1122222" ,"select * from aaaaaaa" ) ;
int i = p->Fetch3();
return i ;
}