Hu Chunming / Project_VideoStructure

#include "BaseSocket.h"
#include <signal.h>
#include <unistd.h>

CBaseSocket::CBaseSocket()
{
	m_hSocket = INVALID_SOCKET;
	m_lRecvBufferPos = 0;
	m_pDataExchangePipe = new CDataExchangePipe(10);
	m_pReceiveThread = NULL;
	m_pSendThread = NULL;
	m_pReceiveThreadToDelete = NULL;
	m_pSendThreadToDelete = NULL;
	SetStatus(BASE_SOCKET_STATUS_NOT_CREATED);
}

CBaseSocket::~CBaseSocket()
{
	Close();
	if (m_pDataExchangePipe)
	{
		delete m_pDataExchangePipe;
		m_pDataExchangePipe = NULL;
	}

	//TODO:
//	if(m_pReceiveThread)
//	{
//		delete m_pReceiveThread;
//		m_pReceiveThread = NULL;
//	}
//
//	if(m_pSendThread)
//	{
//		delete m_pSendThread;
//		m_pSendThread = NULL;
//	}

	if (m_pReceiveThreadToDelete)
	{
		delete m_pReceiveThreadToDelete;
		m_pReceiveThreadToDelete = NULL;
	}

	if (m_pSendThreadToDelete)
	{
		delete m_pSendThreadToDelete;
		m_pSendThreadToDelete = NULL;
	}
}

void CBaseSocket::SetStatus(UINT nStatus)
{
	CONSOLE_LOG("socket status is set to %d.\n", nStatus);
	m_nStatus = nStatus;
}

UINT CBaseSocket::GetStatus()
{
	return m_nStatus;
}

BOOL CBaseSocket::SetNonBlocking(BOOL bNonBlocking)
{
	int opts;
	opts = fcntl(m_hSocket, F_GETFL);

	if (opts < 0)
	{
		CONSOLE_LOG("get option failed.\n");
		return FALSE;
	}

	if (bNonBlocking)
		opts = (opts | O_NONBLOCK);
	else
		opts = (opts & ~O_NONBLOCK);

	opts = fcntl(m_hSocket, F_SETFL, opts);
	if (opts < 0)
		return FALSE;

	return TRUE;
}

BOOL CBaseSocket::Create(UINT nSocketPort, int nSocketType)
{
	m_hSocket = socket(AF_INET, nSocketType, 0);
	if (m_hSocket != INVALID_SOCKET)
	{
		int iReUserAddr;
		iReUserAddr = 1;
		setsockopt(m_hSocket, SOL_SOCKET, SO_REUSEADDR, &iReUserAddr, sizeof(iReUserAddr));
		CONSOLE_LOG("the port is %d.\n",nSocketPort);
		struct sockaddr_in name;
		name.sin_family = AF_INET;
		name.sin_port = htons(nSocketPort);
		name.sin_addr.s_addr = htonl(INADDR_ANY);
		if (bind(m_hSocket, (struct sockaddr *) &name, sizeof(name)) < 0)
		{
			CONSOLE_LOG("the errno is %d.\n", errno);
			close(m_hSocket);
			m_hSocket = INVALID_SOCKET;
			return FALSE;
		}
		if (SetNonBlocking() == FALSE)
		{
			CONSOLE_LOG("Failed to set nonBlocking socket.\n");
			close(m_hSocket);
			m_hSocket = INVALID_SOCKET;
			return FALSE;
		}
		SetStatus(BASE_SOCKET_STATUS_CREATED);
		return TRUE;
	}
	else
	{
		CONSOLE_LOG("in socket the errno is %d.\n", errno);
		return FALSE;
	}
}

