awflow.cpp

一个多线程

}

#define API_CMD_TIMEOUT_MSG 200

void CAWFlow::OnTimer(int iNum,DWORD dwIdent)
{
	switch(dwIdent){
	case FlowSleepTimerId:
		FunctionRet(TRUE,APIBUSFlowSleep);
		break;
	case FlowWaitTimerOut:
		if(m_iFlowTimeOutSec){
			m_iFlowTimeOutSec=0;
			FunctionRet(TRUE,APIBUSSetTimeOut);
		}
	default:
		if(m_btCurRoad!=m_pBus->m_iTimeRoad)
			AddCommand(TRUE,API_CMD_TIMEOUT_MSG,0,0,0,0,UniVar(dwIdent-FlowCmdTimerSet),UniVar(iNum*TIME_ELEAPS/1000),InValidVar);
		else
			AddCommand(FALSE,API_CMD_TIMEOUT_MSG,0,0,0,0,UniVar(dwIdent-FlowCmdTimerSet),UniVar(iNum*TIME_ELEAPS/1000),InValidVar);
		break;
	}
}

void CAWFlow::OnLockSuccess(HANDLE hkey,CAsynWorker * pWorker,DWORD dwID)
{
	if(dwID==FLOWLOCKID&&pWorker==this){
		FunctionRet(TRUE,APIBUSLockOwn,UniVar((int)hkey),InValidVar);
	}
	else{
		CAsynWorker::OnLockSuccess(hkey,pWorker,dwID);
	}
}

void CAWFlow::OnMsgCmd(CUniMsg *pMsg)
{
	if(m_dwFlowWaitMach==(int)pMsg->m_dwSendMode&&m_iFlowWaitObject==(int)pMsg->m_pSender&&(m_iFlowWaitCommand==0xffffffff||m_iFlowWaitCommand==(int)pMsg->m_dwCommand)&&m_pCurMsg){
		CAsynWorker * pWorker=m_pCurMsg->m_pSender;
		m_pCurMsg->Clear();
		m_pCurMsg->m_pSender=pWorker;
		m_pCurMsg->m_pReceivor=pMsg->m_pReceivor;
		for(int i=0;i<pMsg->GetSize();i++){
			m_pCurMsg->Add(pMsg->GetAt(i));
		}
		m_iFlowWaitObject=0;
		m_iFlowWaitCommand=0;
		if(m_dwPauseCommand==APIBUSSendEventEx||m_dwPauseCommand==APIBUSSendRespEx){
			FunctionRet(TRUE,m_dwPauseCommand);
		}
		return;
	}
	else{
		if((CAsynWorker *)this==pMsg->m_pSender)
			AddCommand(TRUE,pMsg->m_dwCommand,(DWORD)pMsg->m_pSender,0,0,pMsg->m_dwSendMode,pMsg->GetSize(),pMsg->m_pTV);
		else{
			AddCommand(TRUE,pMsg->m_dwCommand,(DWORD)pMsg->m_pSender,2,pMsg->m_dwIdent,pMsg->m_dwSendMode,pMsg->GetSize(),pMsg->m_pTV);
		}
	}
}

