CloudServer.h

Go to the documentation of this file.

/*
 *  Original work: Copyright (c) 2014, Oculus VR, Inc.
 *  All rights reserved.
 *
 *  This source code is licensed under the BSD-style license found in the
 *  RakNet License.txt file in the licenses directory of this source tree. An additional grant 
 *  of patent rights can be found in the RakNet Patents.txt file in the same directory.
 *
 *
 *  Modified work: Copyright (c) 2017, SLikeSoft UG (haftungsbeschränkt)
 *
 *  This source code was modified by SLikeSoft. Modifications are licensed under the MIT-style
 *  license found in the license.txt file in the root directory of this source tree.
 */



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

#ifndef __CLOUD_SERVER_H
#define __CLOUD_SERVER_H

#include "PluginInterface2.h"
#include "memoryoverride.h"
#include "NativeTypes.h"
#include "string.h"
#include "DS_Hash.h"
#include "CloudCommon.h"
#include "DS_OrderedList.h"

#define CLOUD_SERVER_DATA_STACK_SIZE 32

namespace SLNet
{
class RakPeerInterface;

class RAK_DLL_EXPORT CloudServerQueryFilter
{
public:
    CloudServerQueryFilter() {}
    virtual ~CloudServerQueryFilter() {}

    virtual bool OnPostRequest(RakNetGUID clientGuid, SystemAddress clientAddress, CloudKey key, uint32_t dataLength, const char *data)=0;

    virtual bool OnReleaseRequest(RakNetGUID clientGuid, SystemAddress clientAddress, DataStructures::List<CloudKey> &cloudKeys)=0;

    virtual bool OnGetRequest(RakNetGUID clientGuid, SystemAddress clientAddress, CloudQuery &query, DataStructures::List<RakNetGUID> &specificSystems)=0;

    virtual bool OnUnsubscribeRequest(RakNetGUID clientGuid, SystemAddress clientAddress, DataStructures::List<CloudKey> &cloudKeys, DataStructures::List<RakNetGUID> &specificSystems)=0;
};

class RAK_DLL_EXPORT CloudServer : public PluginInterface2, CloudAllocator
{
public:
    // GetInstance() and DestroyInstance(instance*)
    STATIC_FACTORY_DECLARATIONS(CloudServer)

    CloudServer();
    virtual ~CloudServer();

    void SetMaxUploadBytesPerClient(uint64_t bytes);

    void SetMaxBytesPerDownload(uint64_t bytes);

    void AddServer(RakNetGUID systemIdentifier);

    void RemoveServer(RakNetGUID systemIdentifier);

    void GetRemoteServers(DataStructures::List<RakNetGUID> &remoteServersOut);

    void Clear(void);

    void ForceExternalSystemAddress(SystemAddress forcedAddress);

    void AddQueryFilter(CloudServerQueryFilter* filter);

    void RemoveQueryFilter(CloudServerQueryFilter* filter);

    void RemoveAllQueryFilters(void);

protected:
    virtual void Update(void);
    virtual PluginReceiveResult OnReceive(Packet *packet);
    virtual void OnClosedConnection(const SystemAddress &systemAddress, RakNetGUID rakNetGUID, PI2_LostConnectionReason lostConnectionReason );
    virtual void OnRakPeerShutdown(void);


    virtual void OnPostRequest(Packet *packet);
    virtual void OnReleaseRequest(Packet *packet);
    virtual void OnGetRequest(Packet *packet);
    virtual void OnUnsubscribeRequest(Packet *packet);
    virtual void OnServerToServerGetRequest(Packet *packet);
    virtual void OnServerToServerGetResponse(Packet *packet);

    uint64_t maxUploadBytesPerClient, maxBytesPerDowload;

    // ----------------------------------------------------------------------------
    // For a given data key, quickly look up one or all systems that have uploaded
    // ----------------------------------------------------------------------------
    struct CloudData
    {
        CloudData() {}
        ~CloudData() {if (allocatedData) rakFree_Ex(allocatedData, _FILE_AND_LINE_);}
        bool IsUnused(void) const {return isUploaded==false && specificSubscribers.Size()==0;}
        void Clear(void) {if (dataPtr==allocatedData) rakFree_Ex(allocatedData, _FILE_AND_LINE_); allocatedData=0; dataPtr=0; dataLengthBytes=0; isUploaded=false;}

        unsigned char stackData[CLOUD_SERVER_DATA_STACK_SIZE];
        unsigned char *allocatedData; // Uses allocatedData instead of stackData if length of data exceeds CLOUD_SERVER_DATA_STACK_SIZE
        unsigned char *dataPtr; // Points to either stackData or allocatedData
        uint32_t dataLengthBytes;
        bool isUploaded;

