socketcomm.cpp

多线程通信

	// nType: (SOCK_STREAM, SOCK_DGRAM)
	SOCKET sock = socket(nFamily, nType, 0);
	if (INVALID_SOCKET != sock)
	{
		if (uOptions & SO_REUSEADDR)
		{
			// Inform Windows Sockets provider that a bind on a socket should not be disallowed
			// because the desired address is already in use by another socket
			BOOL optval = TRUE;
			if ( SOCKET_ERROR == setsockopt( sock, SOL_SOCKET, SO_REUSEADDR, (char *) &optval, sizeof( BOOL ) ) )
			{
				closesocket( sock );
				return false;
			}
		}

		if (nType == SOCK_DGRAM)
		{
			if (uOptions & SO_BROADCAST)
			{
				// Inform Windows Sockets provider that broadcast messages are allowed
				BOOL optval = TRUE;
				if ( SOCKET_ERROR == setsockopt( sock, SOL_SOCKET, SO_BROADCAST, (char *) &optval, sizeof( BOOL ) ) )
				{
					closesocket( sock );
					return false;
				}

				// we may proceed
				m_bBroadcast = true;
			}

			// we need mutex only for UDP - broadcast socket
			m_hMutex = CreateMutex(NULL, FALSE, NULL);
			if (NULL == m_hMutex)
			{
				closesocket( sock );
				return false;
			}
		}

		// Associate a local address with the socket
		SockAddrIn sockAddr;
		sockAddr.CreateFrom(NULL, strServiceName, nFamily);

		if ( SOCKET_ERROR == bind(sock, (LPSOCKADDR)sockAddr, sockAddr.Size()))
		{
			closesocket( sock );
			m_bBroadcast = false;
			if (NULL != m_hMutex)
				CloseHandle( m_hMutex );
			m_hMutex = NULL;
			return false;
		}

		// Listen to the socket, only valid for connection socket
		if (SOCK_STREAM == nType)
		{
			if ( SOCKET_ERROR == listen(sock, SOMAXCONN))
			{
				closesocket( sock );
				return false;
			}
		}

		// Success, now we may save this socket
		m_hComm = (HANDLE) sock;
	}

	return (INVALID_SOCKET != sock);
}


///////////////////////////////////////////////////////////////////////////////
// ConnectTo
///////////////////////////////////////////////////////////////////////////////
// DESCRIPTION:
//				Establish connection with a server service or port
// PARAMETERS:
//	LPCTSTR strDestination: hostname or address to connect (in .dot format)
//	LPCTSTR strServiceName: Service name or port number
//	int nFamily: address family to use (set to AF_INET)
//	int nType: type of socket to create (SOCK_STREAM, SOCK_DGRAM)
///////////////////////////////////////////////////////////////////////////////
bool CSocketComm::ConnectTo(LPCTSTR strDestination, LPCTSTR strServiceName, int nFamily, int nType)
{
	// Socket is already opened
	if ( IsOpen() )
		return false;

	// Create a Socket that is bound to a specific service provide
	// nFamily: (AF_INET)
	// nType: (SOCK_STREAM, SOCK_DGRAM)
	SOCKET sock = socket(nFamily, nType, 0);
	if (INVALID_SOCKET != sock)
	{
		// Associate a local address with the socket
		SockAddrIn sockAddr;
		if (false == sockAddr.CreateFrom(NULL, TEXT("0"), nFamily))
		{
			closesocket( sock );
			return false;
		}

		if ( SOCKET_ERROR == bind(sock, (LPSOCKADDR)sockAddr, sockAddr.Size() ))
		{
			closesocket( sock );
			return false;
		}

		// Now get destination address & port
		sockAddr.CreateFrom( strDestination, strServiceName );

		// try to connect - if fail, server not ready
		if (SOCKET_ERROR == connect( sock, (LPSOCKADDR)sockAddr, sockAddr.Size()))
		{
			closesocket( sock );
			return false;
		}

		// Success, now we may save this socket
		m_hComm = (HANDLE) sock;
	}
	return (INVALID_SOCKET != sock);
}