void CAWFlow::OnMsgResp(CUniMsg * resp,CUniMsg * cmd)
{
	if(resp->m_dwIdent==m_dwFlowCommadnIDent){
		switch(resp->m_dwCommand){
		case CMD_TASK_RUNTASK:
			{
			if(m_dwPauseCommand==APIBUSFlowEdau||m_dwPauseCommand==APIBUSExcuteOdbc){
				try{
					if(resp->m_ierror){
						m_iErrorCode=-25000;
						RestoreCurrentParam(UniVar(-25000),3);
					}
					else{
						m_iErrorCode=(int)resp->GetAt(3);
						RestoreCurrentParam(resp->GetAt(0),3);
						RestoreCurrentParam(resp->GetAt(1),4);
						if(resp->GetSize()>2){
							RestoreCurrentParam(resp->GetAt(4),5);
							RestoreCurrentParam(resp->GetAt(3),6);
							int ip=(int)resp->GetAt(2);
							for(int i=5;i<ip+5;i++)
								RestoreCurrentParam(resp->GetAt(i),7+i);
							int irow=(int)resp->GetAt(4);
							int icol=(int)resp->GetAt(3);
							if(irow&&icol){
								m_DefaultResult.m_ierror=m_iErrorCode;
								m_DefaultResult.m_iTimeStamp=icol;
								m_DefaultResult.m_hKey=(void *)irow;
								for(i=0;i<irow;i++){
									for(int j=0;j<icol;j++){
										if(resp->GetSize()>5+ip+i*icol+j)
											m_DefaultResult.Add(resp->GetAt(5+ip+i*icol+j));
									}
								}
							}
							else{
								m_DefaultResult.Clear();
							}
						}
						else {
							m_DefaultResult.Clear();
						}
					}
				}
				catch(...){
				}
				FunctionRet(TRUE,m_dwPauseCommand);
				return;
			}
			break;
		}
		case CMD_TASK_DLLPROXY:
			{
			if(m_dwPauseCommand==APIBUSAsynDllCall){
				for(int i=0;i<resp->GetSize();i++){
					RestoreCurrentParam(resp->GetAt(i),i+3);
				}
				FunctionRet(TRUE,m_dwPauseCommand);
				return;
			}
			break;
		}
		case CMD_FREE_RES:
		case CMD_LOCK_RES:
		case CMD_APPLY_RES:
			{
				switch(m_dwPauseCommand){
				case APIBUSFreeRes:
				case APIBUSAddRes:
				case APIBUSLockRes:
				case APIBUSApplyResbySkill:
				case APIBUSApplyResbyGroup:
					RestoreCurrentParam(resp->GetAt(0),0);
					FunctionRet(TRUE,m_dwPauseCommand);
					return;
				default:
					if(resp->m_dwCommand==CMD_APPLY_RES){
						AddCommand(TRUE,resp->m_dwCommand,(DWORD)resp->m_pSender,0,0,resp->m_dwSendMode,resp->GetSize(),resp->m_pTV);
					}
					break;
				}
			break;
		}
		default:
			{
			try{
				CAsynWorker * pWorker=m_pCurMsg->m_pSender;
				m_pCurMsg->Clear();
				m_pCurMsg->m_pSender=pWorker;
				m_pCurMsg->m_pReceivor=resp->m_pReceivor;
				for(int i=0;i<resp->GetSize();i++){
					m_pCurMsg->Add(resp->GetAt(i));
				}
			}
			catch(...){
			}
			if(m_dwPauseCommand==APIBUSSendCommandEx){
				if(resp->m_ierror)
					m_iErrorCode=-25000;
				m_dwFlowCommadnIDent=0xffffffff;
				FunctionRet(TRUE,APIBUSSendCommandEx);
				return;
			}
			else {
				m_dwFlowCommadnIDent=0xffffffff;
			}
			break;
		}
		}
	}
	else {
		CAsynWorker::OnMsgResp(resp,cmd);
	}
}

void CAWFlow::OnCreate()
{
}

void CAWFlow::OnRelease(){
	try{
		if(!m_bRelease){
			m_bRelease=TRUE;
			if(!m_dwCreateMod||m_dwCreateMod==GetModId()){
				SendEvent(m_dwCreateMod,m_pCreator,EVT_WORKERFLOW_RELEASE,NULL,-1,UniVar((int)this),InValidVar);
				UnRegister();
				if(m_bAutoDelete)
					delete this;
			}
			else{
				if(m_pCreator->IsValid()){
					SendEvent(m_pCreator,EVT_WORKERFLOW_RELEASE,NULL,-1,UniVar((int)this),InValidVar);
				}
				else{
					UnRegister();
					if(m_bAutoDelete)
						delete this;
				}
			}
		}
	}
	catch(...){
		TRACE("release error");
	}
};

