tran.cc

此源代码只用于学习,不得用于其他商业活动 .

			if(connect(iSockfd, &ServerAddr, sizeof(ServerAddr)) < 0)
			#endif
			{
				gv_TranMonitor.ReportStatus("ERROR", "服务器没有连接");
				sprintf(chAlertInfo, "%s%d error:%s", "SocketNO", iSockfd, strerror(errno));
				close(iSockfd);
				iSockfd = -1;
				pthread_mutex_lock(&gv_GobalMutex.AlertMutex);
				WriteAlert(CONNECT_ERROR, chAlertInfo, ALERT_LEVEL_3);
				pthread_mutex_unlock(&gv_GobalMutex.AlertMutex);
				sleep(gv_iSleepTime);
				continue;
			}
		}
		switch(gv_ThreadConfig[(int)iThreadNo].TranType)
		{
			case CMD_GET:
				iReturn = GetFile(iSockfd, &TransData, gv_ThreadConfig[(int)iThreadNo].LocalPath, (int)iThreadNo, &ThreadFileInfo);
				break;

			case CMD_PUT:
				iReturn = PutDir(iSockfd, &TransData, gv_ThreadConfig[(int)iThreadNo].LocalPath, (int)iThreadNo, &ThreadFileInfo);
				break;

			default:
				iReturn = -1;
				break;
		}
		close(iSockfd);
		iSockfd = -1;
		TransData.iStatus = htonl(STATUS_FILEINFO);
		TransData.lFilePos = htonl(0);
		memset(TransData.chFileName, 0, sizeof(TransData.chFileName));
		memset(gv_ThreadConfig[(int)iThreadNo].FileName, 0, sizeof(gv_ThreadConfig[(int)iThreadNo].FileName));
		g_iArrayExitFlag[(int)iThreadNo] = 0;
		if(gv_iTermFlag == 0)
		{
			g_iArrayExitFlag[(int)iThreadNo] = 0;
			pthread_exit(0);
		}
		if(1 == iReturn)
		{
			return 0;
		}
		if(iReturn == -1)
		{
			#ifdef	_DEBUG_FOR_TRAN_
			printf("iSockfd %d in thread %d\n", iSockfd, iThreadNo);
			printf("thread error sleep!!\n");
			#endif

			sleep(gv_iSleepTime);
		}
		else if(iReturn >= 0)
		{
			if(0 == iReturn)
			{
				#ifdef	_DEBUG_FOR_TRAN_
				printf("iSockfd %d in thread %d\n", iSockfd, iThreadNo);
				printf("thread normal sleep!!\n");
				#endif

				sleep(gv_iSleepTime);
			}
			else if(1 == iReturn)
			{
				break;
			}
		}
	}//end for(;;)
	return 0;
}

