www.pudn.com > SMS PLAN.rar > func.cpp
#include "stdafx.h"
#include
#include
#include
#include
#include
#include
#include
#include "gw.h"
#include "func.h"
#include
#include
#define BEEPER 0
CArg arg;
DWORD GetNum(_RecordsetPtr rs,LPCTSTR index);
int GetStr(_RecordsetPtr rs,LPCTSTR index,_bstr_t Item);
_bstr_t GetStr(_RecordsetPtr rs,LPCTSTR index);
DWORD GetNum(_RecordsetPtr rs,LPCTSTR index)
{
long out=0;
if(rs->GetRecordCount()==0)return 0;
sscanf((LPCSTR)(_bstr_t)rs->GetFields(
)->GetItem(index)->Value,
"%d",&out);
return (unsigned long)out;
}
int GetStr(_RecordsetPtr rs,LPCTSTR index,_bstr_t Item)
{
if(rs->Fields->GetItem(index)->GetActualSize()<1)return 0;
Item=_bstr_t(rs->Fields->GetItem(index)->Value);
return strlen((LPSTR)Item);
}
_bstr_t GetStr(_RecordsetPtr rs,LPCTSTR index)
{
if(rs->Fields->GetItem(index)->GetActualSize()<1)
return _bstr_t((char*)NULL);
return _bstr_t(rs->Fields->GetItem(index)->Value);
}
GW_API DWORD BeginGW(LPVOID pp)
{
DWORD dwRet;
struct sockaddr_in server;
WSAData wsaData;
HANDLE hEvent[4];
memset(&server,0,sizeof(server));
WSAStartup(0x202,&wsaData);
dwRet=0;
while(1)
{
if(!GetArg(&arg))
{
//SleepEx(1000*10,1);
#if BEEPER
Beep(17600,10000);
#endif
return 1;
}
#if !_DEBUG
if(atoi(arg.ESMEID)<10000)
{
AddLog("不合法的服务代码。\n");
return 1;
}
#endif
server.sin_family = AF_INET;
server.sin_port = htons(arg.SMSCPORT);
server.sin_addr.s_addr = inet_addr(arg.SMSCIP);
arg.s = socket(AF_INET,SOCK_STREAM,0); /* Open a socket */
setsockopt(arg.s,IPPROTO_TCP,SO_REUSEADDR,(char*)1,4);
if (arg.s <0 )
{
AddLog("打开socket错误\n");
WSACleanup();
if (WSAStartup(0x202,&wsaData) == SOCKET_ERROR)
{
AddLog(TEXT("调用WSAStartup失败。\n"));
#if BEEPER
Beep(8800,1000);
#endif
WSACleanup();
}
}
else if(connect(arg.s,
(struct sockaddr*)&server,
sizeof(server))== SOCKET_ERROR)
{
if(dwRet<1)
AddLog("不能连接EHOO中心服务器<错误号：%d>.\n",GetLastError());
dwRet++;
if(dwRet>arg.nMaxReconnect)
{
if(arg.isDebug)
AddLog("重新连接%d次仍然失败。重新运行服务。\n",arg.nMaxReconnect);
#if BEEPER
Beep(4400,10000);
#endif
//return 1;
}
//SleepEx(10*1000,1);
closesocket(arg.s);
WSACleanup();
if (WSAStartup(0x202,&wsaData) == SOCKET_ERROR)
{
AddLog(TEXT("调用WSAStartup失败。\n"));
WSACleanup();
}
}else break;
}
AddLog("成功连接EHOO中心服务器。\n");
#if 0
////////////////testfun runs here.
testfun2(arg.s);
return 0;
///////////////////
#endif
ResetEvent(arg.hGW);
//begin service.
//create the read thread
hEvent[0]=(HANDLE)_beginthreadex(NULL,0,(unsigned int(__stdcall*)(void*))SenderMsg,0,0,0);
hEvent[1]=(HANDLE)_beginthreadex(NULL,0,(unsigned int(__stdcall*)(void*))ReceiverMsg,0,0,0);
hEvent[2]=(HANDLE)_beginthreadex(NULL,0,(unsigned int(__stdcall*)(void*))DeliverMsg,0,0,0);
hEvent[3]=arg.hGW;
while(1)
{
dwRet=WaitForMultipleObjects(4,hEvent,FALSE,INFINITE);
switch(dwRet)
{
case WAIT_FAILED: //error:
AddLog("调用WaitForMultipleObjects失败。\n");
return ExitFromWaitFailed;
break;
case WAIT_TIMEOUT:
break;
case WAIT_OBJECT_0: //sender exit here
CloseHandle(hEvent[0]);
hEvent[0]=(HANDLE)_beginthreadex(NULL,0,(unsigned int(__stdcall*)(void*))SenderMsg,
(LPVOID)arg.s,0,0);
if(hEvent[0]==NULL){StopGW();return 0;}
break;
case WAIT_OBJECT_0 +1://receiver exit here.
arg.isActive=0;
GetExitCodeThread(hEvent[1],&dwRet);
if(dwRet==NOERROR)return NOERROR;
CloseHandle(hEvent[1]);
if(dwRet==ExitFromNetClose)
{
closesocket(arg.s);
WSACleanup();
WSAStartup(0x202,&wsaData);
dwRet=0;
while(1)
{
if(dwRet>arg.nMaxReconnect)
{
AddLog("重新连接 %d 次仍然失败，退出服务.\n",arg.nMaxReconnect);
Exec("update msg_smsc set msg_state=100 where msg_state=101");
StopGW();
return 0;
}
arg.s = socket(AF_INET,SOCK_STREAM,0);
if (arg.s <0 )
{
AddLog("打开socket失败。\n");
WSACleanup();
if (WSAStartup(0x202,&wsaData) == SOCKET_ERROR)
{
AddLog(TEXT("调用WSAStartup失败。\n"));
WSACleanup();
}
}
else if(connect(arg.s,
(struct sockaddr*)&server,
sizeof(server))== SOCKET_ERROR)
{
SleepEx(20*1000,1);
AddLog("连接EHOO中心服务器失败。\n");
closesocket(arg.s);
dwRet++;
}else
{
AddLog("重连EHOO中心服务器成功。\n");
//dwRet=InitMsg(BIND,buffer,0,0);
//send(arg.s,buffer,dwRet,0);
break;
}
}
}
hEvent[1]=(HANDLE)_beginthreadex(NULL,0,(unsigned int(__stdcall*)(void*))ReceiverMsg,
(LPVOID)arg.s,0,0);
if(hEvent[1]==NULL)
{
StopGW();
return 0;
}
break;
case WAIT_OBJECT_0+2:
hEvent[2]=(HANDLE)_beginthreadex(NULL,0,(unsigned int(__stdcall*)(void*))DeliverMsg,0,0,0);
if(hEvent[2]==NULL)
{
StopGW();
return 0;
}
break;
case WAIT_OBJECT_0+3: //EventForExit
//ResetEvent(arg.hGW);
AddLog("得到消息，准备退出子服务。\n");
ExitThread(0);
break;
}//switch(dwRet)
}//end fo while (1)
ExitThread(0);
}
void PutByteString(char* dest,
char* source,
int length)
{
int i=0;
if(source==NULL)return;
for(i=0;i4)b=4;
for(i=0;i>=8;
}
}
DWORD GetTime()
{
char date[20],time[20];
int ret=0;
_strdate(date);
_strtime(time);
date[2]=date[3];
date[3]=date[4];
memcpy(date+4,time,2);
memcpy(date+6,time+3,2);
ret=atoi(date);
return ret;
}
void GetStamp(char* buffer)
{
char date[20],time[20];
_strdate(date);
_strtime(time);
date[2]=date[3];
date[3]=date[4];
memcpy(date+4,time,2);
memcpy(date+6,time+3,2);
memcpy(date+8,time+6,2);
PutNByte(buffer,4,atoi(date));
}
void GetLastday(char* buffer)
{
char current[36];
char tp[36];
int day[]={31,28,31,30,31,30,31,31,30,31,30,31};
struct _timeb timebuffer;
int i,j,k;
_strdate(current);
j=atoi(current);//month
k=atoi(current+3);//day
_ftime(&timebuffer);
strcpy(tp,ctime(&(timebuffer.time)));
tp[strlen(tp)-1]=0;
i=atoi(strrchr(tp,' '));//year
if(k>1)
{
sprintf(buffer,"%.4d%.2d%.2d",i,j,k-1);
return;
}
else
{
if(j==1){i--;j=12;}
j--;
if(j==2)
{
if(j%4)k=28;
else if(j%100==0&&j%400)k=28;
else k=29;
}
else k=day[j-1];
sprintf(buffer,"%.4d%.2d%.2d",i,j,k);
}
}
DWORD GetNByte(char* buffer,BYTE b)
{
int i;
DWORD rt=0;
if(b>4)b=4;
for(i=0;i=size)
strncpy(dest,source,size);
else
strcpy(dest,source);
return (long)strlen(dest);
}
int InitMsg(DWORD nCOMMANDID,
char* buffer,
char* mobile,
char* sm,DWORD id_time,DWORD id_sn,DWORD validate,DWORD state)
{
int i,j,k;
unsigned __int16 tp;
BYTE b;
char tmp[5];
ZeroMemory(buffer,SMPPMAXLENGTH);
tp=j=k=i=0;
i+=4;
PutNByte(buffer+i,4,nCOMMANDID);
i+=4;
PutNByte(buffer+i,4,id_time);//sequence no
i+=4;
switch(nCOMMANDID)
{//hbzb@public.wh.hb.cn
case SUBMIT:// 3
/////////////////////////////////
//mobile number
if(strlen(mobile)<11)return 0;
if(strlen(sm)<1||strlen(sm)>MSGLEN)return 0;
PutNByte(buffer+i,4,id_time);
i+=4;
PutNByte(buffer+i,4,1);//id_sn);
i+=4;
buffer[i]=1;//pk_total
i++;
buffer[i]=1;//pk_number
i++;
buffer[i]=1;//need_reply
i++;
buffer[i]=0;//msg level
i++;
sprintf(buffer+i,"%s\0",arg.ESMEID);//service_id
i+=10;
buffer[i]=2;//Fee_UserType
i++;
ZeroMemory(buffer+i,21);
i+=21;//the fee_dest_termid is skimmed
buffer[i]=0;//TP_pid
i++;
buffer[i]=0;//TP_udhi
i++;
/////////////////////////////////////
buffer[i]=(BYTE)id_sn;
i++;
PutNByte(buffer+i,4,atoi(arg.ESMEID));//sp_id
i+=6;
sprintf(buffer+i,"%.2d",5);
i+=2;//fee_type
sprintf(buffer+i,"%.6d",0);
i+=6;//fee_code
//j=day
//k=hour
if((validate/60)>=48)j=2;
else if((validate/60)>=24)j=1;
else j=0;
k=(validate%1440)/60;//yymmddhhmmsstnnp
sprintf(buffer+i,"%.6d%.2d%.2d00320+\0",j,k,validate%60);//validate time
i+=17;
sprintf(buffer+i,"%.10d00320+\0",1);//重发时间间隔
i+=17;
ZeroMemory(buffer+i,21);
i+=21;//src_termid is skimmed
buffer[i]=1;//number of dsttermid
i++;
sprintf(buffer+i,"%s\0",mobile);//dst_termid
i+=21;
if(id_sn==3||id_sn==4||id_sn==8)
{
buffer[i]=strlen(sm)/2;
i++;
for(int j=0;j<(BYTE)buffer[i-1];j++)
{
sprintf(tmp,"0x%.2s\0",sm+j*2);
sscanf(tmp,"%x",&b);
buffer[i+j]=b;
}
i+=buffer[i-1];
}else
{
buffer[i]=strlen(sm);
i++;
if(strlen(sm)<=160)
strncpy(buffer+i,sm,strlen(sm));
i+=strlen(sm);
}
i+=8;//reserved 8 bytes
//////////////////////////////
break;
case BIND:// over
sprintf(buffer+i,"%.10s\0",arg.ESMEID);//service_id
i+=10;
Jstrncpy(buffer+i,arg.ESMEPWD,16);
//sprintf(buffer+i,"%d\0",getHardDriveComputerID());
i+=16;
buffer[i]=12;
i++;
GetStamp(buffer+i);
i+=4;
break;
case UNBIND://over
break;
case DELIVERYRESP://over
i-=4;
PutNByte(buffer+i,4,validate);//sequence no
i+=4;
memcpy(buffer+i,sm,8);
i+=8;
buffer[i]=(BYTE)state;//0=right result
i++;
break;
case CANCEL://over
//cancel the messagea to MSG_id in first 8 bytes of mobile
memcpy(buffer+i,mobile,8);
i+=8;
break;
case QUERYSTATUS:
//query the message to mobile
GetLastday(buffer+i);
PutNByte(buffer+i,4,atoi(arg.ESMEID));//station_id
PutNByte(buffer+i+4,4,id_time);//msg_id
i+=8;
buffer[i]=2;
i++;
Jstrncpy(buffer+i,sm,10);
i+=10;
i+=8;//8 bytes to reserve
break;
case ALIVE://over
break;
case ALIVERESP://over
buffer[i]=0;
i++;
break;
}
PutNByte(buffer,4,i);
if(arg.isDebug)
AddBinLog(buffer,i);
return i;
}
DWORD WINAPI ReceiverMsg(LPVOID pp)
{
int i,j,AA;
WORD wAlive=0;
fd_set rfds,afds;
int binderror=0;
DWORD dwRet=0;
char strSQL[400];
char msg[256];
char buffer[SMPPMAXLENGTH];
_ConnectionPtr pCnnGSM=NULL;
timeval tv;
tv.tv_sec=20;
tv.tv_usec=0;
FD_ZERO(&afds);
FD_SET(arg.s,&afds);
try
{
::CoInitialize(NULL);
TESTHR(pCnnGSM.CreateInstance(__uuidof(Connection)));
pCnnGSM->ConnectionTimeout = 120;
pCnnGSM->CommandTimeout=300;
TESTHR(pCnnGSM->Open(arg.strCnn,"","",adConnectUnspecified));
}//try
catch (_com_error &e)
{
AddLog("数据库连接失败。\n");
AddLog("失败原因： '%s'\n", (char*) e.Description());
if(pCnnGSM->GetState())pCnnGSM->Close();
::CoUninitialize();
SleepEx(1000,1);
return 1;
}
i=InitMsg(BIND,buffer,0,0);
if(send(arg.s,buffer,i,0)2)
{
return (ExitFromNetClose);
}
//AddLog("Beyond the time limitation to SMSC .");
if(!arg.isActive)
{
j=InitMsg(BIND,buffer,0,0);
if(j!=send(arg.s,buffer,j,0))
{
AddLog("send BIND PACKAGE error no: %d\n",GetLastError());
return (ExitFromNetClose);
}
break;
}
j=InitMsg(ALIVE,buffer,0,0);
if(j!=send(arg.s,buffer,j,0))
{
AddLog("send ALIVE PACKAGE error no: %d\n",GetLastError());
return (ExitFromNetClose);
}
else wAlive++;
break;
default: //data is coming.
if(!FD_ISSET(arg.s,&rfds))return ExitFromNetClose;
ZeroMemory(buffer,SMPPMAXLENGTH);
i=j=recv(arg.s,buffer,4,0);
//i=the bytes of the message received
while(i<4)
{
j=recv(arg.s,buffer+i,4-i,0);
SleepEx(1000,1);
if(j<1)return ExitFromNetClose;
i+=j;
}
while(i<(int)GetNByte(buffer,4))
{
j=recv(arg.s,buffer+i,GetNByte(buffer,4)-i,0);
if(j<1)return ExitFromNetClose;
i+=j;
}
if(arg.isDebug)
AddBinLog(buffer,i);
switch(GetNByte(buffer+4,4))
{
case BINDRESP:
if(buffer[12])
{
if(binderror<1)
AddLog("登录EHOO中心服务器错误<%d>\n",buffer[12]);
SleepEx(3000,1);
binderror++;
i=InitMsg(BIND,buffer,0,0);
if(send(arg.s,buffer,i,0)Errors->Clear();
pCnnGSM->Execute((char*)strSQL,NULL,adExecuteNoRecords);
break;
case UNBINDRESP:
AddLog("从EHOO中心服务器退登录。\n");
break;
case DELIVERY:
memcpy(msg,buffer+12,8);
if(buffer[75]==0)//deliver the msg to esme
{
#if _DEBUG
AddLog("dest no: %d\n",atoi(buffer+20));
AddLog("src_term: %s\n",buffer+54);//src mobile
AddLog("msg : %s\n",buffer+77);
#endif
sprintf(strSQL,"INSERT INTO MSG_ESME (ID_TIME,ID_SN,SERVICE_ID,DST_TERMID,SRC_TERMID,MSG) "
"VALUES(%d,%d,' ','%21s','%.11s','%s')",
GetNByte(buffer+12,4),GetNByte(buffer+16,4),buffer+20,buffer+54,buffer+77);
AA=0;
}
else//the status report to msg from esme
{
if(buffer[92]&&buffer[102])
sprintf(strSQL,"EXEC SP_STATEREPORT %d,%.21s,'%.10s','%.10s'",
GetNByte(buffer+133,4),buffer+85,buffer+92,buffer+102);
else
sprintf(strSQL,"EXEC SP_STATEREPORT %d,%d,' ',' '",
GetNByte(buffer+133,4),atoi(buffer+85));
#if _DEBUG
AddLog("status %d of %d\n",GetNByte(buffer+133,4),atoi(buffer+85));
#endif
if(atoi(buffer+54)==atoi(arg.ESMEID))AA=0;
else AA=1;
}
pCnnGSM->Errors->Clear();
pCnnGSM->Execute((char*)strSQL,NULL,adExecuteNoRecords);
i=InitMsg(DELIVERYRESP,buffer,0,msg,GetNByte(buffer+12,4),GetNByte(buffer+16,4),GetNByte(buffer+8,4),AA);
send(arg.s,buffer,i,0);
arg.isActive=1;
break;
case CANCELRESP:
//GetID(msg,buffer+12);
AddLog("Cancel the message <%d-%d>.\n",GetNByte(buffer+12,3),GetNByte(buffer+18,2));
break;
}//end of swtich the type of the package
//end of processing a package.
}//end of switch the wait for result
}//end of while 1
}//end of try
catch (_com_error &e)
{
AddLog("ReceiverMsg执行SQL错误。 <%s>.\n",strSQL);
AddLog("错误描述：'%s'\n", (char*) e.Description());
if(pCnnGSM->GetState())pCnnGSM->Close();
::CoUninitialize();
SleepEx(500,1);
return 1;
}
}
int Jstrchr(const char* s,char c)
{
WORD i;
if(s==NULL)return -1;
if(strlen(s)<1)return -1;
for(i=0;iConnectionTimeout = 120;
pCnn->CommandTimeout=300;
TESTHR(pCnn->Open(arg.strCnn,"","",adConnectUnspecified));
//pCnn->BeginTrans();
pCnn->Errors->Clear();
pCnn->Execute((char*)str,NULL,adExecuteNoRecords);
//pCnn->CommitTrans();
pCnn->Close();
return 0;
}
catch (_com_error &e)
{
AddLog("Exec内SQL错误。r<%s>.\n",(char*)str);
AddLog("错误描述：'%s'\n", (char*) e.Description());
return 1;
}
return 0;
}
DWORD WINAPI ExecM(LPVOID pp)
{
_ConnectionPtr pCnn=NULL;
char* str=NULL;
BYTE i=0,ret=0;
#if ONESECTION3
EnterCriticalSection(&(arg.syc3));
#endif
rerun:
try
{
::CoInitialize(NULL);
TESTHR(pCnn.CreateInstance(__uuidof(Connection)));
pCnn->ConnectionTimeout = 120;
pCnn->CommandTimeout=300;
TESTHR(pCnn->Open(arg.strCnn,"","",adConnectUnspecified));
//pCnn->BeginTrans();
for(i=0;iErrors->Clear();
str=((char**)pp)[i+1];
pCnn->Execute(str,NULL,adExecuteNoRecords);
}
//pCnn->CommitTrans();
ret=0;
}
catch (_com_error &e)
{
if(str)AddLog("ExecM执行SQL错误: <%s>\n",str);
AddLog("描述：<%s>\n", (char*) e.Description());
ret=1;
if(pCnn->GetState()==0)goto rerun;
if(pCnn->GetState())pCnn->Close();
#if ONESECTION3
LeaveCriticalSection(&(arg.syc3));
#endif
::CoUninitialize();
for(i=0;iGetState())pCnn->Close();
#if ONESECTION3
LeaveCriticalSection(&(arg.syc3));
#endif
::CoUninitialize();
for(i=0;idwRet=0;
GetModuleFileName(NULL,tchar,MAX_PATH);
sprintf(strrchr(tchar,'\\')+1,"gw.ini\0");
pf=fopen(tchar,"r");
if(pf==NULL)
{
AddLog("%s 文件打开失败。\n",tchar);
return 0;
}
memset(arg,0,sizeof(CArg));
fgets(tchar,_MAX_PATH,pf);
sprintf(arg->SMSCCODE,"%s\0",strchr(tchar,'=')+1);
fgets(tchar,_MAX_PATH,pf);
sprintf(arg->servicecode,"%s\0",strchr(tchar,'=')+1);
fgets(tchar,_MAX_PATH,pf);
//sscanf(strchr(tchar,'=')+1,"%s",arg->SqlServerName);
arg->strCnn=_bstr_t("Provider=sqloledb;Data Source=")+_bstr_t(strchr(tchar,'=')+1);
fgets(tchar,_MAX_PATH,pf);
//sscanf(strchr(tchar,'=')+1,"%s",arg->SqlInitCatalog);
arg->strCnn+=_bstr_t(";Initial Catalog=")+_bstr_t(strchr(tchar,'=')+1);
fgets(tchar,_MAX_PATH,pf);
//sscanf(strchr(tchar,'=')+1,"%s",arg->SqlServerUser);
arg->strCnn+=_bstr_t(";User Id=")+_bstr_t(strchr(tchar,'=')+1);
fgets(tchar,_MAX_PATH,pf);
//sscanf(strchr(tchar,'=')+1,"%s",arg->SqlServerPWD);
arg->strCnn+=_bstr_t(";Password=")+_bstr_t(strchr(tchar,'=')+1);
fgets(tchar,_MAX_PATH,pf);
sscanf(strchr(tchar,'=')+1,"%s",arg->ESMEID);
fgets(tchar,_MAX_PATH,pf);
#if _DEBUG
sscanf(strchr(tchar,'=')+1,"%s",arg->ESMEPWD);
#else
if((id=getHardDriveComputerID())<2)
id=GetVersion();
if(id<0x1000)id+=0x12345678;
else if(id>0x80000001)id-=0x12345678;
sprintf(arg->ESMEPWD,"%d\0",id);
#endif
fgets(tchar,_MAX_PATH,pf);
sscanf(strchr(tchar,'=')+1,"%s",arg->SMSCIP);
fgets(tchar,_MAX_PATH,pf);
sscanf(strchr(tchar,'=')+1,"%d",&tp);
arg->SMSCPORT=tp;
fgets(tchar,_MAX_PATH,pf);
sscanf(strchr(tchar,'=')+1,"%d",&tp);
arg->nMaxReconnect=tp;
fgets(tchar,_MAX_PATH,pf);
sscanf(strchr(tchar,'=')+1,"%d",&tp);
arg->nMaxLost=tp;
fgets(tchar,_MAX_PATH,pf);
sscanf(strchr(tchar,'=')+1,"%d",&tp);
if(tp<5)tp=5;
arg->nMaxInterval=tp;
fgets(tchar,_MAX_PATH,pf);
sscanf(strchr(tchar,'=')+1,"%s",arg->szManager);
fgets(tchar,_MAX_PATH,pf);
sscanf(strchr(tchar,'=')+1,"%d",&tp);
arg->nMaxResend=tp;
fgets(tchar,_MAX_PATH,pf);
sscanf(strchr(tchar,'=')+1,"%d",&tp);
arg->isDebug=(tp==1);
fgets(tchar,_MAX_PATH,pf);
sscanf(strchr(tchar,'=')+1,"%d",&tp);
if(tp>5||tp<0)tp=0;
arg->dwStationID=tp;
fclose(pf);
arg->hGW=CreateEvent(NULL,TRUE,FALSE,NULL);
if(arg->hGW==NULL)return 0;
// arg->hMutex=NULL;
arg->isActive=0;
#if ONESECTION1
InitializeCriticalSection(&(arg->syc1));
#endif
#if ONESECTION2
InitializeCriticalSection(&(arg->syc2));
#endif
#if ONESECTION3
InitializeCriticalSection(&(arg->syc3));
#endif
return 1;
}
DWORD WINAPI SenderMsg(LPVOID pp)
{
char buffer[SMPPMAXLENGTH];
char *p=0;
char mobile[MOBILELEN];
char msg[MSGLENP1];
DWORD errorno[250],sendno[250];
DWORD msg_id,station_id,msg_period;
BYTE msg_format;
WORD wRead,i,j,k,tk;
_bstr_t strRecord;
char strSQL[512];
_ConnectionPtr pCnnGSM=NULL;
_RecordsetPtr rsGSM=NULL;
try
{
::CoInitialize(NULL);
TESTHR(pCnnGSM.CreateInstance(__uuidof(Connection)));
pCnnGSM->ConnectionTimeout = 120;
pCnnGSM->CommandTimeout=300;
TESTHR(pCnnGSM->Open(arg.strCnn,"","",adConnectUnspecified));
TESTHR(rsGSM.CreateInstance(__uuidof(Recordset)));
}//try
catch (_com_error &e)
{
AddLog("SenderMsg连接数据库失败。\n");
AddLog("描述：'%s'\n", (char*) e.Description());
if(pCnnGSM->GetState())pCnnGSM->Close();
::CoUninitialize();
SleepEx(1000,1);
return 1;
}
#if _DEBUG
AddLog("entering the sender thread.\n");
#endif
while(1)
{
#if 1
if(arg.isActive<1)
{
SleepEx(500,1);continue;
}
if(arg.dwRet>=COUNT2SMSC)
{
SleepEx(500,1);continue;
}
#endif
try
{
/////////////////////////////////////////////////////////////////////////////////////
sprintf(strSQL,
"SELECT TOP %d MSG_ID,STATION_ID,MSG_PERIOD,MSG_FORMAT,CALLED,MSG "
"FROM MSG_SMSC WHERE MSG_STATE=101"
"AND RESEND<%d ORDER BY MSG_ID",110,arg.nMaxResend);
rsGSM->Open(strSQL,_variant_t((IDispatch *)pCnnGSM),
adOpenStatic,adLockPessimistic,adCmdText);
if((!rsGSM->BOF)&&(!rsGSM->EndOfFile))
{
//COPY the data into the buffer
if(rsGSM->GetRecordCount())
strRecord=rsGSM->GetString(adClipString,
rsGSM->RecordCount,"\1","\2","\0");
rsGSM->Close();
}
else
{
rsGSM->Close();
SleepEx(300,1);
sprintf(strSQL,"exec sp_beforesend");
pCnnGSM->Errors->Clear();
pCnnGSM->Execute(strSQL,NULL,adExecuteNoRecords);
continue;
}
#if _DEBUG
AddLog("before sending msgs.\n");
#endif
i=j=k=0;
p=(char*)strRecord;
while(p)//the next record is avairable
{
if(arg.isActive<1)
{
SleepEx(100,1);
continue;
}
if(arg.dwRet>=COUNT2SMSC)
{
SleepEx(500,1);continue;
}
//MSG_ID,STATION_ID,MSG_PERIOD,CALLED,CONTENT
//p is the head of fields
msg_id=atoi(p);p=strchr(p,1)+1;
station_id=atoi(p);p=strchr(p,1)+1;
msg_period=atoi(p);p=strchr(p,1)+1;
msg_format=atoi(p);p=strchr(p,1)+1;
strncpy(mobile,p,strchr(p,1)-p);
mobile[strchr(p,1)-p]=0;
p=strchr(p,1)+1;
if(strchr(p,2))k=2;
else k=0;
strncpy(msg,p,strchr(p,k)-p);
msg[strchr(p,k)-p]=0;
if(k==0)p=NULL;
else
{
p=strchr(p,k)+1;
if(strlen(p)<1)p=NULL;
}
//if currrent record isn't the end of the rows
//p should be the head of the next record, otherwise p=NULL;
wRead=MultiByteToWideChar(CP_ACP,
MB_PRECOMPOSED,msg,strlen(msg),NULL,0);
if((wRead!=strlen(msg))&&wRead>MSGLEN/2)
{
MultiByteToWideChar(CP_ACP,MB_PRECOMPOSED,msg,strlen(msg),
(LPWSTR)buffer,MSGLEN);
wRead=WideCharToMultiByte(CP_ACP,WC_COMPOSITECHECK,(LPWSTR)buffer,MSGLEN/2,
msg,MSGLEN,NULL,NULL);
msg[wRead]=0;
}
wRead=InitMsg(SUBMIT,buffer,mobile,msg,
msg_id,msg_format,msg_period,station_id);
if(wRead>1)
{
k=tk=0;
while(k0)
{
arg.dwRet++;
#if _DEBUG
//AddLog("Sending to: %d\n",msg_id);
#endif
}
}
else
{
errorno[i]=msg_id;//errorno contains the msgs can not be sent.
i++;
}
}//end of while not eof record set.
pCnnGSM->Errors->Clear();
pCnnGSM->Execute("exec sp_aftersend",NULL,adExecuteNoRecords);
k=0;
while(COUNT2CHANGE*kExecute(strSQL,NULL,adExecuteNoRecords);
k++;
}
k=0;
while(COUNT2CHANGE*kExecute(strSQL,NULL,adExecuteNoRecords);
k++;
}
#if _DEBUG
AddLog("After sending msg.\n");
#endif
}//try
catch (_com_error &e)
{
AddLog("SenderMsg执行SQL错误<%s>.\n",strSQL);
AddLog("错误描述：'%s'\n", (char*) e.Description());
if(pCnnGSM->GetState())pCnnGSM->Close();
::CoUninitialize();
SleepEx(500,1);
#if 0
goto resend;
#else
return 1;
#endif
}
}//END of while(1)
return NOERROR;
}
DWORD WINAPI DeliverMsg(LPVOID pp)
{
HANDLE hEvent=arg.hGW;
char msg[50];
char tp[50];
struct _timeb timebuffer;
while(1)
{
switch(WaitForSingleObject(hEvent,arg.nMaxInterval*60000))
{
case WAIT_FAILED: //error:
AddLog("DeliverMsg调用WSAWaitFor失败。\n");
return ExitFromWaitFailed;
break;
case WAIT_TIMEOUT://insert msg into the database to tell the manager the time
if(arg.isActive)
{
_strdate(tp);
_ftime(&timebuffer);
strcpy(tp+strlen(tp),ctime(&(timebuffer.time)));
sprintf(msg,"%.4s-%.2s-%.2s %.8s.%d",strrchr(tp,' ')+1,tp,tp+3,strchr(tp,':')-2,timebuffer.millitm);
sprintf(tp,"exec sp_insertmessage '%s','%s'",arg.szManager,msg);
Exec(tp);
}
break;
case WAIT_OBJECT_0: //thread exit here
return 0;
break;
}
}
}
GW_API void StopGW()
{
WSACleanup();
AddLog("调用StopGW，发送消息终止子服务.\n");
if(arg.hGW!=NULL)SetEvent(arg.hGW);
#if ONESECTION1
DeleteCriticalSection(&(arg.syc1));
#endif
#if ONESECTION2
DeleteCriticalSection(&(arg.syc2));
#endif
#if ONESECTION3
DeleteCriticalSection(&(arg.syc3));
#endif
}
int TranState(char* buffer)
{
char sstate[8];
int state=0;
strncpy(sstate,buffer,7);sstate[7]=0;
if(strcmp(sstate,"EN_ROUT")==0)state=41;
if(strcmp(sstate,"DELIVRD")==0)state=44;
if(strcmp(sstate,"EXPIRED")==0)state=42;
if(strcmp(sstate,"DELETED")==0)state=43;
if(strcmp(sstate,"UNDELIV")==0)state=45;
if(strcmp(sstate,"ACCEPTE")==0)state=44;
if(strcmp(sstate,"INVALID")==0)state=40;
return state;
}
char** AddSQL(char** ppStr,int index,char* str)
{
if(str==NULL&&index>0&&index<=10)
{
ppStr=new char*[index+1];
ppStr[0]=new char[3];
sprintf(ppStr[0],"%d",index);
}
else if(str)
{
ppStr[index]=new char[strlen(str)+1];
memcpy(ppStr[index],str,strlen(str));
ppStr[index][strlen(str)]=0;
}
return ppStr;
}
void AddBinLog(char* buffer,int i)
{
int j=0;
FILE* pfLog;
char current[36];
char tp[36];
struct _timeb timebuffer;
char lpFileName[MAX_PATH];
_strdate(current);
_ftime(&timebuffer);
strcpy(tp,ctime(&(timebuffer.time)));
current[2]='_';
current[5]=0;
tp[strlen(tp)-1]=0;
GetModuleFileName(NULL,lpFileName,MAX_PATH);
sprintf(strrchr(lpFileName,'\\')+1,
"Bin%s_%s.txt\0",strrchr(tp,' ')+1,current);
current[2]='-';
tp[strrchr(tp,' ')-tp]=0;
pfLog=fopen(lpFileName,"a");
fprintf(pfLog,"\n");
for(j=0;jOpen(arg.strCnn,"","",adConnectUnspecified));
AddLog("begin to insert the 100,000 msg into database.\n");
for(i=0;i<100000;i++)
{
sprintf(buffer,"exec sp_INSERTmessage '13910304490','test longggggggggggggggggggggggggggggggggg"
"ggggggggggggggggggggggggggggggggggggggggggggggggggggg");
pCnn->Errors->Clear();
pCnn->Execute(buffer,NULL,adExecuteNoRecords);
}
AddLog("After sending the 100,000 msg.\n");
pCnn->Close();
return;
}
catch (_com_error &e)
{
AddLog("Exec Error<%s>.\n",buffer);
AddLog("Description = '%s'\n", (char*) e.Description());
return;
}
#endif
//////////////////////////////////////////////////////////////////////////////////////////
i=InitMsg(BIND,buffer,0,0);
send(s,buffer,i,0);
i=j=0;
ZeroMemory(buffer,1024);
while(j<4)
{
i=recv(s,buffer+j,4-j,0);
j+=i;
if(i<1)break;
}//get the msg header
while(j<(int)GetNByte(buffer,4)&&i>0)
{
i=recv(s,buffer+j,GetNByte(buffer,4)-j,0);
j+=i;
if(i<1)break;
}//get the msg body
AddBinLog(buffer,j);
k=0;
while(1){
k++;
switch(GetNByte(buffer+4,4))
{
case BINDRESP:
if(buffer[12])
{
AddLog("Bind to SMSC error for <%d>\n",buffer[12]);
StopGW();
return;
}
AddLog("Bind result is coming.\n");
arg.isActive=1;
arg.dwRet=0;
break;
case ALIVE://over
i=InitMsg(ALIVERESP,buffer,0,0);
i=send(arg.s,buffer,i,0);
arg.isActive=1;
break;
case ALIVERESP://over
AddLog("Server is alive.\n");
arg.isActive=1;
break;
case SUBMITRESP:
arg.isActive=1;
arg.dwRet--;
if(buffer[20])//submit with error
{
AddLog("submit error for<%d>\n",buffer[20]);
}
else
{
AddLog("Send to: %d\n",GetNByte(buffer+8,4));
}
break;
case UNBINDRESP:
AddLog("Unbind from VAS platform.\n");
break;
case DELIVERY:
if(buffer[75]==0)//deliver the msg to esme
{//no processing to this currently
#if _DEBUG
AddLog("deliver: %.4d-%.4d\n",GetNByte(buffer+12,4),GetNByte(buffer+16,4));
AddLog("dest no: %.21s\n",buffer+20);
AddLog("service id :%.10s\n",buffer+41);
AddLog("src_term: %.21s\n",buffer+54);//src mobile
AddLog("msg : %s\n",buffer+77);
#endif
}
else//the status report to msg from esme
{
AddLog("status to %d-%d\n",GetNByte(buffer+77,4),GetNByte(buffer+81,4));
}
i=InitMsg(DELIVERYRESP,buffer,0,0);
send(arg.s,buffer,i,0);
arg.isActive=1;
break;
case QUERYSTATUSRESP:
AddLog("query msg state is return.\n");
arg.isActive=1;
break;
case CANCELRESP:
//GetID(msg,buffer+12);
AddLog("Cancel the message <%d-%d>.\n",GetNByte(buffer+12,3),GetNByte(buffer+18,2));
break;
}//end of swtich the type of the package
//end of processing a package.
i=InitMsg(com[rand()%count],buffer,"13910304490","test msg for server.",k+162,0,20,atoi(arg.ESMEID));
send(s,buffer,i,0);
i=j=0;
ZeroMemory(buffer,1024);
while(j<4)
{
i=recv(s,buffer+j,4-j,0);
j+=i;
if(i<1)break;
}//get the msg header
while(j<(int)GetNByte(buffer,4)&&i>0)
{
i=recv(s,buffer+j,GetNByte(buffer,4)-j,0);
j+=i;
if(i<1)break;
}//get the msg body
AddBinLog(buffer,j);
}
return;
}