GrpcPrint/PrintS/Communication/BaseClient.cpp

#include "BaseClient.h"
#include "S7Command.h"
#include "../global.h"
#include "../Config/ConfigManager.h"

BaseClient::BaseClient()
	:m_Thread(INVALID_HANDLE_VALUE)
	,m_RunFlag(false)
	//, m_interval(500)
{
	InitializeCriticalSection(&m_ValueCS);
	InitializeCriticalSection(&m_RtcCS);
}


BaseClient::~BaseClient()
{
	
	//Close();
	//if (m_Client.IsOpen())m_Client.Close();
	EnterCriticalSection(&m_RtcCS);
	while (!m_RTCommands.empty())
	{
		Command* pcommand = m_RTCommands.front();
		delete pcommand;
		pcommand = NULL;
		m_RTCommands.pop();
	}
	LeaveCriticalSection(&m_RtcCS);
	while (!m_CycleCommands.empty()) {
		Command* pcommand = m_CycleCommands.back();
		delete pcommand;
		pcommand = NULL;
		m_CycleCommands.pop_back();
	}

	EnterCriticalSection(&m_ValueCS);
	for (auto item = m_baseMp.begin(); item != m_baseMp.end(); ++item) {
		delete item->second;
	}
	m_baseMp.clear();
	LeaveCriticalSection(&m_ValueCS);

	DeleteCriticalSection(&m_ValueCS);
	DeleteCriticalSection(&m_RtcCS);

	
}

void BaseClient::Startup()
{
	m_RunFlag = true;
	m_Thread = AtlCreateThread(ThreadProc, this);
}

void BaseClient::Shutdown()
{
	m_RunFlag = false;
	if (m_Thread != INVALID_HANDLE_VALUE)
	{
		if (WaitForSingleObject(m_Thread, 1000) == WAIT_TIMEOUT) {
			TerminateThread(m_Thread, 1);
		}
		CloseHandle(m_Thread);
		m_Thread = INVALID_HANDLE_VALUE;
	}
}

void BaseClient::SetRunFlag(bool runflag)
{
	m_RunFlag = runflag;
}

DWORD WINAPI BaseClient::ThreadProc(BaseClient* pclient)
{
	if (pclient) {
		pclient->RunProc();
	}
	return 0;
}

void BaseClient::ClearCycleCommand()
{
	if (m_Thread != INVALID_HANDLE_VALUE)return;
	while (!m_CycleCommands.empty())
	{
		Command* command = m_CycleCommands.back();
		m_CycleCommands.pop_back();
		delete command;
	}
	m_CycleCommands.clear();
}

Command* BaseClient::GetSendCycleCommand()
{
	Command* command = NULL;
	for (size_t i = 0; i < m_CycleCommands.size(); ++i) {
		if (!m_CycleCommands[i]->isFinished)
		{
			command = m_CycleCommands[i];
			break;
		}
	}
	return command;
}

void BaseClient::UpdateCycleCommands()
{
	uint64_t currentTickcount = GetTickCount64();
	uint64_t dif = currentTickcount - m_LastCycleTickcount;
	if (dif < GetInterval())return;
	m_LastCycleTickcount = currentTickcount;

	bool isAllFinished = true;
	for (size_t i = 0; i < m_CycleCommands.size(); ++i) {
		if (!m_CycleCommands[i]->isFinished) {
			isAllFinished = false;
			break;
		}
	}
	if (isAllFinished)
	{
		for (size_t i = 0; i < m_CycleCommands.size(); ++i) {
			m_CycleCommands[i]->isFinished = false;
		}
	}
}

void  BaseClient::InsertMp(void* startPtr, size_t count, const string& suff) {
	size_t ptrSize = sizeof(nullptr);   //指针大小
	for (size_t i = 0; i < count; ++i) {
		BaseData* bd = *((BaseData**)((char*)startPtr + ptrSize * i));
		string key = bd->GetCode() + suff;
		if (m_baseMp.find(key) != m_baseMp.end()) {
			printf("%s is repeated...\n", key.data());
		}
		else { m_baseMp.insert(make_pair(key, bd)); }
	}
}