        SystemAddress serverSystemAddress;

        SystemAddress clientSystemAddress;

        RakNetGUID serverGUID;

        RakNetGUID clientGUID;

        DataStructures::OrderedList<RakNetGUID, RakNetGUID> specificSubscribers;
    };
    void WriteCloudQueryRowFromResultList(unsigned int i, DataStructures::List<CloudData*> &cloudDataResultList, DataStructures::List<CloudKey> &cloudKeyResultList, BitStream *bsOut);
    void WriteCloudQueryRowFromResultList(DataStructures::List<CloudData*> &cloudDataResultList, DataStructures::List<CloudKey> &cloudKeyResultList, BitStream *bsOut);

    static int KeyDataPtrComp( const RakNetGUID &key, CloudData* const &data );
    struct CloudDataList
    {
        bool IsUnused(void) const {return keyData.Size()==0 && nonSpecificSubscribers.Size()==0;}
        bool IsNotUploaded(void) const {return uploaderCount==0;}
        bool RemoveSubscriber(RakNetGUID g) {
            bool objectExists;
            unsigned int index;
            index = nonSpecificSubscribers.GetIndexFromKey(g, &objectExists);
            if (objectExists)
            {
                subscriberCount--;
                nonSpecificSubscribers.RemoveAtIndex(index);
                return true;
            }
            return false;
        }

        unsigned int uploaderCount, subscriberCount;
        CloudKey key;

        // Data uploaded from or subscribed to for various systems
        DataStructures::OrderedList<RakNetGUID, CloudData*, CloudServer::KeyDataPtrComp> keyData;

        DataStructures::OrderedList<RakNetGUID, RakNetGUID> nonSpecificSubscribers;
    };

    static int KeyDataListComp( const CloudKey &key, CloudDataList * const &data );
    DataStructures::OrderedList<CloudKey, CloudDataList*, CloudServer::KeyDataListComp> dataRepository;

    struct KeySubscriberID
    {
        CloudKey key;
        DataStructures::OrderedList<RakNetGUID, RakNetGUID> specificSystemsSubscribedTo;
    };
    static int KeySubscriberIDComp(const CloudKey &key, KeySubscriberID * const &data );

    // Remote systems
    struct RemoteCloudClient
    {
        bool IsUnused(void) const {return uploadedKeys.Size()==0 && subscribedKeys.Size()==0;}

        DataStructures::OrderedList<CloudKey,CloudKey,CloudKeyComp> uploadedKeys;
        DataStructures::OrderedList<CloudKey,KeySubscriberID*,CloudServer::KeySubscriberIDComp> subscribedKeys;
        uint64_t uploadedBytes;
    };
    DataStructures::Hash<RakNetGUID, RemoteCloudClient*, 2048, RakNetGUID::ToUint32> remoteSystems;

    // For a given user, release all subscribed and uploaded keys
    void ReleaseSystem(RakNetGUID clientAddress );

    // For a given user, release a set of keys
    void ReleaseKeys(RakNetGUID clientAddress, DataStructures::List<CloudKey> &keys );

    void NotifyClientSubscribersOfDataChange( CloudData *cloudData, CloudKey &key, DataStructures::OrderedList<RakNetGUID, RakNetGUID> &subscribers, bool wasUpdated );
    void NotifyClientSubscribersOfDataChange( CloudQueryRow *row, DataStructures::OrderedList<RakNetGUID, RakNetGUID> &subscribers, bool wasUpdated );
    void NotifyServerSubscribersOfDataChange( CloudData *cloudData, CloudKey &key, bool wasUpdated );

    struct RemoteServer
    {
        RakNetGUID serverAddress;
        // This server needs to know about these keys when they are updated or deleted
        DataStructures::OrderedList<CloudKey,CloudKey,CloudKeyComp> subscribedKeys;
        // This server has uploaded these keys, and needs to know about Get() requests
        DataStructures::OrderedList<CloudKey,CloudKey,CloudKeyComp> uploadedKeys;

        // Just for processing
        bool workingFlag;

        // If false, we don't know what keys they have yet, so send everything
        bool gotSubscribedAndUploadedKeys;
    };

    static int RemoteServerComp(const RakNetGUID &key, RemoteServer* const &data );
    DataStructures::OrderedList<RakNetGUID, RemoteServer*, CloudServer::RemoteServerComp> remoteServers;

    struct BufferedGetResponseFromServer
    {
        void Clear(CloudAllocator *allocator);

        RakNetGUID serverAddress;
        CloudQueryResult queryResult;
        bool gotResult;
    };