void CAWFlow::OnMsgEvent(CUniMsg * pMsg)
{
	if(m_dwFlowWaitMach==(int)pMsg->m_dwSendMode&&m_iFlowWaitObject==(int)pMsg->m_pSender&&(m_iFlowWaitCommand==0xffffffff||m_iFlowWaitCommand==(int)pMsg->m_dwCommand)){
		CAsynWorker * pWorker;
		if(!m_pCurMsg){
			pWorker=pMsg->m_pSender;
			if(!m_pCurMsg)m_pCurMsg=new CUniMsg();
		}
		else {
			pWorker=m_pCurMsg->m_pSender;
		}
		m_pCurMsg->Clear();
		m_pCurMsg->m_pSender=pWorker;
		m_pCurMsg->m_pReceivor=pMsg->m_pReceivor;
		for(int i=0;i<pMsg->GetSize();i++){
			m_pCurMsg->Add(pMsg->GetAt(i));
		}
		m_iFlowWaitObject=0;
		m_iFlowWaitCommand=0;
		if(m_dwPauseCommand==APIBUSSendEventEx||m_dwPauseCommand==APIBUSSendRespEx){
			FunctionRet(TRUE,m_dwPauseCommand);
		}
		return;
	}
	else{
		switch(pMsg->m_dwCommand){
		case EVT_EXCUTE_APPLY:
			{
				AddExprObj(pMsg->m_pSender,(DWORD)pMsg->m_pTV[0],(DWORD)pMsg->m_pTV[1],pMsg->m_dwSendMode);
			}
			return;
		case EVT_GETOBJ_OVER:
			{
				FunctionRet(TRUE,APIBUSGetRegistedWorker,pMsg->m_pTV,2);
			}
			return;
		case EVT_WORKERFLOW_RELEASE:
			{
				if(pMsg->m_pSender->IsValid()){
					m_pBus->UnRegister(pMsg->m_pSender);
					CAWFlow * pfw=(CAWFlow *)pMsg->m_pSender;
					delete pfw;
				}
				Sleep(5);
				FunctionRet(TRUE,APIBUSReleaseFlow);
				return;
			}
		case EVENT_SET_CLOSE_FLOW_EVENT:
			{
				Sleep(5);
				WaitForSingleObject(m_hMutex,1000);
				try{
					for(int i=0;i<m_CommandList.GetSize();i++)
						delete m_CommandList[i];
					m_CommandList.RemoveAll();
					for(i=0;i<m_ResultList.GetSize();i++)
						delete m_ResultList[i];
					m_ResultList.RemoveAll();
				}
				catch(...){
				}
				ReleaseMutex(m_hMutex);
				WaitStop();
				WaitForSingleObject(m_hMutex,1000);
				try{
					if(!SetEvent(0x00f00000,UniVar(pMsg->m_dwSendMode),UniVar((DWORD)pMsg->m_pReceivor))){
						ReleaseMutex(m_hMutex);
						try{
							m_bRelease=TRUE;
							if(!m_dwCreateMod||m_dwCreateMod==GetModId()){
								SendEvent(m_dwCreateMod,m_pCreator,EVT_WORKERFLOW_RELEASE,NULL,-1,UniVar((int)this),InValidVar);
								UnRegister();
								if(m_bAutoDelete)
									delete this;
							}
							else{
								if(m_pCreator->IsValid()){
									SendEvent(m_pCreator,EVT_WORKERFLOW_RELEASE,NULL,-1,UniVar((int)this),InValidVar);
									return ;
								}
								else{
									UnRegister();
									if(m_bAutoDelete)
										delete this;
									return ;
								}
							}
						}
						catch(...){
							return;
						}
					}
				}
				catch(...){
				}
				ReleaseMutex(m_hMutex);
				return;
			}
		default:
			{
				if((CAsynWorker *)this==pMsg->m_pSender){
					AddCommand(TRUE,pMsg->m_dwCommand,(DWORD)pMsg->m_pSender,0,0,pMsg->m_dwSendMode,pMsg->GetSize(),pMsg->m_pTV);
				}
				else{
					AddCommand(TRUE,pMsg->m_dwCommand,(DWORD)pMsg->m_pSender,2,pMsg->m_dwIdent,pMsg->m_dwSendMode,pMsg->GetSize(),pMsg->m_pTV);
				}
				return;
			}
		}
	}
}

CAWFlow::CAWFlow(CAWFlow * pCreator,DWORD dwCreateMod)
{
	m_iFlowTimeOutSec=0;
	m_pCreator=pCreator;
	m_dwCreateMod=dwCreateMod;
	m_dwPauseCommand=0xffffffff;
	m_pCurMsg=NULL;
	m_hMutex=CreateMutex(NULL,FALSE,NULL);
	m_bNeedCommand=FALSE;
	m_dwFlowWaitMach=0;
	m_bRelease=0;
	m_dwFlowCommadnIDent=0;
	m_iFlowWaitObject=m_iFlowWaitCommand=m_iFlowWaitType=0;
	m_dwFlowCreateId=0;
	m_nNumExprObj=0;
	m_nMaxExprObj=0;
	m_pExprObj=NULL;
	m_bAutoDelete=TRUE;
};