void BaseClient::SendToClients(WRITETYPE type,const string& addKey) {
	list<Item> its;

	EnterCriticalSection(&m_ValueCS);
	auto baseItem = m_baseMp.begin();
	while (baseItem != m_baseMp.end()) {
		its.emplace_back(Item{ baseItem->first+ addKey,baseItem->second->GetValueStr(),baseItem->second->GetDataType() });
		++baseItem;
	}
	LeaveCriticalSection(&m_ValueCS);
	ClientWrapper::Instance()->PushAllClient(WriteData(type, its));

	
	//auto cfgItem = m_alarmCfgMp.begin();
	//while (cfgItem != m_alarmCfgMp.end()) {
	//	its.emplace_back(Item{ cfgItem->first,cfgItem->second->m_AlarmInfo,DATATYPE::iSTRING});
	//	++baseItem;
	//}
	//ClientWrapper::Instance()->PushAllClient(new  WriteData(type, its));
}



S7Client::S7Client(CommunicationCfg* pconfig)
:m_S7Client(nullptr)
,m_Config(pconfig)
, m_WriteTimeout(1000)
,m_ReadTimeout(2000)
, m_LocalTsap(0x0200)
,m_RemoteTsap(0x0200)
{
}

S7Client::~S7Client() {
	Shutdown();
	if (m_S7Client)
	{
		delete m_S7Client;
	}
}

void S7Client::Shutdown()
{
	BaseClient::Shutdown();
	if (m_S7Client && m_S7Client->Connected)m_S7Client->Disconnect();
}

void S7Client::Init()
{
	//m_interval=m_Config->m_interval;
	m_S7Client = new TSnap7Client();
	m_S7Client->SetConnectionParams(m_Config->m_IP.c_str(), 0x0200, 0x0200);
	int sendtimeout = m_WriteTimeout;
	int readtimeout = m_ReadTimeout;
	int pdusize = 800;
	m_S7Client->SetParam(p_i32_PDURequest, &pdusize);
	m_S7Client->SetParam(p_i32_SendTimeout, &sendtimeout);
	m_S7Client->SetParam(p_i32_RecvTimeout, &readtimeout);
#ifdef _DEBUG
	int pingTimeout = 0;
	m_S7Client->SetParam(p_i32_PingTimeout, &pingTimeout);
#endif
	InitCommand();
}

void S7Client::RunProc()
{
	while (m_RunFlag)
	{
		if (!m_S7Client->Connected) {

			m_S7Client->SetConnectionParams(m_Config->m_IP.c_str(), 0x0200, 0x0200);
			int rel = m_S7Client->Connect();
			
			if (rel != 0) {
				int count = 0;
				while (m_RunFlag && (count < 5)) {
					Sleep(100);
					count++;
				}
				continue;
			}

		}
		Command* pCommand = NULL;
		EnterCriticalSection(&m_RtcCS);
		if (!m_RTCommands.empty()) {
			pCommand = m_RTCommands.front();
			S7Command* ps7 = (S7Command*)pCommand;
			if (ps7->m_waitTime > 0) {
				uint64_t nowtick = GetTickCount64();
				if ((nowtick - ps7->m_CreateTime) > ps7->m_waitTime) {
					m_RTCommands.pop();
				}
				else {
					pCommand = NULL;
				}
			}
			else {
				m_RTCommands.pop();
			}
		}
		LeaveCriticalSection(&m_RtcCS);

		if (pCommand == NULL) {
			pCommand = GetSendCycleCommand();
		}
		if (pCommand != NULL) {

			switch (pCommand->m_id)
			{
			case S7_COMMAND_READ_INFO:
			{
				S7Command* ps7 = (S7Command*)pCommand;
				TS7DataItem* items = ps7->getDataItems();
				int rel = m_S7Client->ReadMultiVars(items, ps7->getItemCount());
				if (rel == 0)
				{
					EnterCriticalSection(&m_ValueCS);
					m_BaseStat.readTimeoutFlag = 0;
					m_BaseStat.isConnected = true;
					LeaveCriticalSection(&m_ValueCS);
					if (ps7->m_Fun)ps7->m_Fun(ps7->m_Ref, ps7);
				}
				else if (rel == errIsoRecvPacket) {
					EnterCriticalSection(&m_ValueCS);
					m_BaseStat.readTimeoutFlag++;
					if (m_BaseStat.readTimeoutFlag >= m_Config->m_AlarmTimeoutTimes) {
						m_BaseStat.isConnected = false;
						m_S7Client->Disconnect();
					}
					LeaveCriticalSection(&m_ValueCS);
				}
				else {
					EnterCriticalSection(&m_ValueCS);
					m_BaseStat.readTimeoutFlag = 0;
					m_BaseStat.isConnected = false;
					m_S7Client->Disconnect();
					LeaveCriticalSection(&m_ValueCS);
				}
			}break;
			case S7_COMMAND_WRITE:
			{
				S7Command* ps7 = (S7Command*)pCommand;
				TS7DataItem* items = ps7->getDataItems();
				//TS7DataItem item = items[0];
				if (ps7->m_PreWrite)ps7->m_PreWrite(ps7->m_Ref, ps7);
				unsigned int itemCount = ps7->getItemCount();
				int rel = m_S7Client->WriteMultiVars(items, itemCount);
				
				if (rel == 0)
				{
					if (ps7->m_Fun)ps7->m_Fun(ps7->m_Ref, ps7);
				}
				else {
					EnterCriticalSection(&m_ValueCS);
					m_BaseStat.isConnected = false;
					m_S7Client->Disconnect();
					LeaveCriticalSection(&m_ValueCS);
				}

			}break;
			}
			pCommand->isFinished = true;
			if (pCommand->isNeedDel)delete pCommand;
		}
		UpdateCycleCommands();
		Sleep(50);
	}
}