BOOL CBaseSocket::Connect(LPCTSTR lpszHostAddress, UINT nHostPort)
{
	int ret;
	struct sockaddr_in serverName;
	ret = inet_pton( AF_INET, lpszHostAddress, &serverName.sin_addr);
	CONSOLE_LOG("%s\n", lpszHostAddress);
//    if (errno == EAFNOSUPPORT)//wg: EAFNOSUPPORT stands for not supported family
	if (ret != 1)
	{
		CONSOLE_LOG("inet_pton failed!\n");
		return FALSE;
	}

	serverName.sin_family = AF_INET;
	serverName.sin_port = htons(nHostPort);
	ret = connect(m_hSocket, (struct sockaddr*) &serverName, sizeof(serverName));

	if (ret)
	{
		if (errno == EINPROGRESS)
		{
			SetStatus(BASE_SOCKET_STATUS_CONNECTING);
			CONSOLE_LOG("connect in progress!\n");
			return TRUE;
		}
		else
		{
			CONSOLE_LOG("error is %s.\n",strerror(errno));CONSOLE_LOG("connect failed!\n");
			return FALSE;
		}
	}
	else
	{
		CONSOLE_LOG("connect ok!");
		SetStatus(BASE_SOCKET_STATUS_CONNECTED);
		return TRUE;
	}
}

BOOL CBaseSocket::Accept(CBaseSocket* pBaseSocket)
{
	SOCKET hNewSocket;
	struct sockaddr_in clientName;
	socklen_t size;
	size = sizeof(clientName);
	hNewSocket = accept(m_hSocket, (struct sockaddr *) &clientName, &size);
	//如果套接字被标记为Non-blocking,需要通过select或者poll等函数来获取调用accept的时机
	if (hNewSocket < 0)
	{
		return FALSE;
	}CONSOLE_LOG("the new socket handle is %d.\n",hNewSocket);
	pBaseSocket->Attach(hNewSocket);
	pBaseSocket->SetStatus(BASE_SOCKET_STATUS_CONNECTED);
	return TRUE;
}

void CBaseSocket::Attach(SOCKET hSocket)
{
	m_hSocket = hSocket;
}
void CBaseSocket::OnClose(int nErrorCode)
{

}

void CBaseSocket::OnSend(int nErrorCode)
{
	//m_SendAvailable.SetEvent();
}

void CBaseSocket::OnConnect(int nErrorCode)
{

}

void CBaseSocket::OnReceive(int nErrorCode)
{

}

void CBaseSocket::OnAccept(int nErrorCode)
{

}

BOOL CBaseSocket::NothingIsReceived()
{
	return (m_lLastReceiveCount <= 0);
}

int CBaseSocket::Listen(long nConnectBackblog)
{
	int res = listen(m_hSocket, nConnectBackblog);
	CONSOLE_LOG("listen res is %d\n", res);
	if (res == 0)
		SetStatus(BASE_SOCKET_STATUS_LISTENNING);
	return res;
}

long CBaseSocket::GetPeerName(char *IPAddress, long *plPort)
{
	if (m_nStatus != BASE_SOCKET_STATUS_CONNECTED)
	{
		//no correct peer ip and port can be get until socket connected
		return RS_FAIL;
	}
	struct sockaddr_in peerAddr;
	socklen_t len = sizeof(peerAddr);
	if (getpeername(m_hSocket, (struct sockaddr*) &peerAddr, (socklen_t*) &len) == 0)
	{
		strcpy(IPAddress, inet_ntoa(peerAddr.sin_addr));
		*plPort = ntohs(peerAddr.sin_port);
		return RS_OK;
	}
	else
		return RS_FAIL;
	return RS_OK;
}

void CBaseSocket::Close()
{
	if (m_hSocket != INVALID_SOCKET)
	{
		//just stop receiving data
		shutdown(m_hSocket, 0);
		close(m_hSocket);
		m_hSocket = INVALID_SOCKET;
		SetStatus(BASE_SOCKET_STATUS_NOT_CREATED);
	}
}

long CBaseSocket::PutData(void *pData, INT32 nDataLen, DWORD nTimeOut)
{
	Lock();
	if (m_pDataExchangePipe->PutDataEx(pData, nDataLen, nTimeOut))
	{
		Unlock();
		return RS_OK;
	}
	else
	{
		Unlock();
		return RS_SOCKET_BUFFER_FULL;
	}
}