/****************************************************************
**
**	Function: GetFile
**	Purpose:  从服务器下载文件
** 	Input Parammeters:
			iSockfd -- socket句柄
			data -- 传输数据结构体
			chLocalDir -- 文件存放路径
			iThreadNo -- 线程号
**	Return:	-1 -- 错误  0 -- 成功 1--中断退出
**	datetime:
**
*****************************************************************/
int GetFile(int iSockfd, struct S_DataInfo *data, char *chLocalDir, int iThreadNo, S_FileInfo *ThreadFileInfo)
{
	long lFileLen, lFilePos = 0;
	int iLen, iEnd, iResult;
	char chFile[STRING_LEN];
	char chFullFile[STRING_LEN];
	char chDate[8 + 1], chTime[6 + 1];
	char chStartDateTime[STRING_LEN];
	char chEndDateTime[STRING_LEN];
	char chMsgBuffer[STRING_LEN];
	char chAlertInfo[STRING_LEN];
	FILE *fp = 0;

	g_iArrayExitFlag[iThreadNo] = 1;

	gv_TranMonitor.AddRate(sizeof(S_DataInfo));
	iLen = WriteSockData(iSockfd, data);
	if(-1 == iLen)
	{
		sprintf(chAlertInfo, "%s", data->chFileName);
		pthread_mutex_lock(&gv_GobalMutex.AlertMutex);
		WriteAlert(WRITESOCKDATA_ERROR, chAlertInfo, ALERT_LEVEL_3);
		pthread_mutex_unlock(&gv_GobalMutex.AlertMutex);
		return -1;
	}
	gv_TranMonitor.AddRate(sizeof(S_DataInfo));
	if(GetSockData(iSockfd, data) == -1)
	{
		sprintf(chAlertInfo, "%s", data->chFileName);
		pthread_mutex_lock(&gv_GobalMutex.AlertMutex);
		WriteAlert(GETSOCKDATA_ERROR, chAlertInfo, ALERT_LEVEL_3);
		pthread_mutex_unlock(&gv_GobalMutex.AlertMutex);
		return -1;
	}
	strcpy(gv_ThreadConfig[ThreadFileInfo->m_iThreadNo].FileName, data->chFileName);
	gv_ThreadConfig[ThreadFileInfo->m_iThreadNo].FilePos = ntohl(data->lFilePos);
	WriteReMemory(ThreadFileInfo);
	iEnd = 0;
	while(g_iArrayFlag[iThreadNo] == 1)
	{
		switch(ntohl(data->iStatus))
		{
			/*状态错误*/
			case STATUS_ERROR:
				data->iStatus = htonl(STATUS_SENDFILE);
				data->lFilePos = htonl(lFilePos);
				break;

			/*无文件*/
			case STATUS_NOFILE:
				data->iStatus = htonl(STATUS_BYBY);
				gv_TranMonitor.AddRate(sizeof(S_DataInfo));
				if(WriteSockData(iSockfd, data) == -1)
				{
					sprintf(chAlertInfo, "%s", data->chFileName);
					pthread_mutex_lock(&gv_GobalMutex.AlertMutex);
					WriteAlert(WRITESOCKDATA_ERROR, chAlertInfo, ALERT_LEVEL_3);
					pthread_mutex_unlock(&gv_GobalMutex.AlertMutex);
					return -1;
				}
				iEnd = 1;
				break;

			/*存在文件*/
			case STATUS_FILEEXIST:
				GetCurrentDate(chDate);
				GetCurrentTime(chTime);
				sprintf(chStartDateTime, "%s%s", chDate, chTime);
				data->iStatus = htonl(STATUS_SENDFILE);
				data->lFilePos = htonl(lFilePos);
				break;

			/*现在正在发送文件*/
			case STATUS_SENDFILE:
				if(SaveFile(data, chLocalDir) == -1)
				{
					data->iStatus = htonl(STATUS_BYBY);
					gv_TranMonitor.AddRate(sizeof(S_DataInfo));
					if(WriteSockData(iSockfd, data) == -1)
					{
						sprintf(chAlertInfo, "%s", data->chFileName);
						pthread_mutex_lock(&gv_GobalMutex.AlertMutex);
						WriteAlert(WRITESOCKDATA_ERROR, chAlertInfo, ALERT_LEVEL_3);
						pthread_mutex_unlock(&gv_GobalMutex.AlertMutex);
						return -1;
					}
					strcpy(chAlertInfo, "SaveFile error!");
					pthread_mutex_lock(&gv_GobalMutex.AlertMutex);
					WriteAlert(SAVEFILE_ERROR, chAlertInfo, ALERT_LEVEL_3);
					pthread_mutex_unlock(&gv_GobalMutex.AlertMutex);
					return -1;
				}
				lFilePos += ntohl(data->lDataLen);
				strcpy(gv_ThreadConfig[iThreadNo].FileName,data->chFileName);
				//gv_ThreadConfig[iThreadNo].FilePos = ntohl(data->lFilePos);
				gv_ThreadConfig[iThreadNo].FilePos = lFilePos;
				WriteReMemory(ThreadFileInfo);
				data->iStatus=htonl(STATUS_SENDFILE);
				data->lFilePos=htonl(lFilePos);
				break;

			/*这是最后的文件块*/
			case STATUS_LASTDATA:
				if(SaveFile(data, chLocalDir) == -1)
				{
					data->iStatus = htonl(STATUS_BYBY);
					gv_TranMonitor.AddRate(sizeof(S_DataInfo));
					if(WriteSockData(iSockfd, data) == -1)
					{
						sprintf(chAlertInfo, "%s", data->chFileName);
						pthread_mutex_lock(&gv_GobalMutex.AlertMutex);
						WriteAlert(WRITESOCKDATA_ERROR, chAlertInfo, ALERT_LEVEL_3);
						pthread_mutex_unlock(&gv_GobalMutex.AlertMutex);
						return -1;
					}
					strcpy(chAlertInfo, "SaveFile error!");
					pthread_mutex_lock(&gv_GobalMutex.AlertMutex);
					WriteAlert(SAVEFILE_ERROR, chAlertInfo, ALERT_LEVEL_3);
					pthread_mutex_unlock(&gv_GobalMutex.AlertMutex);
					return -1;
				}
				lFilePos = 0;
				data->iStatus=htonl(STATUS_LASTDATA);
				if(TempToFile(chLocalDir,data->chFileName,chLocalDir,chFile) == -1)
				{
					data->iStatus = htonl(STATUS_BYBY);
					gv_TranMonitor.AddRate(sizeof(S_DataInfo));
					if(WriteSockData(iSockfd, data) == -1)
					{
						sprintf(chAlertInfo, "%s", data->chFileName);
						pthread_mutex_lock(&gv_GobalMutex.AlertMutex);
						WriteAlert(WRITESOCKDATA_ERROR, chAlertInfo, ALERT_LEVEL_3);
						pthread_mutex_unlock(&gv_GobalMutex.AlertMutex);
						return -1;
					}
					strcpy(chAlertInfo, "TempToFile error!");
					pthread_mutex_lock(&gv_GobalMutex.AlertMutex);
					WriteAlert(TEMPTOFILE_ERROR, chAlertInfo, ALERT_LEVEL_3);
					pthread_mutex_unlock(&gv_GobalMutex.AlertMutex);
					return -1;
				}
				strcpy(chFullFile,chLocalDir);
				FullPath(chFullFile);
				strcat(chFullFile,chFile);
				if(1 == gv_ThreadConfig[(int)iThreadNo].CompressFlag)
				{
					UnZipFile(chFullFile);
					chFile[strlen(chFile)-2] = '\0';
				}

				if(1 == gv_ThreadConfig[(int)iThreadNo].MoveFlag)
				{
					iResult = FileToTargetPath(data, iThreadNo);
					if(-1 == iResult)
					{
						data->iStatus = htonl(STATUS_BYBY);
						gv_TranMonitor.AddRate(sizeof(S_DataInfo));
						if(WriteSockData(iSockfd, data) == -1)
						{
							sprintf(chAlertInfo, "%s", data->chFileName);
							pthread_mutex_lock(&gv_GobalMutex.AlertMutex);
							WriteAlert(WRITESOCKDATA_ERROR, chAlertInfo, ALERT_LEVEL_3);
							pthread_mutex_unlock(&gv_GobalMutex.AlertMutex);
							return -1;
						}
						sprintf(chAlertInfo, "File(%s)ToTargetPath(%s) error!", chFile, gv_ThreadConfig[(int)iThreadNo].LocalTargetPath);
						pthread_mutex_lock(&gv_GobalMutex.AlertMutex);
						WriteAlert(FILETOTARGETPATH_ERROR, chAlertInfo, ALERT_LEVEL_3);
						pthread_mutex_unlock(&gv_GobalMutex.AlertMutex);
						return -1;
					}
				}
				/*写传输日志*/
				GetCurrentDate(chDate);
				GetCurrentTime(chTime);
				sprintf(chEndDateTime, "%s%s", chDate, chTime);
				memset(chMsgBuffer, 0, sizeof(chMsgBuffer));
				sprintf(
					chMsgBuffer,
					"%s,%s,%s,%s,%s,%s,%s,%s,%s,%ld,%s,%s,%s,%s,%s,%s,%s,%s",
					LOG_TYPE,
					MODEL_CODE,
					OUT_MODEL_CODE,
					chFile,
					chFile,
					chStartDateTime,
					chEndDateTime,
					gv_chChannelCode,
					OUT_CHANNEL_CODE,
					ntohl(data->lFileLen),
					"\0",
					"\0",
					"s",
					"g",
					gv_chServerIp,
					gv_chClientIp,
					CHANNEL_ATTR,
					"\0"
					);
				pthread_mutex_lock(&gv_GobalMutex.LogMutex);
				gv_TranLog.Write(chMsgBuffer);
				pthread_mutex_unlock(&gv_GobalMutex.LogMutex);
				memset(gv_ThreadConfig[iThreadNo].FileName, 0, sizeof(gv_ThreadConfig[iThreadNo].FileName));
				gv_ThreadConfig[iThreadNo].FilePos = 0;
				WriteReMemory(ThreadFileInfo);
				WriteBackupFile(ThreadFileInfo);
				break;
			/*文件发送完成*/
			case STATUS_FINISH:
				data->iStatus = htonl(STATUS_FILEINFO);
				data->lFilePos = htonl(0);
				memset(data->chFileName, 0, sizeof(data->chFileName));
				break;
			default:
				data->iStatus = htonl(STATUS_BYBY);
				gv_TranMonitor.AddRate(sizeof(S_DataInfo));
				if(WriteSockData(iSockfd, data) == -1)
				{
					sprintf(chAlertInfo, "%s", data->chFileName);
					pthread_mutex_lock(&gv_GobalMutex.AlertMutex);
					WriteAlert(WRITESOCKDATA_ERROR, chAlertInfo, ALERT_LEVEL_3);
					pthread_mutex_unlock(&gv_GobalMutex.AlertMutex);
					return -1;
				}
				sprintf(chAlertInfo, "%s error!", "unknownstatus");
				pthread_mutex_lock(&gv_GobalMutex.AlertMutex);
				WriteAlert(UNKNOWNSTATUS_ERROR, chAlertInfo, ALERT_LEVEL_3);
				pthread_mutex_unlock(&gv_GobalMutex.AlertMutex);
				return -1;
		}//end switch(ntohl(data->iStatus))
		if(iEnd)
		{
			return 0;
		}
		gv_TranMonitor.AddRate(sizeof(S_DataInfo));
		if(WriteSockData(iSockfd, data) == -1)
		{
			sprintf(chAlertInfo, "%s error!", data->chFileName);
			pthread_mutex_lock(&gv_GobalMutex.AlertMutex);
			WriteAlert(WRITESOCKDATA_ERROR, chAlertInfo, ALERT_LEVEL_3);
			pthread_mutex_unlock(&gv_GobalMutex.AlertMutex);
			return -1;
		}
		gv_TranMonitor.AddRate(sizeof(S_DataInfo));
		if(GetSockData(iSockfd, data) == -1)
		{
			sprintf(chAlertInfo, "%s error!", data->chFileName);
			pthread_mutex_lock(&gv_GobalMutex.AlertMutex);
			WriteAlert(GETSOCKDATA_ERROR, chAlertInfo, ALERT_LEVEL_3);
			pthread_mutex_unlock(&gv_GobalMutex.AlertMutex);
			return -1;
		}
	}//end while(g_iArrayFlag[iThreadNo] == 1)
	WriteBackupFile(ThreadFileInfo);
	data->iStatus = htonl(STATUS_BYBY);
	gv_TranMonitor.AddRate(sizeof(S_DataInfo));
	if(WriteSockData(iSockfd, data) == -1)
	{
		sprintf(chAlertInfo, "%s error!", data->chFileName);
		pthread_mutex_lock(&gv_GobalMutex.AlertMutex);
		WriteAlert(WRITESOCKDATA_ERROR, chAlertInfo, ALERT_LEVEL_3);
		pthread_mutex_unlock(&gv_GobalMutex.AlertMutex);
		return -1;
	}
	return 1;
}
/****************************************************************
**
**	Function: PutDir
**	Purpose:  上传目录中指定的文件到服务器
** 	Input Parammeters:
			iSockfd -- socket句柄
			data -- 传输数据结构体指针
			chDir -- 需上传文件目录名称
			iThreadNo--线程的索引号
			ThreadFileInfo--线程数据结构体
**	Return:	-1 -- 错误  0 -- 成功
**	datetime:
**
*****************************************************************/
int PutDir(int iSockfd, struct S_DataInfo *data, const char *chDir, int iThreadNo, S_FileInfo *ThreadFileInfo)
{
	char chFullName[STRING_LEN];
	char chShortName[STRING_LEN];
	char chDate[8+1], chTime[6+1];
	char chMsgBuffer[STRING_LEN];
	char chStartDateTime[STRING_LEN];
	char chEndDateTime[STRING_LEN];
	char chFileName[STRING_LEN];
	char chAlertInfo[STRING_LEN];
	long lZipFileLen = 0, lFileLen = 0;
	int iResult = 0;

	g_iArrayExitFlag[iThreadNo] = 1;

	for(;;)
	{
		if(strlen(data->chFileName) != 0)
		{
			//断点续传
			strcpy(chFullName, chDir);
			strcat(chFullName, data->chFileName);
			if(!access(chFullName, F_OK))
			{
				if(IsZipFile(data->chFileName))
				{
					iResult = lZipFileLen = GetFileSize(chFullName);
					if(-1 == lZipFileLen)
					{
						sprintf(chAlertInfo, "%s", "GetFileSize error!");
						pthread_mutex_lock(&gv_GobalMutex.AlertMutex);
						WriteAlert(GETFILESIZE_ERROR, chAlertInfo, ALERT_LEVEL_3);
						pthread_mutex_unlock(&gv_GobalMutex.AlertMutex);
						break;
					}
					data->lFileLen = htonl(lZipFileLen);
					data->lZipSize = htonl(lZipFileLen);
				}
				else
				{
					iResult = lFileLen = GetFileSize(chFullName);
					if(-1 == lFileLen)
					{
						sprintf(chAlertInfo, "%s", "GetFileSize error!");
						pthread_mutex_lock(&gv_GobalMutex.AlertMutex);
						WriteAlert(GETFILESIZE_ERROR, chAlertInfo, ALERT_LEVEL_3);
						pthread_mutex_unlock(&gv_GobalMutex.AlertMutex);
						break;
					}
					data->lFileLen = htonl(lFileLen);
				}
			}
			else
			{
				iResult = -1;
				sprintf(chAlertInfo, "%s error!", data->chFileName);
				pthread_mutex_lock(&gv_GobalMutex.AlertMutex);
				WriteAlert(FILENOTEXIST_ERROR, chAlertInfo, ALERT_LEVEL_3);
				pthread_mutex_unlock(&gv_GobalMutex.AlertMutex);
				break;
			}
		}
		else
		{
			pthread_mutex_lock(&gv_GobalMutex.SearchFileMutex);
			iResult = SearchOneFile(chDir, data->chFileFilter, chShortName);
			if(iResult < 0)
			{
				pthread_mutex_unlock(&gv_GobalMutex.SearchFileMutex);
				sprintf(chAlertInfo, "%s error!", data->chFileFilter);
				pthread_mutex_lock(&gv_GobalMutex.AlertMutex);
				WriteAlert(SEARCHONEFILE_ERROR, chAlertInfo, ALERT_LEVEL_3);
				pthread_mutex_unlock(&gv_GobalMutex.AlertMutex);
				break;
			}
			else if(0 == iResult)
			{
				pthread_mutex_unlock(&gv_GobalMutex.SearchFileMutex);
//				sprintf(chAlertInfo, "%s error!", data->chFileFilter);
//				pthread_mutex_lock(&gv_GobalMutex.AlertMutex);
//				WriteAlert(FILENOTEXIST_ERROR, chAlertInfo, ALERT_LEVEL_9);
//				pthread_mutex_unlock(&gv_GobalMutex.AlertMutex);
				break;
			}
			strcpy(chFullName, chDir);
			FullPath(chFullName);
			if((iResult = FileToTemp(chFullName, chShortName, chFullName, chShortName)) == -1)
			{
				pthread_mutex_unlock(&gv_GobalMutex.SearchFileMutex);
				sprintf(chAlertInfo, "%s error!", chShortName);
				pthread_mutex_lock(&gv_GobalMutex.AlertMutex);
				WriteAlert(FILETOTEMP_ERROR, chAlertInfo, ALERT_LEVEL_3);
				pthread_mutex_unlock(&gv_GobalMutex.AlertMutex);
				break;
			}
			pthread_mutex_unlock(&gv_GobalMutex.SearchFileMutex);
			strcat(chFullName, chShortName);
			if(COMPRESS == ntohl(data->iCompress))
			{
				ZipFile(chFullName);
				strcat(chShortName, ".Z");
				strcat(chFullName, ".Z");
			}
			data->iStatus = htonl(STATUS_FILEINFO);
			strcpy(data->chFileName, chShortName);
			iResult = lFileLen = GetFileSize(chFullName);
			if(-1 == lFileLen)
			{
				sprintf(chAlertInfo, "%s", "GetFileSize error!");
				pthread_mutex_lock(&gv_GobalMutex.AlertMutex);
				WriteAlert(GETFILESIZE_ERROR, chAlertInfo, ALERT_LEVEL_3);
				pthread_mutex_unlock(&gv_GobalMutex.AlertMutex);
				break;
			}
			data->lFileLen = htonl(lFileLen);
		}//end if(strlen(data->chFileName) != 0)
		memset(chDate, 0, sizeof(chDate));