ComClient::ComClient(CommunicationCfg* cfg)
:m_Config(cfg)
,m_Freq(50)
,m_ReadTimeout(500)
,m_WriteTimeout(250)
{
	
}

ComClient::~ComClient() {
	Shutdown();
}

void ComClient::Shutdown()
{
	BaseClient::Shutdown();
	if (m_ComClient.IsOpen())m_ComClient.Close();
}

void ComClient::Init()
{
//	m_interval = m_Config->m_interval;
	InitCommand();
	InitSerialPort();
}

void ComClient::InitSerialPort()
{
	m_ComClient.Close();
	try {
		m_ComClient.Open(m_Config->m_COM.c_str(), m_Config->m_BaudRate, CSerialPort2::Parity(m_Config->m_Parity),m_Config->m_DataBits, CSerialPort2::StopBits(m_Config->m_StopBits-1));
	}
	catch (CSerialException& cse) {
		(void)cse;
	}
	if (m_ComClient.IsOpen()) {
		COMMTIMEOUTS timeouts = { 0 };
		m_ComClient.GetTimeouts(timeouts);
		//timeouts.ReadIntervalTimeout = 50;
		timeouts.ReadIntervalTimeout = 0L;
		timeouts.ReadTotalTimeoutMultiplier = 0L;
		timeouts.ReadTotalTimeoutConstant = m_ReadTimeout;
		timeouts.WriteTotalTimeoutConstant = m_WriteTimeout;
		timeouts.WriteTotalTimeoutMultiplier = 0L;
		m_ComClient.SetTimeouts(timeouts);
	}
}