///////////////////////////////////////////////////////////////////////////////
// CloseComm
///////////////////////////////////////////////////////////////////////////////
// DESCRIPTION:
//		Close Socket Communication
// PARAMETERS:
//		None
///////////////////////////////////////////////////////////////////////////////
void CSocketComm::CloseComm()
{
	if (IsOpen())
	{
		ShutdownConnection((SOCKET)m_hComm);
		m_hComm = INVALID_HANDLE_VALUE;
		m_bBroadcast = false;
	}
}


///////////////////////////////////////////////////////////////////////////////
// WatchComm
/////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
// DESCRIPTION:
//		Starts Socket Communication Working thread
// PARAMETERS:                                                              起线程
//		None
//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
bool CSocketComm::WatchComm()
{
	if (!IsStart())
	{
		if (IsOpen())
		{
			HANDLE hThread;
			UINT uiThreadId = 0;
			hThread = (HANDLE)_beginthreadex(NULL,	// Security attributes
									  0,	// stack
						SocketThreadProc,	// Thread proc
									this,	// Thread param
						CREATE_SUSPENDED,	// creation mode
							&uiThreadId);	// Thread ID

			if ( NULL != hThread)
			{
				//SetThreadPriority(hThread, THREAD_PRIORITY_HIGHEST);
				ResumeThread( hThread );
				m_hThread = hThread;
				return true;
			}
		}
	}
	return false;
}


///////////////////////////////////////////////////////////////////////////////
// StopComm
/////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
// DESCRIPTION:
//		Close Socket and Stop Communication thread
// PARAMETERS:                                                               关线程
//		None
//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
void CSocketComm::StopComm()
{
	// Close Socket
	if (IsOpen())
	{
		CloseComm();
		Sleep(50);
	}

	// Kill Thread
	if (IsStart())
	{
		if (WaitForSingleObject(m_hThread, 5000L) == WAIT_TIMEOUT)
			TerminateThread(m_hThread, 1L);
		CloseHandle(m_hThread);
		m_hThread = NULL;
	}

	// Clear Address list
	if (!m_AddrList.empty())
		m_AddrList.clear();

	// Destroy Synchronization objects
	if (NULL != m_hMutex)
	{
		CloseHandle( m_hMutex );
		m_hMutex = NULL;
	}

}


///////////////////////////////////////////////////////////////////////////////
// ReadComm
///////////////////////////////////////////////////////////////////////////////
// DESCRIPTION:
//		Reads the Socket Communication
// PARAMETERS:
//		LPBYTE lpBuffer: buffer to place new data
//		DWORD dwSize: maximum size of buffer
//		DWORD dwTimeout: timeout to use in millisecond
///////////////////////////////////////////////////////////////////////////////
DWORD CSocketComm::ReadComm(LPBYTE lpBuffer, DWORD dwSize, DWORD dwTimeout)
{
	_ASSERTE( IsOpen() );
	_ASSERTE( lpBuffer != NULL );

	if (lpBuffer == NULL || dwSize < 1L)
		return 0L;

	fd_set	fdRead  = { 0 };
	TIMEVAL	stTime;
	TIMEVAL	*pstTime = NULL;

	if ( INFINITE != dwTimeout ) {
		stTime.tv_sec = dwTimeout/1000;
		stTime.tv_usec = dwTimeout % 1000;
		pstTime = &stTime;
	}

	SOCKET s = (SOCKET) m_hComm;
	// Set Descriptor
	if ( !FD_ISSET( s, &fdRead ) )
		FD_SET( s, &fdRead );

	// Select function set read timeout
	DWORD dwBytesRead = 0L;
	int res = select( s+1, &fdRead, NULL, NULL, pstTime );
	if ( res > 0)
	{
		if (IsBroadcast() || IsSmartAddressing())
		{
			SockAddrIn sockAddr;
			int nLen = sockAddr.Size();
			int nOffset = IsSmartAddressing() ? nLen : 0; // use offset for Smart addressing
			if ( dwSize < (DWORD) nOffset)	// error - buffer to small
			{
				SetLastError( ERROR_INVALID_USER_BUFFER );
				return -1L;
			}
			LPSTR lpszData = (LPSTR)(lpBuffer + nOffset);
			res = recvfrom( s, lpszData, dwSize-nOffset, 0, (LPSOCKADDR)sockAddr, &nLen);

			// clear 'sin_zero', we will ignore them with 'SockAddrIn' anyway!
			memset(&sockAddr.sin_zero, 0, sizeof(sockAddr.sin_zero));

			// Lock the list...
			LockList();
			m_AddrList.remove( sockAddr );
			
			if ( res >= 0)
			{
				// insert unique address
				m_AddrList.insert(m_AddrList.end(), sockAddr);

				if (IsSmartAddressing())
				{
					memcpy(lpBuffer, &sockAddr, sockAddr.Size());
					res += sockAddr.Size();
				}
			}

			UnlockList(); // unlock this object addresses-list
		}
		else
		{
			res = recv( s, (LPSTR)lpBuffer, dwSize, 0);
		}

		dwBytesRead = (DWORD)((res > 0)?(res) : (-1L));
	}

	return dwBytesRead;
}


