CloudServer.cpp

/*
 *  Copyright (c) 2014, Oculus VR, Inc.
 *  All rights reserved.
 *
 *  This source code is licensed under the BSD-style license found in the
 *  LICENSE file in the root directory of this source tree. An additional grant 
 *  of patent rights can be found in the PATENTS file in the same directory.
 *
 */

#include "NativeFeatureIncludes.h"
#if _RAKNET_SUPPORT_CloudServer==1

#include "CloudServer.h"
#include "GetTime.h"
#include "MessageIdentifiers.h"
#include "BitStream.h"
#include "RakPeerInterface.h"

enum ServerToServerCommands
{
	STSC_PROCESS_GET_REQUEST,
	STSC_PROCESS_GET_RESPONSE,
	STSC_ADD_UPLOADED_AND_SUBSCRIBED_KEYS,
	STSC_ADD_UPLOADED_KEY,
	STSC_ADD_SUBSCRIBED_KEY,
	STSC_REMOVE_UPLOADED_KEY,
	STSC_REMOVE_SUBSCRIBED_KEY,
	STSC_DATA_CHANGED,
};

using namespace RakNet;

int CloudServer::RemoteServerComp(const RakNetGUID &key, RemoteServer* const &data )
{
	if (key < data->serverAddress)
		return -1;
	if (key > data->serverAddress)
		return 1;
	return 0;
}
int CloudServer::KeySubscriberIDComp(const CloudKey &key, KeySubscriberID * const &data )
{
	if (key.primaryKey < data->key.primaryKey)
		return -1;
	if (key.primaryKey > data->key.primaryKey)
		return 1;
	if (key.secondaryKey < data->key.secondaryKey)
		return -1;
	if (key.secondaryKey > data->key.secondaryKey)
		return 1;
	return 0;
}
int CloudServer::KeyDataPtrComp( const RakNetGUID &key, CloudData* const &data )
{
	if (key < data->clientGUID)
		return -1;
	if (key > data->clientGUID)
		return 1;
	return 0;
}
int CloudServer::KeyDataListComp( const CloudKey &key, CloudDataList * const &data )
{
	if (key.primaryKey < data->key.primaryKey)
		return -1;
	if (key.primaryKey > data->key.primaryKey)
		return 1;
	if (key.secondaryKey < data->key.secondaryKey)
		return -1;
	if (key.secondaryKey > data->key.secondaryKey)
		return 1;
	return 0;
}
int CloudServer::BufferedGetResponseFromServerComp(const RakNetGUID &key, CloudServer::BufferedGetResponseFromServer* const &data )
{
	if (key < data->serverAddress)
		return -1;
	if (key > data->serverAddress)
		return 1;
	return 0;
}
int CloudServer::GetRequestComp(const uint32_t &key, CloudServer::GetRequest* const &data )
{
	if (key < data->requestId)
		return -1;
	if (key > data->requestId)
		return -1;
	return 0;
}
void CloudServer::CloudQueryWithAddresses::Serialize(bool writeToBitstream, BitStream *bitStream)
{
	cloudQuery.Serialize(writeToBitstream, bitStream);

	if (writeToBitstream)
	{
		bitStream->WriteCasted<uint16_t>(specificSystems.Size());
		RakAssert(specificSystems.Size() < (uint16_t)-1 );
		for (uint16_t i=0; i < specificSystems.Size(); i++)
		{
			bitStream->Write(specificSystems[i]);
		}
	}
	else
	{
		uint16_t specificSystemsCount;
		RakNetGUID addressOrGuid;
		bitStream->Read(specificSystemsCount);
		for (uint16_t i=0; i < specificSystemsCount; i++)
		{
			bitStream->Read(addressOrGuid);
			specificSystems.Push(addressOrGuid, _FILE_AND_LINE_);
		}
	}
}
bool CloudServer::GetRequest::AllRemoteServersHaveResponded(void) const
{
	unsigned int i;
	for (i=0; i < remoteServerResponses.Size(); i++)
		if (remoteServerResponses[i]->gotResult==false)
			return false;
	return true;
}
void CloudServer::GetRequest::Clear(CloudAllocator *allocator)
{
	unsigned int i;
	for (i=0; i < remoteServerResponses.Size(); i++)
	{
		remoteServerResponses[i]->Clear(allocator);
		RakNet::OP_DELETE(remoteServerResponses[i], _FILE_AND_LINE_);
	}
	remoteServerResponses.Clear(false, _FILE_AND_LINE_);
}
void CloudServer::BufferedGetResponseFromServer::Clear(CloudAllocator *allocator)
{
	unsigned int i;
	for (i=0; i < queryResult.rowsReturned.Size(); i++)
	{
		allocator->DeallocateRowData(queryResult.rowsReturned[i]->data);
		allocator->DeallocateCloudQueryRow(queryResult.rowsReturned[i]);
	}
	queryResult.rowsReturned.Clear(false, _FILE_AND_LINE_);
}
CloudServer::CloudServer()
{
	maxUploadBytesPerClient=0;
	maxBytesPerDowload=0;
	nextGetRequestId=0;
	nextGetRequestsCheck=0;
}
CloudServer::~CloudServer()
{
	Clear();
}
void CloudServer::SetMaxUploadBytesPerClient(uint64_t bytes)
{
	maxUploadBytesPerClient=bytes;
}
void CloudServer::SetMaxBytesPerDownload(uint64_t bytes)
{
	maxBytesPerDowload=bytes;
}
void CloudServer::Update(void)
{
	// Timeout getRequests
	RakNet::Time time = RakNet::Time();
	if (time > nextGetRequestsCheck)
	{
		nextGetRequestsCheck=time+1000;

		unsigned int i=0;
		while (i < getRequests.Size())
		{
			if (time - getRequests[i]->requestStartTime > 3000)
			{
				// Remote server is not responding, just send back data with whoever did respond
				ProcessAndTransmitGetRequest(getRequests[i]);
				getRequests[i]->Clear(this);
				RakNet::OP_DELETE(getRequests[i],_FILE_AND_LINE_);
				getRequests.RemoveAtIndex(i);
			}
			else
			{
				i++;
			}
		}
	}
}
PluginReceiveResult CloudServer::OnReceive(Packet *packet)
{
	switch (packet->data[0])
	{
	case ID_CLOUD_POST_REQUEST:
		OnPostRequest(packet);
		return RR_STOP_PROCESSING_AND_DEALLOCATE;
	case ID_CLOUD_RELEASE_REQUEST:
		OnReleaseRequest(packet);
		return RR_STOP_PROCESSING_AND_DEALLOCATE;
	case ID_CLOUD_GET_REQUEST:
		OnGetRequest(packet);
		return RR_STOP_PROCESSING_AND_DEALLOCATE;
	case ID_CLOUD_UNSUBSCRIBE_REQUEST:
		OnUnsubscribeRequest(packet);
		return RR_STOP_PROCESSING_AND_DEALLOCATE;
	case ID_CLOUD_SERVER_TO_SERVER_COMMAND:
		if (packet->length>1)
		{
			switch (packet->data[1])
			{
			case STSC_PROCESS_GET_REQUEST:
				OnServerToServerGetRequest(packet);
				return RR_STOP_PROCESSING_AND_DEALLOCATE;
			case STSC_PROCESS_GET_RESPONSE:
				OnServerToServerGetResponse(packet);
				return RR_STOP_PROCESSING_AND_DEALLOCATE;
			case STSC_ADD_UPLOADED_AND_SUBSCRIBED_KEYS:
				OnSendUploadedAndSubscribedKeysToServer(packet);
				return RR_STOP_PROCESSING_AND_DEALLOCATE;
			case STSC_ADD_UPLOADED_KEY:
				OnSendUploadedKeyToServers(packet);
				return RR_STOP_PROCESSING_AND_DEALLOCATE;
			case STSC_ADD_SUBSCRIBED_KEY:
				OnSendSubscribedKeyToServers(packet);
				return RR_STOP_PROCESSING_AND_DEALLOCATE;
			case STSC_REMOVE_UPLOADED_KEY:
				OnRemoveUploadedKeyFromServers(packet);
				return RR_STOP_PROCESSING_AND_DEALLOCATE;
			case STSC_REMOVE_SUBSCRIBED_KEY:
				OnRemoveSubscribedKeyFromServers(packet);
				return RR_STOP_PROCESSING_AND_DEALLOCATE;
			case STSC_DATA_CHANGED:
				OnServerDataChanged(packet);
				return RR_STOP_PROCESSING_AND_DEALLOCATE;
			}
		}
		return RR_STOP_PROCESSING_AND_DEALLOCATE;
	}
	return RR_CONTINUE_PROCESSING;
}
void CloudServer::OnPostRequest(Packet *packet)
{
	RakNet::BitStream bsIn(packet->data, packet->length, false);
	bsIn.IgnoreBytes(sizeof(MessageID));
	CloudKey key;
	key.Serialize(false,&bsIn);
	uint32_t dataLengthBytes;
	bsIn.Read(dataLengthBytes);
	if (maxUploadBytesPerClient>0 && dataLengthBytes>maxUploadBytesPerClient)
		return; // Exceeded max upload bytes

	bsIn.AlignReadToByteBoundary();
	for (unsigned int filterIndex=0; filterIndex < queryFilters.Size(); filterIndex++)
	{
		if (queryFilters[filterIndex]->OnPostRequest(packet->guid, packet->systemAddress, key, dataLengthBytes, (const char*) bsIn.GetData()+BITS_TO_BYTES(bsIn.GetReadOffset()))==false)
			return;
	}

	unsigned char *data;
	if (dataLengthBytes>CLOUD_SERVER_DATA_STACK_SIZE)
	{
		data = (unsigned char *) rakMalloc_Ex(dataLengthBytes,_FILE_AND_LINE_);
		if (data==0)
		{
			notifyOutOfMemory(_FILE_AND_LINE_);
			return;
		}
		bsIn.ReadAlignedBytes(data,dataLengthBytes);
	}
	else
		data=0;

	// Add this system to remoteSystems if they aren't there already
	DataStructures::HashIndex remoteSystemsHashIndex = remoteSystems.GetIndexOf(packet->guid);
	RemoteCloudClient *remoteCloudClient;
	if (remoteSystemsHashIndex.IsInvalid())
	{
		remoteCloudClient = RakNet::OP_NEW<RemoteCloudClient>(_FILE_AND_LINE_);
		remoteCloudClient->uploadedKeys.Insert(key,key,true,_FILE_AND_LINE_);
		remoteCloudClient->uploadedBytes=0;
		remoteSystems.Push(packet->guid, remoteCloudClient, _FILE_AND_LINE_);
	}
	else
	{
		remoteCloudClient = remoteSystems.ItemAtIndex(remoteSystemsHashIndex);
		bool objectExists;
		// Add to RemoteCloudClient::uploadedKeys if it isn't there already
		unsigned int uploadedKeysIndex = remoteCloudClient->uploadedKeys.GetIndexFromKey(key,&objectExists);
		if (objectExists==false)
		{
			remoteCloudClient->uploadedKeys.InsertAtIndex(key, uploadedKeysIndex, _FILE_AND_LINE_);
		}
	}

	bool cloudDataAlreadyUploaded;
	unsigned int dataRepositoryIndex;
	bool dataRepositoryExists;
	CloudDataList* cloudDataList = GetOrAllocateCloudDataList(key, &dataRepositoryExists, dataRepositoryIndex);
	if (dataRepositoryExists==false)
	{
		cloudDataList->uploaderCount=1;
		cloudDataAlreadyUploaded=false;
	}
	else
	{
		cloudDataAlreadyUploaded=cloudDataList->uploaderCount>0;
		cloudDataList->uploaderCount++;
	}

	CloudData *cloudData;
	bool keyDataListExists;
	unsigned int keyDataListIndex = cloudDataList->keyData.GetIndexFromKey(packet->guid, &keyDataListExists);
	if (keyDataListExists==false)
	{
		if (maxUploadBytesPerClient>0 && remoteCloudClient->uploadedBytes+dataLengthBytes>maxUploadBytesPerClient)
		{
			// Undo prior insertion of cloudDataList into cloudData if needed
			if (keyDataListExists==false)
			{
				RakNet::OP_DELETE(cloudDataList,_FILE_AND_LINE_);
				dataRepository.RemoveAtIndex(dataRepositoryIndex);
			}

			if (remoteCloudClient->IsUnused())
			{
				RakNet::OP_DELETE(remoteCloudClient, _FILE_AND_LINE_);
				remoteSystems.Remove(packet->guid, _FILE_AND_LINE_);
			}

			if (dataLengthBytes>CLOUD_SERVER_DATA_STACK_SIZE)
				rakFree_Ex(data, _FILE_AND_LINE_);

			return;
		}

		cloudData = RakNet::OP_NEW<CloudData>(_FILE_AND_LINE_);
		cloudData->dataLengthBytes=dataLengthBytes;
		cloudData->isUploaded=true;
		if (forceAddress!=UNASSIGNED_SYSTEM_ADDRESS)
		{
			cloudData->serverSystemAddress=forceAddress;
			cloudData->serverSystemAddress.SetPortHostOrder(rakPeerInterface->GetExternalID(packet->systemAddress).GetPort());
		}
		else
		{
			cloudData->serverSystemAddress=rakPeerInterface->GetExternalID(packet->systemAddress);
			if (cloudData->serverSystemAddress.IsLoopback())
				cloudData->serverSystemAddress.FromString(rakPeerInterface->GetLocalIP(0));
		}
		if (cloudData->serverSystemAddress.GetPort()==0)
		{
			// Fix localhost port
			cloudData->serverSystemAddress.SetPortHostOrder(rakPeerInterface->GetSocket(UNASSIGNED_SYSTEM_ADDRESS)->GetBoundAddress().GetPort());
		}
		cloudData->clientSystemAddress=packet->systemAddress;
		cloudData->serverGUID=rakPeerInterface->GetMyGUID();
		cloudData->clientGUID=packet->guid;
		cloudDataList->keyData.Insert(packet->guid,cloudData,true,_FILE_AND_LINE_);
	}
	else
	{
		cloudData = cloudDataList->keyData[keyDataListIndex];

		if (cloudDataAlreadyUploaded==false)
		{
			if (forceAddress!=UNASSIGNED_SYSTEM_ADDRESS)
			{
				cloudData->serverSystemAddress=forceAddress;
				cloudData->serverSystemAddress.SetPortHostOrder(rakPeerInterface->GetExternalID(packet->systemAddress).GetPort());
			}
			else
			{
				cloudData->serverSystemAddress=rakPeerInterface->GetExternalID(packet->systemAddress);
			}
			if (cloudData->serverSystemAddress.GetPort()==0)
			{
				// Fix localhost port
				cloudData->serverSystemAddress.SetPortHostOrder(rakPeerInterface->GetSocket(UNASSIGNED_SYSTEM_ADDRESS)->GetBoundAddress().GetPort());
			}

			cloudData->clientSystemAddress=packet->systemAddress;
		}

		if (maxUploadBytesPerClient>0 && remoteCloudClient->uploadedBytes-cloudData->dataLengthBytes+dataLengthBytes>maxUploadBytesPerClient)
		{
			// Undo prior insertion of cloudDataList into cloudData if needed
			if (dataRepositoryExists==false)
			{
				RakNet::OP_DELETE(cloudDataList,_FILE_AND_LINE_);
				dataRepository.RemoveAtIndex(dataRepositoryIndex);
			}
			return;
		}
		else
		{
			// Subtract already used bytes we are overwriting
			remoteCloudClient->uploadedBytes-=cloudData->dataLengthBytes;
		}

		if (cloudData->allocatedData!=0)
			rakFree_Ex(cloudData->allocatedData,_FILE_AND_LINE_);
	}

	if (dataLengthBytes>CLOUD_SERVER_DATA_STACK_SIZE)
	{
		// Data already allocated
		cloudData->allocatedData=data;
		cloudData->dataPtr=data;
	}
	else
	{
		// Read to stack
		if (dataLengthBytes>0)
			bsIn.ReadAlignedBytes(cloudData->stackData,dataLengthBytes);
		cloudData->allocatedData=0;
		cloudData->dataPtr=cloudData->stackData;
	}
	// Update how many bytes were written for this data
	cloudData->dataLengthBytes=dataLengthBytes;
	remoteCloudClient->uploadedBytes+=dataLengthBytes;

	if (cloudDataAlreadyUploaded==false)
	{
		// New data field
		SendUploadedKeyToServers(cloudDataList->key);
	}

	// Existing data field changed
	NotifyClientSubscribersOfDataChange(cloudData, cloudDataList->key, cloudData->specificSubscribers, true );
	NotifyClientSubscribersOfDataChange(cloudData, cloudDataList->key, cloudDataList->nonSpecificSubscribers, true );

	// Send update to all remote servers that subscribed to this key
	NotifyServerSubscribersOfDataChange(cloudData, cloudDataList->key, true);

	// I could have also subscribed to a key not yet updated locally
	// This means I have to go through every RemoteClient that wants this key
	// Seems like cloudData->specificSubscribers is unnecessary in that case
}
void CloudServer::OnReleaseRequest(Packet *packet)
{
	RakNet::BitStream bsIn(packet->data, packet->length, false);
	bsIn.IgnoreBytes(sizeof(MessageID));

	uint16_t keyCount;
	bsIn.Read(keyCount);

	if (keyCount==0)
		return;

	DataStructures::HashIndex remoteSystemIndex = remoteSystems.GetIndexOf(packet->guid);
	if (remoteSystemIndex.IsInvalid()==true)
		return;

	RemoteCloudClient* remoteCloudClient = remoteSystems.ItemAtIndex(remoteSystemIndex);

	CloudKey key;

	// Read all in a list first so I can run filter on it
	DataStructures::List<CloudKey> cloudKeys;
	for (uint16_t keyCountIndex=0; keyCountIndex < keyCount; keyCountIndex++)
	{
		key.Serialize(false, &bsIn);
		cloudKeys.Push(key, _FILE_AND_LINE_);
	}

	for (unsigned int filterIndex=0; filterIndex < queryFilters.Size(); filterIndex++)
	{
		if (queryFilters[filterIndex]->OnReleaseRequest(packet->guid, packet->systemAddress, cloudKeys)==false)
			return;
	}

	for (uint16_t keyCountIndex=0; keyCountIndex < keyCount; keyCountIndex++)
	{
		// Serialize in list above so I can run the filter on it
		// key.Serialize(false, &bsIn);
		key=cloudKeys[keyCountIndex];

		// Remove remote systems uploaded keys
		bool objectExists;
		unsigned int uploadedKeysIndex = remoteCloudClient->uploadedKeys.GetIndexFromKey(key,&objectExists);
		if (objectExists)
		{
			bool dataRepositoryExists;
			unsigned int dataRepositoryIndex = dataRepository.GetIndexFromKey(key, &dataRepositoryExists);
			CloudDataList* cloudDataList = dataRepository[dataRepositoryIndex];
			RakAssert(cloudDataList);

			CloudData *cloudData;
			bool keyDataListExists;
			unsigned int keyDataListIndex = cloudDataList->keyData.GetIndexFromKey(packet->guid, &keyDataListExists);
			cloudData = cloudDataList->keyData[keyDataListIndex];

			remoteCloudClient->uploadedKeys.RemoveAtIndex(uploadedKeysIndex);
			remoteCloudClient->uploadedBytes-=cloudData->dataLengthBytes;
			cloudDataList->uploaderCount--;

			// Broadcast destruction of this key to subscribers
			NotifyClientSubscribersOfDataChange(cloudData, cloudDataList->key, cloudData->specificSubscribers, false );
			NotifyClientSubscribersOfDataChange(cloudData, cloudDataList->key, cloudDataList->nonSpecificSubscribers, false );
			NotifyServerSubscribersOfDataChange(cloudData, cloudDataList->key, false );

			cloudData->Clear();

			if (cloudData->IsUnused())
			{
				RakNet::OP_DELETE(cloudData, _FILE_AND_LINE_);
				cloudDataList->keyData.RemoveAtIndex(keyDataListIndex);
				if (cloudDataList->IsNotUploaded())
				{
					// Tell other servers that this key is no longer uploaded, so they do not request it from us
					RemoveUploadedKeyFromServers(cloudDataList->key);
				}

				if (cloudDataList->IsUnused())
				{
					RakNet::OP_DELETE(cloudDataList, _FILE_AND_LINE_);
					dataRepository.RemoveAtIndex(dataRepositoryIndex);
				}
			}

			if (remoteCloudClient->IsUnused())
			{
				RakNet::OP_DELETE(remoteCloudClient, _FILE_AND_LINE_);
				remoteSystems.RemoveAtIndex(remoteSystemIndex, _FILE_AND_LINE_);
				break;
			}
		}
	}
}
void CloudServer::OnGetRequest(Packet *packet)
{
	RakNet::BitStream bsIn(packet->data, packet->length, false);
	bsIn.IgnoreBytes(sizeof(MessageID));
	uint16_t specificSystemsCount;
	CloudKey cloudKey;

	// Create a new GetRequest
	GetRequest *getRequest;
	getRequest = RakNet::OP_NEW<GetRequest>(_FILE_AND_LINE_);
	getRequest->cloudQueryWithAddresses.cloudQuery.Serialize(false, &bsIn);
	getRequest->requestingClient=packet->guid;

	RakNetGUID addressOrGuid;
	bsIn.Read(specificSystemsCount);
	for (uint16_t i=0; i < specificSystemsCount; i++)
	{
		bsIn.Read(addressOrGuid);
		getRequest->cloudQueryWithAddresses.specificSystems.Push(addressOrGuid, _FILE_AND_LINE_);
	}

	if (getRequest->cloudQueryWithAddresses.cloudQuery.keys.Size()==0)
	{
		RakNet::OP_DELETE(getRequest, _FILE_AND_LINE_);
		return;
	}

	for (unsigned int filterIndex=0; filterIndex < queryFilters.Size(); filterIndex++)
	{
		if (queryFilters[filterIndex]->OnGetRequest(packet->guid, packet->systemAddress, getRequest->cloudQueryWithAddresses.cloudQuery, getRequest->cloudQueryWithAddresses.specificSystems )==false)
			return;
	}

	getRequest->requestStartTime=RakNet::GetTime();
	getRequest->requestId=nextGetRequestId++;

	// Send request to servers that have this data
	DataStructures::List<RemoteServer*> remoteServersWithData;
	GetServersWithUploadedKeys(getRequest->cloudQueryWithAddresses.cloudQuery.keys, remoteServersWithData);

	if (remoteServersWithData.Size()==0)
	{
		ProcessAndTransmitGetRequest(getRequest);
	}
	else
	{
		RakNet::BitStream bsOut;
		bsOut.Write((MessageID)ID_CLOUD_SERVER_TO_SERVER_COMMAND);
		bsOut.Write((MessageID)STSC_PROCESS_GET_REQUEST);
		getRequest->cloudQueryWithAddresses.Serialize(true, &bsOut);
		bsOut.Write(getRequest->requestId);

		for (unsigned int remoteServerIndex=0; remoteServerIndex < remoteServersWithData.Size(); remoteServerIndex++)
		{
			BufferedGetResponseFromServer* bufferedGetResponseFromServer = RakNet::OP_NEW<BufferedGetResponseFromServer>(_FILE_AND_LINE_);
			bufferedGetResponseFromServer->serverAddress=remoteServersWithData[remoteServerIndex]->serverAddress;
			bufferedGetResponseFromServer->gotResult=false;
			getRequest->remoteServerResponses.Insert(remoteServersWithData[remoteServerIndex]->serverAddress, bufferedGetResponseFromServer, true, _FILE_AND_LINE_);

			SendUnified(&bsOut, HIGH_PRIORITY, RELIABLE_ORDERED, 0, remoteServersWithData[remoteServerIndex]->serverAddress, false);
		}

		// Record that this system made this request
		getRequests.Insert(getRequest->requestId, getRequest, true, _FILE_AND_LINE_);
	}

	if (getRequest->cloudQueryWithAddresses.cloudQuery.subscribeToResults)
	{
		// Add to key subscription list for the client, which contains a keyId / specificUploaderList pair
		DataStructures::HashIndex remoteSystemsHashIndex = remoteSystems.GetIndexOf(packet->guid);
		RemoteCloudClient *remoteCloudClient;
		if (remoteSystemsHashIndex.IsInvalid())
		{
			remoteCloudClient = RakNet::OP_NEW<RemoteCloudClient>(_FILE_AND_LINE_);
			remoteCloudClient->uploadedBytes=0;
			remoteSystems.Push(packet->guid, remoteCloudClient, _FILE_AND_LINE_);
		}
		else
		{
			remoteCloudClient = remoteSystems.ItemAtIndex(remoteSystemsHashIndex);
		}

		unsigned int keyIndex;
		for (keyIndex=0; keyIndex < getRequest->cloudQueryWithAddresses.cloudQuery.keys.Size(); keyIndex++)
		{
			cloudKey = getRequest->cloudQueryWithAddresses.cloudQuery.keys[keyIndex];

			unsigned int keySubscriberIndex;
			bool hasKeySubscriber;
			keySubscriberIndex = remoteCloudClient->subscribedKeys.GetIndexFromKey(cloudKey, &hasKeySubscriber);
			KeySubscriberID* keySubscriberId;
			if (hasKeySubscriber)
			{
				DataStructures::List<RakNetGUID> specificSystems;
				UnsubscribeFromKey(remoteCloudClient, packet->guid, keySubscriberIndex, cloudKey, specificSystems);
			}

			keySubscriberId = RakNet::OP_NEW<KeySubscriberID>(_FILE_AND_LINE_);
			keySubscriberId->key=cloudKey;

			unsigned int specificSystemIndex;
			for (specificSystemIndex=0; specificSystemIndex < getRequest->cloudQueryWithAddresses.specificSystems.Size(); specificSystemIndex++)
			{
				keySubscriberId->specificSystemsSubscribedTo.Insert(getRequest->cloudQueryWithAddresses.specificSystems[specificSystemIndex], getRequest->cloudQueryWithAddresses.specificSystems[specificSystemIndex], true, _FILE_AND_LINE_);
			}

			remoteCloudClient->subscribedKeys.InsertAtIndex(keySubscriberId, keySubscriberIndex, _FILE_AND_LINE_);

			// Add CloudData in a similar way
			unsigned int dataRepositoryIndex;
			bool dataRepositoryExists;
			CloudDataList* cloudDataList = GetOrAllocateCloudDataList(cloudKey, &dataRepositoryExists, dataRepositoryIndex);

			// If this is the first local client to subscribe to this key, call SendSubscribedKeyToServers
			if (cloudDataList->subscriberCount==0)
				SendSubscribedKeyToServers(cloudKey);

			// If the subscription is specific, may have to also allocate CloudData
			if (getRequest->cloudQueryWithAddresses.specificSystems.Size())
			{
				CloudData *cloudData;
				bool keyDataListExists;

				unsigned int specificSystemIndex;
				for (specificSystemIndex=0; specificSystemIndex < getRequest->cloudQueryWithAddresses.specificSystems.Size(); specificSystemIndex++)
				{
					RakNetGUID specificSystem = getRequest->cloudQueryWithAddresses.specificSystems[specificSystemIndex];

					unsigned int keyDataListIndex = cloudDataList->keyData.GetIndexFromKey(specificSystem, &keyDataListExists);
					if (keyDataListExists==false)
					{
						cloudData = RakNet::OP_NEW<CloudData>(_FILE_AND_LINE_);
						cloudData->dataLengthBytes=0;
						cloudData->allocatedData=0;
						cloudData->isUploaded=false;
						cloudData->dataPtr=0;
						cloudData->serverSystemAddress=UNASSIGNED_SYSTEM_ADDRESS;
						cloudData->clientSystemAddress=UNASSIGNED_SYSTEM_ADDRESS;
						cloudData->serverGUID=rakPeerInterface->GetMyGUID();
						cloudData->clientGUID=specificSystem;
						cloudDataList->keyData.Insert(specificSystem,cloudData,true,_FILE_AND_LINE_);
					}
					else
					{
						cloudData = cloudDataList->keyData[keyDataListIndex];
					}

					++cloudDataList->subscriberCount;
					cloudData->specificSubscribers.Insert(packet->guid, packet->guid, true, _FILE_AND_LINE_);
				}
			}
			else
			{
				++cloudDataList->subscriberCount;
				cloudDataList->nonSpecificSubscribers.Insert(packet->guid, packet->guid, true, _FILE_AND_LINE_);

				// Remove packet->guid from CloudData::specificSubscribers among all instances of cloudDataList->keyData
				unsigned int subscribedKeysIndex;
				bool subscribedKeysIndexExists;
				subscribedKeysIndex = remoteCloudClient->subscribedKeys.GetIndexFromKey(cloudDataList->key, &subscribedKeysIndexExists);
				if (subscribedKeysIndexExists)
				{
					KeySubscriberID* keySubscriberId;
					keySubscriberId = remoteCloudClient->subscribedKeys[subscribedKeysIndex];
					unsigned int specificSystemIndex;
					for (specificSystemIndex=0; specificSystemIndex < keySubscriberId->specificSystemsSubscribedTo.Size(); specificSystemIndex++)
					{
						bool keyDataExists;
						unsigned int keyDataIndex = cloudDataList->keyData.GetIndexFromKey(keySubscriberId->specificSystemsSubscribedTo[specificSystemIndex], &keyDataExists);
						if (keyDataExists)
						{
							CloudData *keyData = cloudDataList->keyData[keyDataIndex];
							keyData->specificSubscribers.Remove(packet->guid);
							--cloudDataList->subscriberCount;
						}
					}
				}
			}
		}

		if (remoteCloudClient->subscribedKeys.Size()==0)
		{
			// Didn't do anything
			remoteSystems.Remove(packet->guid, _FILE_AND_LINE_);
			RakNet::OP_DELETE(remoteCloudClient, _FILE_AND_LINE_);
		}
	}

	if (remoteServersWithData.Size()==0)
		RakNet::OP_DELETE(getRequest, _FILE_AND_LINE_);
}
void CloudServer::OnUnsubscribeRequest(Packet *packet)
{
	RakNet::BitStream bsIn(packet->data, packet->length, false);
	bsIn.IgnoreBytes(sizeof(MessageID));

	DataStructures::HashIndex remoteSystemIndex = remoteSystems.GetIndexOf(packet->guid);
	if (remoteSystemIndex.IsInvalid()==true)
		return;

	RemoteCloudClient* remoteCloudClient = remoteSystems.ItemAtIndex(remoteSystemIndex);

	uint16_t keyCount, specificSystemCount;
	DataStructures::List<CloudKey> cloudKeys;
	DataStructures::List<RakNetGUID> specificSystems;
	uint16_t index;

	CloudKey cloudKey;
	bsIn.Read(keyCount);
	for (index=0; index < keyCount; index++)
	{
		cloudKey.Serialize(false, &bsIn);
		cloudKeys.Push(cloudKey, _FILE_AND_LINE_);
	}

	RakNetGUID specificSystem;
	bsIn.Read(specificSystemCount);
	for (index=0; index < specificSystemCount; index++)
	{
		bsIn.Read(specificSystem);
		specificSystems.Push(specificSystem, _FILE_AND_LINE_);
	}

	for (unsigned int filterIndex=0; filterIndex < queryFilters.Size(); filterIndex++)
	{
		if (queryFilters[filterIndex]->OnUnsubscribeRequest(packet->guid, packet->systemAddress, cloudKeys, specificSystems )==false)
			return;
	}

//	CloudDataList *cloudDataList;
	bool dataRepositoryExists;
//	unsigned int dataRepositoryIndex;

	for (index=0; index < keyCount; index++)
	{
		CloudKey cloudKey = cloudKeys[index];

	//	dataRepositoryIndex = 
			dataRepository.GetIndexFromKey(cloudKey, &dataRepositoryExists);
		if (dataRepositoryExists==false)
			continue;
//		cloudDataList = dataRepository[dataRepositoryIndex];

		unsigned int keySubscriberIndex;
		bool hasKeySubscriber;
		keySubscriberIndex = remoteCloudClient->subscribedKeys.GetIndexFromKey(cloudKey, &hasKeySubscriber);

		if (hasKeySubscriber==false)
			continue;

		UnsubscribeFromKey(remoteCloudClient, packet->guid, keySubscriberIndex, cloudKey, specificSystems);
	}

	if (remoteCloudClient->IsUnused())
	{
		RakNet::OP_DELETE(remoteCloudClient, _FILE_AND_LINE_);
		remoteSystems.RemoveAtIndex(remoteSystemIndex, _FILE_AND_LINE_);
	}
}
void CloudServer::OnServerToServerGetRequest(Packet *packet)
{
//	unsigned int remoteServerIndex;
	bool objectExists;
	//remoteServerIndex =
		remoteServers.GetIndexFromKey(packet->guid, &objectExists);
	if (objectExists==false)
		return;

	RakNet::BitStream bsIn(packet->data, packet->length, false);
	bsIn.IgnoreBytes(sizeof(MessageID)*2);

	CloudQueryWithAddresses cloudQueryWithAddresses;
	uint32_t requestId;
	cloudQueryWithAddresses.Serialize(false, &bsIn);
	bsIn.Read(requestId);

	DataStructures::List<CloudData*> cloudDataResultList;
	DataStructures::List<CloudKey> cloudKeyResultList;
	ProcessCloudQueryWithAddresses(cloudQueryWithAddresses, cloudDataResultList, cloudKeyResultList);

	RakNet::BitStream bsOut;
	bsOut.Write((MessageID)ID_CLOUD_SERVER_TO_SERVER_COMMAND);
	bsOut.Write((MessageID)STSC_PROCESS_GET_RESPONSE);
	bsOut.Write(requestId);
	WriteCloudQueryRowFromResultList(cloudDataResultList, cloudKeyResultList, &bsOut);
	SendUnified(&bsOut, HIGH_PRIORITY, RELIABLE_ORDERED, 0, packet->guid, false);
}
void CloudServer::OnServerToServerGetResponse(Packet *packet)
{
	unsigned int remoteServerIndex;
	bool objectExists;
	remoteServerIndex = remoteServers.GetIndexFromKey(packet->guid, &objectExists);
	if (objectExists==false)
		return;

	RemoteServer *remoteServer = remoteServers[remoteServerIndex];
	if (remoteServer==0)
		return;

	RakNet::BitStream bsIn(packet->data, packet->length, false);
	bsIn.IgnoreBytes(sizeof(MessageID)*2);

	uint32_t requestId;
	bsIn.Read(requestId);

	// Lookup request id
	bool hasGetRequest;
	unsigned int getRequestIndex;
	getRequestIndex = getRequests.GetIndexFromKey(requestId, &hasGetRequest);
	if (hasGetRequest==false)
		return;
	GetRequest *getRequest = getRequests[getRequestIndex];
	bool hasRemoteServer;
	unsigned int remoteServerResponsesIndex;
	remoteServerResponsesIndex = getRequest->remoteServerResponses.GetIndexFromKey(packet->guid, &hasRemoteServer);
	if (hasRemoteServer==false)
		return;
	BufferedGetResponseFromServer *bufferedGetResponseFromServer;
	bufferedGetResponseFromServer = getRequest->remoteServerResponses[remoteServerResponsesIndex];
	if (bufferedGetResponseFromServer->gotResult==true)
		return;
	bufferedGetResponseFromServer->gotResult=true;
	uint32_t numRows;
	bufferedGetResponseFromServer->queryResult.SerializeNumRows(false, numRows, &bsIn);
	bufferedGetResponseFromServer->queryResult.SerializeCloudQueryRows(false, numRows, &bsIn, this);

	// If all results returned, then also process locally, and return to user
	if (getRequest->AllRemoteServersHaveResponded())
	{
		ProcessAndTransmitGetRequest(getRequest);

		getRequest->Clear(this);
		RakNet::OP_DELETE(getRequest, _FILE_AND_LINE_);

		getRequests.RemoveAtIndex(getRequestIndex);
	}
}
void CloudServer::OnClosedConnection(const SystemAddress &systemAddress, RakNetGUID rakNetGUID, PI2_LostConnectionReason lostConnectionReason )
{
	(void) lostConnectionReason;
	(void) systemAddress;

	unsigned int remoteServerIndex;
	bool objectExists;
	remoteServerIndex = remoteServers.GetIndexFromKey(rakNetGUID, &objectExists);
	if (objectExists)
	{
		// Update remoteServerResponses by removing this server and sending the response if it is now complete
		unsigned int getRequestIndex=0;
		while (getRequestIndex < getRequests.Size())
		{
			GetRequest *getRequest = getRequests[getRequestIndex];
			bool waitingForThisServer;
			unsigned int remoteServerResponsesIndex = getRequest->remoteServerResponses.GetIndexFromKey(rakNetGUID, &waitingForThisServer);
			if (waitingForThisServer)
			{
				getRequest->remoteServerResponses[remoteServerResponsesIndex]->Clear(this);
				RakNet::OP_DELETE(getRequest->remoteServerResponses[remoteServerResponsesIndex], _FILE_AND_LINE_);
				getRequest->remoteServerResponses.RemoveAtIndex(remoteServerResponsesIndex);

				if (getRequest->AllRemoteServersHaveResponded())
				{
					ProcessAndTransmitGetRequest(getRequest);
					getRequest->Clear(this);
					RakNet::OP_DELETE(getRequest, _FILE_AND_LINE_);

					getRequests.RemoveAtIndex(getRequestIndex);
				}
				else
					getRequestIndex++;
			}
			else
				getRequestIndex++;
		}

		RakNet::OP_DELETE(remoteServers[remoteServerIndex],_FILE_AND_LINE_);
		remoteServers.RemoveAtIndex(remoteServerIndex);
	}

	DataStructures::HashIndex remoteSystemIndex = remoteSystems.GetIndexOf(rakNetGUID);
	if (remoteSystemIndex.IsInvalid()==false)
	{
		RemoteCloudClient* remoteCloudClient = remoteSystems.ItemAtIndex(remoteSystemIndex);
		unsigned int uploadedKeysIndex;
		for (uploadedKeysIndex=0; uploadedKeysIndex < remoteCloudClient->uploadedKeys.Size(); uploadedKeysIndex++)
		{
			// Delete keys this system has uploaded
			bool keyDataRepositoryExists;
			unsigned int dataRepositoryIndex = dataRepository.GetIndexFromKey(remoteCloudClient->uploadedKeys[uploadedKeysIndex], &keyDataRepositoryExists);
			if (keyDataRepositoryExists)
			{
				CloudDataList* cloudDataList = dataRepository[dataRepositoryIndex];
				bool keyDataExists;
				unsigned int keyDataIndex = cloudDataList->keyData.GetIndexFromKey(rakNetGUID, &keyDataExists);
				if (keyDataExists)
				{
					CloudData *cloudData = cloudDataList->keyData[keyDataIndex];
					cloudDataList->uploaderCount--;

					NotifyClientSubscribersOfDataChange(cloudData, cloudDataList->key, cloudData->specificSubscribers, false );
					NotifyClientSubscribersOfDataChange(cloudData, cloudDataList->key, cloudDataList->nonSpecificSubscribers, false );
					NotifyServerSubscribersOfDataChange(cloudData, cloudDataList->key, false );

					cloudData->Clear();

					if (cloudData->IsUnused())
					{
						RakNet::OP_DELETE(cloudData,_FILE_AND_LINE_);
						cloudDataList->keyData.RemoveAtIndex(keyDataIndex);

						if (cloudDataList->IsNotUploaded())
						{
							// Tell other servers that this key is no longer uploaded, so they do not request it from us
							RemoveUploadedKeyFromServers(cloudDataList->key);
						}

						if (cloudDataList->IsUnused())
						{
							// Tell other servers that this key is no longer uploaded, so they do not request it from us
							RemoveUploadedKeyFromServers(cloudDataList->key);

							RakNet::OP_DELETE(cloudDataList, _FILE_AND_LINE_);
							dataRepository.RemoveAtIndex(dataRepositoryIndex);
						}
					}
				}
			}
		}

		unsigned int subscribedKeysIndex;
		for (subscribedKeysIndex=0; subscribedKeysIndex < remoteCloudClient->subscribedKeys.Size(); subscribedKeysIndex++)
		{
			KeySubscriberID* keySubscriberId;
			keySubscriberId = remoteCloudClient->subscribedKeys[subscribedKeysIndex];

			bool keyDataRepositoryExists;
			unsigned int keyDataRepositoryIndex = dataRepository.GetIndexFromKey(remoteCloudClient->subscribedKeys[subscribedKeysIndex]->key, &keyDataRepositoryExists);
			if (keyDataRepositoryExists)
			{
				CloudDataList* cloudDataList = dataRepository[keyDataRepositoryIndex];
				if (keySubscriberId->specificSystemsSubscribedTo.Size()==0)
				{
					cloudDataList->nonSpecificSubscribers.Remove(rakNetGUID);
					--cloudDataList->subscriberCount;
				}
				else
				{
					unsigned int specificSystemIndex;
					for (specificSystemIndex=0; specificSystemIndex < keySubscriberId->specificSystemsSubscribedTo.Size(); specificSystemIndex++)
					{
						bool keyDataExists;
						unsigned int keyDataIndex = cloudDataList->keyData.GetIndexFromKey(keySubscriberId->specificSystemsSubscribedTo[specificSystemIndex], &keyDataExists);
						if (keyDataExists)
						{
							CloudData *keyData = cloudDataList->keyData[keyDataIndex];
							keyData->specificSubscribers.Remove(rakNetGUID);
							--cloudDataList->subscriberCount;
						}
					}
				}
			}

			RakNet::OP_DELETE(keySubscriberId, _FILE_AND_LINE_);
		}

		// Delete and remove from remoteSystems
		RakNet::OP_DELETE(remoteCloudClient, _FILE_AND_LINE_);
		remoteSystems.RemoveAtIndex(remoteSystemIndex, _FILE_AND_LINE_);
	}
}
void CloudServer::OnRakPeerShutdown(void)
{
	Clear();