CAWFlow::~CAWFlow()
{
	if(m_pCurMsg)delete m_pCurMsg;
	CloseHandle(m_hMutex);
	for(int i=0;i<m_CommandList.GetSize();i++){
		delete m_CommandList[i];
	};
	for(i=0;i<m_ResultList.GetSize();i++)
		delete m_ResultList[i];
	m_ResultList.RemoveAll();
	if(m_pExprObj){
		delete [] m_pExprObj;
	}
	m_bAutoDelete=FALSE;
};

BOOL CAWFlow::AddExprObj(CAsynWorker * pFlow,DWORD dwFirstCmd,DWORD dwLastCmd,DWORD dwModId)
{
	if(!dwModId){
		CAsynWorker * pWork=pFlow;
		if(!pWork->IsValid()){
			return FALSE;
		}
	}
	if(m_nNumExprObj+1<m_nMaxExprObj){
		for(int i=0;i<m_nMaxExprObj;i++){
			if(m_pExprObj[i].pFlow==0){
				m_pExprObj[i].pFlow=pFlow;
				m_pExprObj[i].dwFirstCmd=dwFirstCmd;
				m_pExprObj[i].dwLastCmd=dwLastCmd;
				m_pExprObj[i].dwModId=dwModId;
				m_nNumExprObj++;
			}
		}
	}
	else{
		m_nMaxExprObj+=5;
		m_nNumExprObj++;
		EXPROBJECT * pObj=new EXPROBJECT[m_nMaxExprObj];
		memset(pObj,0,m_nMaxExprObj*sizeof(EXPROBJECT));
		pObj[m_nMaxExprObj-5].pFlow=pFlow;
		pObj[m_nMaxExprObj-5].dwFirstCmd=dwFirstCmd;
		pObj[m_nMaxExprObj-5].dwLastCmd=dwLastCmd;
		pObj[m_nMaxExprObj-5].dwModId=dwModId;
		if(m_pExprObj){
			memcpy(pObj,m_pExprObj,(m_nMaxExprObj-5)*sizeof(EXPROBJECT));
			delete [] m_pExprObj;
		}
		m_pExprObj=pObj;
	};
	return TRUE;
}

void CAWFlow::RemoveExprObj(CAsynWorker * pFlow,DWORD dwModId)
{
	for(int i=0;i<m_nMaxExprObj;i++){
		if(m_pExprObj[i].pFlow==pFlow&&m_pExprObj[i].dwModId==dwModId){
			m_pExprObj[i].pFlow=NULL;
			m_nNumExprObj--;
		}
	}
}

BOOL CAWFlow::GetExprObj(DWORD dwCmd,CAsynWorker * & pFlow,DWORD & dwModId)
{
	for(int i=0,j=0;i<m_nMaxExprObj&&j<m_nNumExprObj;i++){
		if(m_pExprObj[i].pFlow!=0){
			j++;
			if(m_pExprObj[i].dwFirstCmd<=dwCmd&&m_pExprObj[i].dwLastCmd>=dwCmd){
				pFlow=m_pExprObj[i].pFlow;
				dwModId=m_pExprObj[i].dwModId;
				return TRUE;
			}
		}
	}
	return FALSE;
}

COMMANDRESULT CAWFlow::ExcuteDirect(DWORD dwCommand,DWORD dwObj,DWORD dwModId,int iNum,UniVar * pVar)
{
	AddCommand(FALSE,dwCommand,dwObj,3,0,dwModId,iNum,pVar);
	return LINE_PAUSE;
}

void CAWFlow::OnExcuteOver(int ierror,DWORD dwCommand,UniVar * pVar,int iNum)
{
	if(m_dwCommandExcute==dwCommand){
		m_iErrorCode=ierror;
		FunctionRet(TRUE,m_dwCommandExcute,pVar,iNum);
	}
}