///////////////////////////////////////////////////////////////////////////////
// WriteComm
///////////////////////////////////////////////////////////////////////////////
// DESCRIPTION:
//		Writes data to the Socket Communication
// PARAMETERS:
//		const LPBYTE lpBuffer: data to write
//		DWORD dwCount: maximum characters to write
//		DWORD dwTimeout: timeout to use in millisecond
///////////////////////////////////////////////////////////////////////////////
DWORD CSocketComm::WriteComm(const LPBYTE lpBuffer, DWORD dwCount, DWORD dwTimeout)
{
	_ASSERTE( IsOpen() );
	_ASSERTE( NULL != lpBuffer );

	// Accept 0 bytes message
	if (!IsOpen() || NULL == lpBuffer)
		return 0L;

	fd_set	fdWrite  = { 0 };
	TIMEVAL	stTime;
	TIMEVAL	*pstTime = NULL;

	if ( INFINITE != dwTimeout ) {
		stTime.tv_sec = dwTimeout/1000;
		stTime.tv_usec = dwTimeout % 1000;
		pstTime = &stTime;
	}

	SOCKET s = (SOCKET) m_hComm;
	// Set Descriptor
	if ( !FD_ISSET( s, &fdWrite ) )
		FD_SET( s, &fdWrite );

	// Select function set write timeout
	DWORD dwBytesWritten = 0L;
	int res = select( s+1, NULL, &fdWrite, NULL, pstTime );
	if ( res > 0)
	{
		// Send message to peer or broadcast it
		if (IsBroadcast() || IsSmartAddressing())
		{
			// use offset for Smart addressing
			int nOffset = IsSmartAddressing() ? sizeof(SOCKADDR_IN) : 0;
			if (IsSmartAddressing())
			{
				if ( dwCount < sizeof(SOCKADDR_IN))	// error - buffer to small
				{
					SetLastError( ERROR_INVALID_USER_BUFFER );
					return -1L;
				}

				// read socket address from buffer
				SockAddrIn sockAddr;
				sockAddr.SetAddr((PSOCKADDR_IN) lpBuffer);

				// Get Address and send data
				if (sockAddr.sin_addr.s_addr != htonl(INADDR_BROADCAST))
				{
					LPSTR lpszData = (LPSTR)(lpBuffer + nOffset);
					res = sendto( s, lpszData, dwCount-nOffset, 0,
						(LPSOCKADDR)sockAddr, sockAddr.Size());
					dwBytesWritten = (DWORD)((res >= 0)?(res) : (-1));
					return dwBytesWritten;
				}
			}

			// Broadcast send to all connected-peer
			LockList(); // Lock this object addresses-list

			CSockAddrList::iterator iter = m_AddrList.begin();
			for( ; iter != m_AddrList.end(); )
			{
				// Fix v1.3 - nOffset was missing
				res = sendto( s, (LPCSTR)&lpBuffer[nOffset], dwCount-nOffset, 0, (LPSOCKADDR)(*iter), iter->Size());
				if (res < 0)
				{
					CSockAddrList::iterator deladdr = iter;
					++iter;	// get next
					m_AddrList.erase( deladdr );
				}
				else
					++iter;	// get next
			}

			UnlockList(); // unlock this object addresses-list

			// always return success - UDP
			res = (int) dwCount - nOffset;
		}
		else // Send to peer-connection
			res = send( s, (LPCSTR)lpBuffer, dwCount, 0);

		dwBytesWritten = (DWORD)((res >= 0)?(res) : (-1));
	}

	return dwBytesWritten;
}