void ComClient::RunProc()
{
	unsigned char sendBuffer[1024];
	unsigned char recvBuffer[1024];
	while (m_RunFlag) {
		if (!m_ComClient.IsOpen())
		{
			InitSerialPort();
			if (!m_ComClient.IsOpen()) {
				ServerDisconnectProc();
				int count = 0;
				while ((count < 10) && m_RunFlag) {
					Sleep(100);
					count++;
				}
				continue;
			}
			else {
				ServerConnectedProc();
			}
		}
		CycleBegin();
		Command* pCommand = NULL;
		EnterCriticalSection(&m_RtcCS);
		if (!m_RTCommands.empty()) {
			pCommand = m_RTCommands.front();
			m_RTCommands.pop();
		}
		LeaveCriticalSection(&m_RtcCS);

		if (pCommand == NULL) {
			pCommand = GetSendCycleCommand();
		}
		if (pCommand != NULL) {
			if (pCommand->isNeedSend) {
				int bytecount = pCommand->GetRequestSequence(sendBuffer);
				DWORD wlength = 0;
				m_ComClient.ClearReadBuffer();
				m_ComClient.ClearWriteBuffer();
				m_ComClient.Write(sendBuffer, bytecount, wlength);
				if (wlength > 0L) {
					WriteSuccessProc(wlength, sendBuffer, pCommand);
					if (pCommand->isNeedRead) {
						DWORD rlength = 0L;
						memset(recvBuffer, 0, sizeof(recvBuffer));
						m_ComClient.Read(recvBuffer, pCommand->ConstReturnSize(), rlength);
						if (rlength > 0L) {
							ReadSuccessProc(rlength, recvBuffer, pCommand);
							if (pCommand->Verify(recvBuffer, rlength))
							{
								pCommand->m_RespSeq = recvBuffer;
								pCommand->m_RespLen = rlength;
								if (pCommand->m_Fun)pCommand->m_Fun(pCommand->m_Ref, pCommand);
							}
							else {
								VerifyFaild(pCommand);
							}
						}
						else if (rlength == 0L) {
							//timeout
							ReadTimeoutProc(pCommand);
						}
						else {
							m_ComClient.Close();
							//g_log->TraceError("comclient close");
						}
					}
				}
				else if (wlength == 0L) {
					//timeout
					WriteTimeoutProc(pCommand);
				}
				else {
					//error
					m_ComClient.Close();
					//g_log->TraceError("comclient close");
				}
			}

			pCommand->isFinished = true;
			if (pCommand->isNeedDel) {
				delete pCommand;
				pCommand = NULL;
			}
		}
		UpdateCycleCommands();
		Sleep(m_Freq);
	}
}

void ComClient::ServerDisconnectProc() {
	EnterCriticalSection(&m_ValueCS);
	m_BaseStat.isConnected = false;
	LeaveCriticalSection(&m_ValueCS);
}


void ComClient::WriteSuccessProc(int wlength, unsigned char* buffer, Command* pcommand)
{
	EnterCriticalSection(&m_ValueCS);
	m_BaseStat.writeTimeoutFlag = 0;
	LeaveCriticalSection(&m_ValueCS);
}
void ComClient::WriteTimeoutProc(Command* pcommand)
{
	EnterCriticalSection(&m_ValueCS);
	m_BaseStat.writeTimeoutFlag++;
	if (m_BaseStat.writeTimeoutFlag >= m_Config->m_AlarmTimeoutTimes) {
		m_BaseStat.isConnected = false;
	}
	LeaveCriticalSection(&m_ValueCS);
}


void ComClient::ReadTimeoutProc(Command* pcommand)
{
	EnterCriticalSection(&m_ValueCS);
	m_BaseStat.readTimeoutFlag++;
	if (m_BaseStat.readTimeoutFlag >= m_Config->m_AlarmTimeoutTimes) {
		m_BaseStat.isConnected = false;
	}
	LeaveCriticalSection(&m_ValueCS);
}


void ComClient::ReadSuccessProc(int rlength, unsigned char* buffer, Command* pcommand)
{
	EnterCriticalSection(&m_ValueCS);
	m_BaseStat.readTimeoutFlag = 0;
	m_BaseStat.isConnected = true;
	LeaveCriticalSection(&m_ValueCS);
}

TcpClient::TcpClient(CommunicationCfg* pconfig, int freq)
	:m_Config(pconfig)
	, m_Freq(freq)
	, m_ReadTimeout(500)
	, m_WriteTimeout(250)
{

}

TcpClient::~TcpClient()
{
	Shutdown();


}

void TcpClient::Init()
{
	//m_interval = m_Config->m_interval;
	InitCommand();
}

