+Unity Runtime codeHEADmaster

1 files changed, 666 insertions, 0 deletions

diff --git a/Runtime/Network/PlayerCommunicator/GeneralConnection.cpp b/Runtime/Network/PlayerCommunicator/GeneralConnection.cpp
new file mode 100644
index 0000000..dc4c89d
--- /dev/null
+++ b/Runtime/Network/PlayerCommunicator/GeneralConnection.cpp
@@ -0,0 +1,666 @@
+#include "UnityPrefix.h"
+
+#if ENABLE_PLAYERCONNECTION
+
+#include "GeneralConnection.h"
+#include "GeneralConnectionInternals.h"
+#include "Runtime/Math/Random/rand.h"
+#include "Runtime/Network/NetworkUtility.h"
+#include "Runtime/Profiler/TimeHelper.h"
+
+bool GeneralConnection::ms_RunningUnitTests = false;
+int GeneralConnection::ms_Version = 0x00100100;
+
+GeneralConnection::GeneralConnection ()
+:	m_LogEnabled(true)
+{
+	char ips[10][16];
+	int count = GetIPs(ips);
+	if (count != 0)
+		m_LocalIP = ips[0];
+	else
+		m_LocalIP = "0.0.0.0";
+
+#if UNITY_BB10
+	for(int x = 0; x < sizeof(ips); x++)
+	{
+		if(std::string(ips[x]).find("169.254") == 0)
+		{
+			m_LocalIP = ips[x];
+			break;
+		}
+	}
+#endif
+
+	m_LocalGuid = Rand((int)GetProfileTime(START_TIME)).Get();
+	// We reserve ANY_PLAYERCONNECTION guid for special use
+	if (m_LocalGuid == ANY_PLAYERCONNECTION) m_LocalGuid = ANY_PLAYERCONNECTION ^ 1;
+}
+
+UInt32 GeneralConnection::GetLocalGuid() const
+{
+	return m_LocalGuid;
+}
+
+void GeneralConnection::WaitForFinish()
+{
+	int msgWait = 60;
+
+	while (HasBytesToSend())
+	{
+		if (msgWait++ == 60)
+		{
+			printf_console("Waiting for finish\n");
+			msgWait = 0;
+		}
+
+		Poll();
+		Thread::Sleep(0.05);
+	}
+}
+
+bool GeneralConnection::HasBytesToSend() const
+{
+	ConnectionMap::const_iterator it = m_Connections.begin();
+	for ( ; it != m_Connections.end(); ++it)
+		if (it->second->HasBytesToSend())
+			return true;
+
+	return false;
+}
+
+GeneralConnection::~GeneralConnection()
+{
+	DisconnectAll();
+}
+
+
+void GeneralConnection::DisconnectAll()
+{
+	ConnectionMap::iterator it = m_Connections.begin();
+	std::vector< int > disconnectGuids;
+	for( ; it != m_Connections.end(); ++it)
+		disconnectGuids.push_back(it->first);
+
+	for (int i = 0; i < disconnectGuids.size(); i++)
+		Disconnect(disconnectGuids[i]);
+
+	Assert(m_Connections.empty());
+}
+
+
+void GeneralConnection::Initialize ()
+{
+	NetworkInitialize();
+}
+
+void GeneralConnection::Cleanup ()
+{
+	NetworkCleanup();
+}
+
+
+GeneralConnection::Connection* GeneralConnection::GetConnection(UInt32 guid)
+{
+	ConnectionMap::iterator it = m_Connections.find(guid);
+	if (it == m_Connections.end())
+		return NULL;
+	return it->second;
+}
+
+void GeneralConnection::RegisterConnection(UInt32 guid, TSocketHandle socketHandle)
+{
+	if (GetConnection(guid))
+		Disconnect(guid);
+
+	LOG_PLAYER_CONNECTION(Format("PlayerConnection registered: %d", guid));
+	m_Connections[guid] = new Connection(socketHandle);
+	for(int i = 0; i < m_ConnectionHandlers.size(); i++)
+		(m_ConnectionHandlers[i])(guid);
+}
+
+void GeneralConnection::Disconnect(UInt32 guid)
+{
+	ConnectionMap::iterator it = m_Connections.find(guid);
+	if (it == m_Connections.end())
+		return;
+
+	LOG_PLAYER_CONNECTION(Format("PlayerConnection disconnecting: %d", guid));
+	for (int i = 0; i < m_DisconnectionHandlers.size(); i++)
+		(m_DisconnectionHandlers[i])(guid);
+
+#if ENABLE_PLAYER_CONNECTION_DEBUG_LOG
+	Connection& connection = *it->second;
+	LOG_PLAYER_CONNECTION(Format("GeneralConnection::Connection send buffer mem usage %d\n", connection.GetSendBufferSize()));
+	LOG_PLAYER_CONNECTION(Format("GeneralConnection::Connection recv buffer mem usage %d\n", connection.GetRecvBufferSize()));
+#endif
+
+	delete it->second;
+
+	m_Connections.erase(it);
+}
+
+
+void GeneralConnection::SendMessage(UInt32 guid, MessageID id, const void* data, UInt32 size)
+{
+	NetworkMessage message;
+	message.InitializeMessage();
+	message.SetID(id);
+	message.SetDataSize(size);
+
+	bool oldlog = IsLogEnabled();
+	SetLogEnabled(false);
+
+	if (guid == ANY_PLAYERCONNECTION)
+	{
+		LOG_PLAYER_CONNECTION(Format("PlayerConnection send message to ALL, id '%d', size '%d'", id, size));
+		ConnectionMap::iterator it = m_Connections.begin();
+		for ( ; it != m_Connections.end(); ++it)
+			it->second->SendMessage(message, data);
+	}
+	else
+	{
+		LOG_PLAYER_CONNECTION(Format("PlayerConnection send message to '%d', id '%d', size '%d'", guid, id, size));
+		ConnectionMap::iterator it = m_Connections.find(guid);
+		if (it != m_Connections.end())
+			it->second->SendMessage(message, data);
+	}
+	SetLogEnabled(oldlog);
+}
+
+void GeneralConnection::Poll()
+{
+	ABSOLUTE_TIME start = START_TIME;
+	ConnectionMap::iterator it = m_Connections.begin();
+	UNITY_TEMP_VECTOR(int) disconnectGuids;
+	for ( ; it != m_Connections.end(); ++it)
+	{
+		Connection& connection = *(it->second);
+		connection.Poll();
+		const void*		messageDataPtr;
+		NetworkMessage	messageHeader;
+		while ( (GetProfileTime(ELAPSED_TIME(start)) < 20*kTimeMillisecond) && ((messageDataPtr = connection.ReceiveMessage(&messageHeader)) != NULL) )
+		{
+			LOG_PLAYER_CONNECTION(Format("PlayerConnection recv message id '%d', size '%d'", messageHeader.GetID(), messageHeader.GetDataSize()));
+			std::map< MessageID, MessageHandlerFunc >::iterator handler = m_HandlerMap.find(messageHeader.GetID());
+			if (handler != m_HandlerMap.end())
+				(handler->second)(messageDataPtr, messageHeader.GetDataSize(), it->first);
+			connection.ReleaseReceivedMessage();
+		}
+		if (!connection.IsValid())
+			disconnectGuids.push_back(it->first);
+	}
+
+	for (int i = 0; i < disconnectGuids.size(); i++)
+		Disconnect(disconnectGuids[i]);
+}
+
+// Handlers
+void GeneralConnection::RegisterMessageHandler( MessageID messageID, MessageHandlerFunc func )
+{
+	if(m_HandlerMap.find(messageID) != m_HandlerMap.end())
+		ErrorString("MessageHandler already registered");
+	m_HandlerMap[messageID] = func;
+}
+
+void GeneralConnection::RegisterConnectionHandler( ConnectionHandlerFunc func )
+{
+	for (int i = 0; i < m_ConnectionHandlers.size(); i++)
+		Assert(m_ConnectionHandlers[i] != func);
+
+	m_ConnectionHandlers.push_back(func);
+
+	// call the connect handler on already connected sockets
+	ConnectionMap::iterator it = m_Connections.begin();
+	for( ; it != m_Connections.end(); ++it)
+		(func)(it->first);
+}
+
+void GeneralConnection::RegisterDisconnectionHandler( ConnectionHandlerFunc func )
+{
+	for (int i = 0; i < m_DisconnectionHandlers.size(); i++)
+		Assert(m_DisconnectionHandlers[i] != func);
+
+	m_DisconnectionHandlers.push_back(func);
+}
+
+void GeneralConnection::UnregisterMessageHandler( MessageID messageID, MessageHandlerFunc func )
+{
+	std::map< MessageID, MessageHandlerFunc >::iterator found = m_HandlerMap.find(messageID);
+	if(found == m_HandlerMap.end())
+		ErrorString("MessageHandler not registered");
+	Assert(found->second == func);
+	m_HandlerMap.erase(found);
+}
+
+void GeneralConnection::UnregisterConnectionHandler( ConnectionHandlerFunc func )
+{
+	std::vector< ConnectionHandlerFunc >::iterator handlerIt = m_ConnectionHandlers.begin();
+	for ( ; handlerIt != m_ConnectionHandlers.end(); ++handlerIt)
+	{
+		if (*handlerIt == func)
+		{
+			m_ConnectionHandlers.erase(handlerIt);
+			return;
+		}
+	}
+}
+
+void GeneralConnection::UnregisterDisconnectionHandler( ConnectionHandlerFunc func )
+{
+	// call the disconnect handler on already connected sockets
+	ConnectionMap::iterator it = m_Connections.begin();
+	for( ; it != m_Connections.end(); ++it)
+		(func)(it->first);
+
+	std::vector< ConnectionHandlerFunc >::iterator handlerIt = m_DisconnectionHandlers.begin();
+	for ( ; handlerIt != m_DisconnectionHandlers.end(); ++handlerIt)
+	{
+		if (*handlerIt == func)
+		{
+			m_DisconnectionHandlers.erase(handlerIt);
+			return;
+		}
+	}
+}
+
+// ------------------------------------------------------------------------------
+// Connection
+// ------------------------------------------------------------------------------
+GeneralConnection::Connection::Connection(TSocketHandle socketHandle)
+: m_SocketStream(socketHandle, kDataBufferMaxSize, kDataBufferMaxSize)
+, m_PendingMessageData(NULL)
+, m_HasPendingMessage(false)
+{
+}
+
+GeneralConnection::Connection::~Connection()
+{
+	if (m_PendingMessageData)
+		ReleaseReceivedMessage();
+}
+
+const void* GeneralConnection::Connection::ReceiveMessage(NetworkMessage* message)
+{
+	AssertBreak(!m_PendingMessageData);
+	Mutex::AutoLock lock (m_RecvMutex);
+
+	ExtendedGrowingRingbuffer& recvBuffer = m_SocketStream.RecvBuffer();
+	UInt32 bytesAvailable = recvBuffer.GetAvailableSize();
+	if (!m_HasPendingMessage && bytesAvailable >= sizeof(NetworkMessage))
+	{
+		m_SocketStream.RecvAll(&m_PendingMessage, sizeof(NetworkMessage));
+		m_HasPendingMessage = true;
+		bytesAvailable = recvBuffer.GetAvailableSize();
+		Assert(recvBuffer.GetSize() >= m_PendingMessage.GetDataSize() && "Buffer is not large enough for the message.");
+
+	}
+
+	if (m_HasPendingMessage)
+	{
+		UInt32 messageSize = m_PendingMessage.GetDataSize();
+		if (m_HasPendingMessage && bytesAvailable >= messageSize)
+		{
+			m_HasPendingMessage = false;
+			UInt32 linearData = messageSize;
+			m_PendingMessageData = (void*) recvBuffer.ReadPtr(&linearData);
+			m_LocallyAllocatedMessageData = linearData < messageSize;
+			if (m_LocallyAllocatedMessageData)
+			{
+				m_PendingMessageData = UNITY_MALLOC(kMemNetwork, messageSize);
+				m_SocketStream.RecvAll(m_PendingMessageData, messageSize);
+			}
+			memcpy(message, &m_PendingMessage, sizeof(NetworkMessage));
+			return m_PendingMessageData;
+		}
+	}
+	return NULL;
+}
+
+void GeneralConnection::Connection::ReleaseReceivedMessage()
+{
+	AssertBreak(m_PendingMessageData);
+	Mutex::AutoLock lock (m_RecvMutex);
+
+	if (m_LocallyAllocatedMessageData)
+		UNITY_FREE(kMemNetwork, m_PendingMessageData);
+	else
+		m_SocketStream.RecvBuffer().ReadPtrUpdate(m_PendingMessageData, m_PendingMessage.GetDataSize());
+	m_PendingMessageData = NULL;
+}
+
+void GeneralConnection::Connection::SendMessage(NetworkMessage& message, const void* data)
+{
+	Mutex::AutoLock lock (m_SendMutex);
+	if(message.AllowSkip() && !m_SocketStream.CanSendNonblocking(sizeof(NetworkMessage) + message.GetDataSize()))
+	{
+		WarningString("Skipping profile frame. Reciever can not keep up with the amount of data sent");
+		return;
+	}
+	m_SocketStream.SendAll(&message, sizeof(NetworkMessage));
+	m_SocketStream.SendAll(data, message.GetDataSize());
+}
+
+// ------------------------------------------------------------------------------
+// NetworkMessage
+// ------------------------------------------------------------------------------
+void GeneralConnection::NetworkMessage::InitializeMessage()
+{
+	UInt16 magicNumber = PLAYER_MESSAGE_MAGIC_NUMBER;
+	SwapEndianBytesLittleToNative(magicNumber);
+	m_MagicNumber = magicNumber;
+}
+
+bool GeneralConnection::NetworkMessage::CheckMessageValidity()
+{
+	UInt16 magicNumber = m_MagicNumber;
+	SwapEndianBytesLittleToNative(magicNumber);
+	return magicNumber == PLAYER_MESSAGE_MAGIC_NUMBER;
+}
+
+void GeneralConnection::NetworkMessage::SetID( GeneralConnection::MessageID messageID )
+{
+	UInt16 id = messageID;
+	SwapEndianBytesLittleToNative(id);
+	m_ID = id;
+}
+
+GeneralConnection::MessageID GeneralConnection::NetworkMessage::GetID()
+{
+	UInt16 id = m_ID;
+	SwapEndianBytesLittleToNative(id);
+	return (MessageID) id;
+}
+
+void GeneralConnection::NetworkMessage::SetDataSize( UInt32 size )
+{
+	SwapEndianBytesLittleToNative(size);
+	m_Size = size;
+}
+
+UInt32 GeneralConnection::NetworkMessage::GetDataSize()
+{
+	UInt32 size = m_Size;
+	SwapEndianBytesLittleToNative(size);
+	return size;
+}
+
+// ------------------------------------------------------------------------------
+// MulticastInfo
+// ------------------------------------------------------------------------------
+#if ENABLE_MULTICAST
+GeneralConnection::MulticastInfo::MulticastInfo()
+{
+	Clear();
+}
+
+void GeneralConnection::MulticastInfo::Clear()
+{
+	m_Valid = false;
+	m_Flags = 0;
+	m_Port = 0;
+	m_Guid = 0;
+	m_EditorGuid = 0;
+}
+bool GeneralConnection::MulticastInfo::Parse( const char* buffer, void* in)
+{
+	char ip[kMulticastBufferSize];
+	char id[kMulticastBufferSize];
+	int version;
+	int debug;
+	if(sscanf(buffer, SERVER_IDENTIFICATION_FORMAT, ip, &m_Port, &m_Flags, &m_Guid, &m_EditorGuid, &version, id, &debug) != 8)
+	{
+		Clear();
+		m_Valid = false;
+		m_IsLocalhost = false;
+		return false;
+		//ErrorString(Format("MulticastInfo should be in this format: '%s' but got: '%s'", SERVER_IDENTIFICATION_FORMAT, buffer));
+	}
+#if !UNITY_WINRT
+	m_Ip = in ? InAddrToIP((sockaddr_in*)in) : std::string(ip);
+#endif
+
+	m_Identifier = std::string(id);
+	m_Valid = (version == ms_Version);
+	SetLocalhostFlag();
+	return true;
+}
+
+void GeneralConnection::MulticastInfo::SetLocalhostFlag()
+{
+	std::string localIP;
+	char ips[10][16];
+	int count = GetIPs(ips);
+	if (count != 0)
+		localIP = ips[0];
+	else
+		localIP = "0.0.0.0";
+	m_IsLocalhost = (m_Ip.compare("0.0.0.0") == 0) || (m_Ip.compare("127.0.0.1") == 0) || (m_Ip.compare(localIP) == 0);
+}
+
+#endif //ENABLE_MULTICAST
+
+
+/// These tests are unstable and run for too long?
+// ---------------------------------------------------------------------------
+#if ENABLE_UNIT_TESTS && 0
+
+#include "External/UnitTest++/src/UnitTest++.h"
+#include "EditorConnection.h"
+#include "PlayerConnection.h"
+
+SUITE (GeneralConnectionTests)
+{
+	bool isClientConnected = false;
+	UInt32 clientConnectionGuid = -1;
+	void ClientConnectionHandler (UInt32 guid) { isClientConnected = true; clientConnectionGuid = guid; }
+	void ClientDisconnectionHandler (UInt32 guid) { isClientConnected = false; clientConnectionGuid = -1;}
+
+	bool isServerConnected = false;
+	UInt32 serverConnectionGuid = -1;
+	void ServerConnectionHandler (UInt32 guid) { isServerConnected = true; serverConnectionGuid = guid;}
+	void ServerDisconnectionHandler (UInt32 guid) { isServerConnected = false; serverConnectionGuid = -1;}
+
+	struct GeneralConnectionFixture
+	{
+		GeneralConnectionFixture ()
+		{
+			Initialize();
+		}
+		~GeneralConnectionFixture ()
+		{
+			Destroy();
+		}
+
+		void Initialize ()
+		{
+			GeneralConnection::RunningUnitTest();
+			Rand r (GetProfileTime(START_TIME));
+			int multicastport = PLAYER_UNITTEST_MULTICAST_PORT + (r.Get() & PLAYER_PORT_MASK);
+			player = new PlayerConnection("", multicastport);
+			editor = new EditorConnection(multicastport);
+			editor->RegisterConnectionHandler( ClientConnectionHandler );
+			player->RegisterConnectionHandler( ServerConnectionHandler );
+			editor->RegisterDisconnectionHandler( ClientDisconnectionHandler );
+			player->RegisterDisconnectionHandler( ServerDisconnectionHandler );
+		}
+
+		void Destroy ()
+		{
+			delete player;
+			delete editor;
+			player = NULL;
+			editor = NULL;
+		}
+
+		void Connect ()
+		{
+			//record time
+			ABSOLUTE_TIME start = START_TIME;
+			while (GetProfileTime(ELAPSED_TIME(start)) < 1*kTimeSecond)
+			{
+				player->Poll();
+				editor->PollWithCustomMessage();
+				if (player->IsConnected() && editor->IsConnected()) break;
+			}
+		}
+
+		void CheckConnectionState(bool state)
+		{
+			// is internal connection state
+			CHECK_EQUAL (state, player->IsConnected() );
+			CHECK_EQUAL (state, editor->IsConnected() );
+			// is custom connect handler called
+			CHECK_EQUAL (state, isClientConnected);
+			CHECK_EQUAL (state, isServerConnected);
+		}
+
+		PlayerConnection* player;
+		EditorConnection* editor;
+	};
+
+	std::string message;
+	void HandleGeneralConnectionMessage (const void* data, UInt32 size, UInt32 guid)
+	{
+		message = std::string((char*)data);
+	}
+
+	int* largeMessage;
+	void HandleLargeMessage(const void* data, UInt32 size, UInt32 guid)
+	{
+		largeMessage = new int[size/sizeof(int)];
+		memcpy(largeMessage, data, size);
+	}
+
+	int messageCount = 0;
+	void HandleManyMessages(const void* data, UInt32 size, UInt32 guid)
+	{
+		CHECK_EQUAL(((char*)data)[10],messageCount);
+		messageCount++;
+	}
+
+	TEST_FIXTURE(GeneralConnectionFixture, CanDisconnectServer)
+	{
+		Connect();
+		CheckConnectionState(true);
+
+		player->DisconnectAll();
+
+		ABSOLUTE_TIME start = START_TIME;
+		while (GetProfileTime(ELAPSED_TIME(start)) < 1*kTimeSecond)
+		{
+			editor->PollWithCustomMessage();
+			if (!editor->IsConnected()) break;
+		}
+		CheckConnectionState(false);
+	}
+
+	TEST_FIXTURE(GeneralConnectionFixture, CanConnect)
+	{
+		Connect();
+		CheckConnectionState(true);
+	}
+
+	TEST_FIXTURE(GeneralConnectionFixture, CanDisconnectClient)
+	{
+		Connect();
+		CheckConnectionState(true);
+
+		editor->DisconnectAll();
+
+		ABSOLUTE_TIME start = START_TIME;
+		while (GetProfileTime(ELAPSED_TIME(start)) < 1*kTimeSecond)
+		{
+			player->Poll();
+			if (!player->IsConnected()) break;
+		}
+		CheckConnectionState(false);
+	}
+
+	TEST_FIXTURE(GeneralConnectionFixture, CanSendMessage)
+	{
+		Connect();
+		editor->RegisterMessageHandler(GeneralConnection::kLastMessageID,HandleGeneralConnectionMessage);
+		player->RegisterMessageHandler(GeneralConnection::kLastMessageID,HandleGeneralConnectionMessage);
+
+		player->SendMessage(serverConnectionGuid, GeneralConnection::kLastMessageID, "Hello World", 12);
+		ABSOLUTE_TIME start = START_TIME;
+		while (GetProfileTime(ELAPSED_TIME(start)) < 1*kTimeSecond)
+		{
+			player->Poll();
+			editor->PollWithCustomMessage();
+			if (!message.empty()) break;
+		}
+		CHECK_EQUAL (std::string("Hello World"), message);
+
+		message = std::string("");
+		editor->SendMessage(clientConnectionGuid, GeneralConnection::kLastMessageID, "Are you there", 14);
+		start = START_TIME;
+		while (GetProfileTime(ELAPSED_TIME(start)) < 1*kTimeSecond)
+		{
+			player->Poll();
+			editor->PollWithCustomMessage();
+			if (!message.empty()) break;
+		}
+		CHECK_EQUAL (std::string("Are you there"), message);
+
+	}
+
+	TEST_FIXTURE(GeneralConnectionFixture, SendLargeMessage)
+	{
+		Connect();
+		editor->RegisterMessageHandler(GeneralConnection::kLastMessageID,HandleLargeMessage);
+
+		int msgSize = 400*1024;
+		int* buffer = new int[msgSize];
+		for(int i = 0; i < msgSize; i++)
+			buffer[i] = i;
+		largeMessage = NULL;
+		player->SendMessage(serverConnectionGuid, GeneralConnection::kLastMessageID, (char*)buffer, msgSize*sizeof(int));
+		ABSOLUTE_TIME start = START_TIME;
+		while (GetProfileTime(ELAPSED_TIME(start)) < 2*kTimeSecond)
+		{
+			player->Poll();
+			editor->PollWithCustomMessage();
+			if (largeMessage != NULL) break;
+		}
+		if (largeMessage != NULL)
+		{
+			CHECK_EQUAL (1, largeMessage[1]);
+			CHECK_EQUAL (1024, largeMessage[1024]);
+			CHECK_EQUAL (10000, largeMessage[10000]);
+			CHECK_EQUAL (msgSize -1, largeMessage[msgSize-1]);
+		}
+		else
+			CHECK(largeMessage != NULL);
+
+		delete[] largeMessage; largeMessage = NULL;
+	}
+
+	TEST_FIXTURE(GeneralConnectionFixture, SendMultipleMessage)
+	{
+		Connect();
+		editor->RegisterMessageHandler(GeneralConnection::kLastMessageID,HandleManyMessages);
+
+		char message[20];
+
+		for(int i = 0; i < 100; i++)
+		{
+			message[10] = (char)i;
+			player->SendMessage(serverConnectionGuid, GeneralConnection::kLastMessageID, message, 20);
+		}
+
+		ABSOLUTE_TIME start = START_TIME;
+		while (GetProfileTime(ELAPSED_TIME(start)) < 1*kTimeSecond)
+		{
+			player->Poll();
+			editor->PollWithCustomMessage();
+			if (messageCount == 100) break;
+		}
+		CHECK_EQUAL (100, messageCount);
+	}
+}
+
+#endif
+
+#endif // ENABLE_PLAYERCONNECTION