///////////////////////////////////////////////////////////////////////////////
// Run
///////////////////////////////////////////////////////////////////////////////
// DESCRIPTION:
//		This function runs the main thread loop
//		this implementation can be overloaded.
//      This function calls CSocketComm::OnDataReceived() (Virtual Function)
// PARAMETERS:
// NOTES:
//		You should not wait on the thread to end in this function or overloads
///////////////////////////////////////////////////////////////////////////////
void CSocketComm::Run()
{
	stMessageProxy stMsgProxy;
	DWORD	dwBytes  = 0L;
	DWORD	dwTimeout = DEFAULT_TIMEOUT;
	LPBYTE  lpData  = (LPBYTE)&stMsgProxy;
	DWORD	dwSize  = sizeof(stMsgProxy);

	if (!IsSmartAddressing())
	{
		lpData = stMsgProxy.data;
		dwSize = sizeof(stMsgProxy.data);
	}

	// Should we run as server mode
	if (IsServer())
	{
		if (!IsBroadcast())
		{
			SOCKET sock = (SOCKET) m_hComm;
			sock = WaitForConnection( sock );

			// Get new connection socket
			if (sock != INVALID_SOCKET)
			{
				ShutdownConnection( (SOCKET) m_hComm);
				m_hComm = (HANDLE) sock;
				OnEvent( EVT_CONSUCCESS ); // connect
			}
			else
			{
				// Do not send event if we are closing
				if (IsOpen())
					OnEvent( EVT_CONFAILURE ); // wait fail
				return;
			}
		}
	}
	else
		GetPeerName( stMsgProxy.address );

	while( IsOpen() )
	{
		// Blocking mode: Wait for event
		dwBytes = ReadComm(lpData, dwSize, dwTimeout);

		// Error? - need to signal error
		if (dwBytes == (DWORD)-1L)
		{
			// Do not send event if we are closing
			if (IsOpen())
				OnEvent( EVT_CONDROP ); // lost connection

			// special case for UDP, alert about the event but do not stop
			if (IsBroadcast())
				continue;
			else
				break;
		}

		// Chars received?
		if (IsSmartAddressing() && dwBytes == sizeof(SOCKADDR_IN))
			OnEvent( EVT_ZEROLENGTH );
		else if (dwBytes > 0L)
		{
			OnDataReceived( lpData, dwBytes);
		}

		Sleep(0);
	}
}


///////////////////////////////////////////////////////////////////////////////
// SocketThreadProc
///////////////////////////////////////////////////////////////////////////////
// DESCRIPTION:
//     Socket Thread function.  This function is the main thread for socket
//     communication - Asynchronous mode.
// PARAMETERS:
//     LPVOID pParam : Thread parameter - a CSocketComm pointer
// NOTES:
///////////////////////////////////////////////////////////////////////////////
UINT WINAPI CSocketComm::SocketThreadProc(LPVOID pParam)
{
	CSocketComm* pThis = reinterpret_cast<CSocketComm*>( pParam );
	_ASSERTE( pThis != NULL );

	pThis->Run();

	return 1L;
} // end SocketThreadProc