    struct CloudQueryWithAddresses
    {
        // Inputs
        CloudQuery cloudQuery;
        DataStructures::List<RakNetGUID> specificSystems;

        void Serialize(bool writeToBitstream, BitStream *bitStream);
    };

    static int BufferedGetResponseFromServerComp(const RakNetGUID &key, BufferedGetResponseFromServer* const &data );
    struct GetRequest
    {
        void Clear(CloudAllocator *allocator);
        bool AllRemoteServersHaveResponded(void) const;
        CloudQueryWithAddresses cloudQueryWithAddresses;

        // When request started. If takes too long for a response from another system, can abort remaining systems
        SLNet::Time requestStartTime;

        // Assigned by server that gets the request to identify response. See nextGetRequestId
        uint32_t requestId;

        RakNetGUID requestingClient;

        DataStructures::OrderedList<RakNetGUID, BufferedGetResponseFromServer*, CloudServer::BufferedGetResponseFromServerComp> remoteServerResponses;
    };
    static int GetRequestComp(const uint32_t &key, GetRequest* const &data );
    DataStructures::OrderedList<uint32_t, GetRequest*, CloudServer::GetRequestComp> getRequests;
    SLNet::Time nextGetRequestsCheck;

    uint32_t nextGetRequestId;

    void ProcessAndTransmitGetRequest(GetRequest *getRequest);

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

    void SendUploadedAndSubscribedKeysToServer( RakNetGUID systemAddress );
    void SendUploadedKeyToServers( CloudKey &cloudKey );
    void SendSubscribedKeyToServers( CloudKey &cloudKey );
    void RemoveUploadedKeyFromServers( CloudKey &cloudKey );
    void RemoveSubscribedKeyFromServers( CloudKey &cloudKey );

    void OnSendUploadedAndSubscribedKeysToServer( Packet *packet );
    void OnSendUploadedKeyToServers( Packet *packet );
    void OnSendSubscribedKeyToServers( Packet *packet );
    void OnRemoveUploadedKeyFromServers( Packet *packet );
    void OnRemoveSubscribedKeyFromServers( Packet *packet );
    void OnServerDataChanged( Packet *packet );

    void GetServersWithUploadedKeys(
        DataStructures::List<CloudKey> &keys,
        DataStructures::List<RemoteServer*> &remoteServersWithData
        );

    CloudServer::CloudDataList *GetOrAllocateCloudDataList(CloudKey key, bool *dataRepositoryExists, unsigned int &dataRepositoryIndex);

    void UnsubscribeFromKey(RemoteCloudClient *remoteCloudClient, RakNetGUID remoteCloudClientGuid, unsigned int keySubscriberIndex, CloudKey &cloudKey, DataStructures::List<RakNetGUID> &specificSystems);
    void RemoveSpecificSubscriber(RakNetGUID specificSubscriber, CloudDataList *cloudDataList, RakNetGUID remoteCloudClientGuid);

    DataStructures::List<CloudServerQueryFilter*> queryFilters;

    SystemAddress forceAddress;
};


} // namespace SLNet

#endif


// Key subscription
//
// A given system can subscribe to one or more keys.
// The subscription can be further be defined as only subscribing to keys uploaded by or changed by a given system.
// It is possible to subscribe to keys not yet uploaded, or uploaded to another system
//
// Operations:
//
// 1. SubscribeToKey() - Get() operation with subscription
// A. Add to key subscription list for the client, which contains a keyId / specificUploaderList pair
// B. Send to remote servers that for this key, they should send us updates
// C. (Done, get operation returns current values)
//
// 2. UpdateData() - Post() operation
// A. Find all subscribers to this data, for the uploading system.
// B. Send them the uploaded data
// C. Find all servers that subscribe to this data
// D. Send them the uploaded data
//
// 3. DeleteData() - Release() operation
// A. Find all subscribers to this data, for the deleting system.
// B. Inform them of the deletion
// C. Find all servers that subscribe to this data
// D. Inform them of the deletion
//
// 4. Unsubscribe()
// A. Find this subscriber, and remove their subscription
// B. If no one else is subscribing to this key for any system, notify remote servers we no longer need subscription updates
//
// Internal operations:
//
// 1. Find if any connected client has subscribed to a given key
// A. This is used add and remove our subscription for this key to remote servers
//
// 2. For a given key and updating address, find all connected clients that care
// A. First find connected clients that have subscribed to this key, regardless of address
// B. Then find connected clients that have subscribed to this key for this particular address
//
// 3. Find all remote servers that have subscribed to a given key
// A. This is so when the key is updated or deleted, we know who to send it to
//
// 4. For a given client (such as on disconnect), remove all records of their subscriptions

#endif // _RAKNET_SUPPORT_*