void TcpClient::RunProc()
{
	unsigned char sendBuffer[1024];
	unsigned char recvBuffer[1024];
	while (m_RunFlag) {
		if (!m_Client.IsOpen()) {
			if (!m_Client.ConnectTo(m_Config->m_IP.c_str(), m_Config->m_Port, AF_INET, SOCK_STREAM, 500)) {
				if (!m_RunFlag)break;
				ServerDisconnectProc();
				int count = 0;
				while ((count < 10) && m_RunFlag) {
					Sleep(100);
					count++;
				}
				if (!m_RunFlag)break;
				continue;
			}
			else {
				ServerConnectedProc();
			}
		}
		CycleBegin();

		Command* pCommand = NULL;
		EnterCriticalSection(&m_RtcCS);
		if (!m_RTCommands.empty()) {
			pCommand = m_RTCommands.front();
			m_RTCommands.pop();
		}
		LeaveCriticalSection(&m_RtcCS);

		if (pCommand == NULL) {
			pCommand = GetSendCycleCommand();
		}
		if (pCommand != NULL) {
			if (pCommand->isNeedSend) {
				int bytecount = pCommand->GetRequestSequence(sendBuffer);
				int wlength = 0;
				m_Client.Read(recvBuffer, sizeof(recvBuffer), NULL, 10);

				wlength = m_Client.Write(sendBuffer, bytecount, NULL, 500);
				if (wlength > 0) {
					WriteSuccessProc(wlength, sendBuffer, pCommand);
					if (pCommand->isNeedRead) {
						int rlength = 0;
						memset(recvBuffer, 0, sizeof(recvBuffer));
						rlength = m_Client.Read(recvBuffer, sizeof(recvBuffer), NULL, 500);
						if (rlength > 0) {
							ReadSuccessProc(rlength, recvBuffer, pCommand);
							if (pCommand->Verify(recvBuffer, rlength))
							{
								pCommand->m_RespSeq = recvBuffer;
								pCommand->m_RespLen = rlength;
								if (pCommand->m_Fun)pCommand->m_Fun(pCommand->m_Ref, pCommand);
							}
							else {
								VerifyFaild(pCommand);
							}
						}
						else if (rlength == 0) {
							//timeout
							ReadTimeoutProc(pCommand);
						}
						else {
							m_Client.Close();
						}
					}
				}
				else if (wlength == 0) {
					//timeout
					WriteTimeoutProc(pCommand);
				}
				else {
					//error
					m_Client.Close();
				}
			}

			pCommand->isFinished = true;
			if (pCommand->isNeedDel) {
				delete pCommand;
				pCommand = NULL;
			}
		}
		UpdateCycleCommands();
		Sleep(m_Freq);
	}
}

void TcpClient::ServerDisconnectProc() {
	EnterCriticalSection(&m_ValueCS);
	m_BaseStat.isConnected = false;
	LeaveCriticalSection(&m_ValueCS);
}


void TcpClient::WriteSuccessProc(int wlength, unsigned char* buffer, Command* pcommand)
{
	EnterCriticalSection(&m_ValueCS);
	m_BaseStat.writeTimeoutFlag = 0;
	LeaveCriticalSection(&m_ValueCS);
}
void TcpClient::WriteTimeoutProc(Command* pcommand)
{
	EnterCriticalSection(&m_ValueCS);
	m_BaseStat.writeTimeoutFlag++;
	if (m_BaseStat.writeTimeoutFlag >= m_Config->m_AlarmTimeoutTimes) {
		m_BaseStat.isConnected = false;
	}
	LeaveCriticalSection(&m_ValueCS);
}


void TcpClient::ReadTimeoutProc(Command* pcommand)
{
	EnterCriticalSection(&m_ValueCS);
	m_BaseStat.readTimeoutFlag++;
	if (m_BaseStat.readTimeoutFlag >= m_Config->m_AlarmTimeoutTimes) {
		m_BaseStat.isConnected = false;
	}
	LeaveCriticalSection(&m_ValueCS);
}


void TcpClient::ReadSuccessProc(int rlength, unsigned char* buffer, Command* pcommand)
{
	EnterCriticalSection(&m_ValueCS);
	m_BaseStat.readTimeoutFlag = 0;
	m_BaseStat.isConnected = true;
	LeaveCriticalSection(&m_ValueCS);
}

void TcpClient::Shutdown()
{
	BaseClient::Shutdown();
	if (m_Client.IsOpen())m_Client.Close();
}