long CBaseSocket::SendData(DWORD nTimeOut)
{
	//check if the socket is connected
	if (m_nStatus != BASE_SOCKET_STATUS_CONNECTED)
	{
		return RS_FAIL;
	}

	DataExchangeBlock* pData;
	if (m_pDataExchangePipe != NULL && (pData = m_pDataExchangePipe->GetDataExchangeBlock(nTimeOut)) != NULL)
	{
		long lSentSize = 0;
		long lRes;
		BOOL bError = FALSE;

		while ((lSentSize < pData->wDataLen) && !bError)
		{
			m_SendAvailable.ResetEvent();
			lRes = send(m_hSocket, (const char*) (pData->lpData + lSentSize), pData->wDataLen - lSentSize, 0);
			if (lRes <= 0)
			{
				switch (errno)
				//modified by wg 2015-12-21 from lRes to errno
				{
				case EWOULDBLOCK:
				{
					CONSOLE_LOG("this is EWOULDBLOCK\n");
					//m_SendAvailable.Wait(100);
					timeval interval;
					interval.tv_sec = 0;
					interval.tv_usec = 10000;
					fd_set fdWriteSet;
					FD_ZERO(&fdWriteSet);
					FD_SET(m_hSocket, &fdWriteSet);
					select(FD_SETSIZE, NULL, &fdWriteSet, NULL, &interval);
				}
					break;
				default:
					if (errno == EAGAIN)
					{
						CONSOLE_LOG("this is EAGAIN\n");

						timeval interval;
						interval.tv_sec = 0;
						interval.tv_usec = 10000;
						fd_set fdWriteSet;
						FD_ZERO(&fdWriteSet);
						FD_SET(m_hSocket, &fdWriteSet);
						select(FD_SETSIZE, NULL, &fdWriteSet, NULL, &interval);
					}
					else
					{
						if (errno == SIGPIPE)
						{
							//the socket had been closed
							CONSOLE_LOG("socket disconnect!\n");
						}
						bError = TRUE;
						CONSOLE_LOG("send lRes is %ld.\n", lRes);CONSOLE_LOG("send socket data errorno is %s, %d.\n", strerror(errno), errno);CONSOLE_LOG("SIGPIPE is %s, %d.\n", strerror(SIGPIPE), SIGPIPE);
						this->SetStatus(BASE_SOCKET_STATUS_DISCONNECTED);
					}
					break;
				}
			}
			else
			{
				lSentSize += lRes;
			}
		}
		return lRes;
	}
	else
		return 0;
}

long CBaseSocket::Start(BOOL bSendSocketData)
{
	if (m_hSocket == INVALID_SOCKET)
	{
		return RS_INVALID_SOCKET;
	}
	m_pReceiveThread = new CBaseSocketReceiveThread();
	m_pReceiveThread->SetSocket(this);
	BOOL bRes = FALSE;
	bRes = m_pReceiveThread->BeginThread();
	if (!bRes)
	{
		delete m_pReceiveThread;
		m_pReceiveThread = NULL;
		return RS_FAIL;
	}

	if (bSendSocketData)
	{
		m_pSendThread = new CBaseSocketSendThread();
		m_pSendThread->SetSocket(this);
		bRes = m_pSendThread->BeginThread();
		if (!bRes)
		{
			delete m_pSendThread;
			m_pSendThread = NULL;
			return RS_FAIL;
		}
	}
	return RS_OK;
}

long CBaseSocket::Stop(BOOL bWaitForStopped)
{
	if (m_hSocket == INVALID_SOCKET)
	{
		return RS_INVALID_SOCKET;
	}
	if (m_pReceiveThread)
	{
		m_pReceiveThread->Shutdown(bWaitForStopped);
		//todo: if m_pReceiveThread can be deleted here??? why or why not???
//		m_pReceiveThreadToDelete = m_pReceiveThread;
//		m_pReceiveThread = NULL;
//		if(bWaitForStopped)
//		{
//			delete m_pReceiveThreadToDelete;
//			m_pReceiveThreadToDelete = NULL;
//		}
	}
	if (m_pSendThread)
	{
		m_pSendThread->Shutdown(TRUE);
//		m_pSendThreadToDelete = m_pSendThread;
//		m_pSendThread = NULL;
//
//		delete m_pSendThreadToDelete;
//		m_pSendThreadToDelete = NULL;
	}
	return RS_OK;
}