NPLMiniRuntime.hpp

#pragma once
//-----------------------------------------------------------------------------
// Class:   NPLMiniRuntime
// Authors: LiXizhi
// Company: ParaEngine Co.
// Date:    2010.4.8
// Desc: NPLMiniRuntime is used by any standalone exe/dll to act as a NPL runtime 
// without the need to link with the full-featured ParaEngine/NPL library. 
// The mini runtime provides basic implementation to send/receive NPL pure data messages. 
// Therefore, it is easy to build loosely coupled systems using NPL for messaging with minimum code overhead and dependency. 
// To distinguish it from the real NPL runtime, we have put it in the NPLInterface namespace
// 
// See comment of CNPLMiniRuntimeT for example usage
//-----------------------------------------------------------------------------

// #define USE_BOOST_SIGNAL_FILE_HANDLER

#include <boost/intrusive_ptr.hpp>
#include <boost/shared_ptr.hpp>
#include "PEtypes.h"
#include "NPLTypes.h"
#include "INPL.h"
#include "INPLRuntime.h"
#include "NPLInterface.hpp"

#include "util/Mutex.hpp"
#include "util/Semaphore.hpp"

#include <vector>
#include <map>
#include <set>
#include <queue>

#ifdef USE_BOOST_SIGNAL_FILE_HANDLER
#include <boost/signals2.hpp>
#else
#include <boost/function.hpp>
#endif

namespace NPLInterface
{
#pragma region Headers

    struct NPLMiniMessage
    {
        NPLMiniMessage(){};
        NPLMiniMessage(const char * sNPLFilename, const char* sCode, int nCodeLength=0)
            : m_type(0)
        {
            if(sNPLFilename)
                m_sFilename = sNPLFilename;
            if(sCode)
            {
                if(nCodeLength <= 0)
                    m_sCode = sCode;
                else
                {
                    m_sCode.reserve(nCodeLength);
                    m_sCode = sCode;
                }
            }
        }
        std::string m_sFilename;

        std::string m_sCode;

        int m_type;
    };

    template <class StringType = std::string>
    struct NPLFileNameT
    {
        StringType sRuntimeStateName;
        StringType sNID;
        StringType sRelativePath;
        StringType sDNSServerName;

    public:
        NPLFileNameT(){}

        NPLFileNameT(const char * filename){
            FromString(filename);
        }

        void SetRelativePath(const char* sPath, int nCount=-1){
            if(nCount <= 0)
            {
                sRelativePath = sPath;
                nCount = (int)sRelativePath.size();
            }
            else
            {
                sRelativePath.assign(sPath, nCount);
            }
            for(int i=0; i<nCount ;i++)
            {
                if(sRelativePath[i] == '\\')
                    sRelativePath[i] = '/';
            }
        }

    public:
        std::string ToString(){
            string output;
            ToString(output);
            return output;
        }
        void ToString(std::string & output){
            output.clear();
            if(!sRuntimeStateName.empty())
            {
                output.append("(");
                output.append(sRuntimeStateName);
                output.append(")");
            }

            // the activation type
            if( ! sNID.empty() )
                output.append(sNID);
            if( ! sRelativePath.empty() )
                output.append(sRelativePath);
            if( ! sDNSServerName.empty() )
            {
                output.append("@");
                output.append(sDNSServerName);
            }
        }

        void FromString(const char* sFilePath){
            // for empty string, default to local Glia file.
            if(sFilePath[0] == '\0')
            {
                // FromString("(gl)script/empty.lua");
                sRuntimeStateName.clear();
                sNID.clear();
                sDNSServerName.clear();
                sRelativePath.clear();
                return;
            }

            int i = 0;
            int nNIDIndex = 0;
            int nRelativePathIndex;
            int nDNSServerIndex;

            bool bExplicitActivationType = false;
            if(sFilePath[i] == '(')
            {
                bExplicitActivationType = true;
                i++;
                while( (sFilePath[i]!=')') && (sFilePath[i]!='\0'))
                {
                    i++;
                }
                i++;
                if( !(i==4 && (sFilePath[1]=='g') && (sFilePath[2]=='l')) )
                    sRuntimeStateName.assign(sFilePath+1, i-2);
                else
                    sRuntimeStateName.clear();
                nNIDIndex = i;
            }

            while( (sFilePath[i]!=':') && (sFilePath[i]!='\0'))
                i++;

            if(sFilePath[i]=='\0')
            {
                sNID.clear();
                sRelativePath.assign(sFilePath+nNIDIndex, i-nNIDIndex);
                sDNSServerName.clear();
                return;
            }
            else 
            {
                if(i>nNIDIndex)
                    sNID.assign(sFilePath+nNIDIndex, i-nNIDIndex);
                else
                    sNID.clear();

                nRelativePathIndex = i+1;
                i++;
            }

            while( (sFilePath[i]!='@') && (sFilePath[i]!='\0') )
                i++;
            SetRelativePath(sFilePath+nRelativePathIndex, i - nRelativePathIndex);

            if(sFilePath[i]=='\0')
            {
                sDNSServerName.clear();
                return;
            }
            else 
                i++;

            nDNSServerIndex = i;

            while( sFilePath[i]!='\0')
                i++;
            sDNSServerName.assign(sFilePath+nDNSServerIndex, i - nDNSServerIndex);
        }
    };
    typedef NPLFileNameT<> NPLFileName;

#pragma endregion Headers

#pragma region CNPLMiniState

    class CNPLMiniState : public NPL::INPLRuntimeState,
        public boost::enable_shared_from_this<CNPLMiniState>,
        private boost::noncopyable
    {
    public:
#ifdef USE_BOOST_SIGNAL_FILE_HANDLER
        typedef boost::signals2::signal< void(int, void*) >  NPLFileActivateHandler_t;
        typedef NPLFileActivateHandler_t::slot_type NPLFileActivateHandlerCallback_t;
        //typedef boost::function< void(int, void*) > NPLFileActivateHandlerCallback_t;
        typedef std::map<std::string, boost::shared_ptr<typename NPLFileActivateHandler_t> > NPLFileHandlerMap_t;
#else
        typedef boost::function< void(int, void*) >  NPLFileActivateHandler_t;
        typedef boost::function< void(int, void*) > NPLFileActivateHandlerCallback_t;
        typedef std::map<std::string, typename NPLFileActivateHandler_t> NPLFileHandlerMap_t;
#endif
        typedef std::queue<NPLMiniMessage> NPLMiniMessageQueue_t;
        

        CNPLMiniState(const char* name=NULL): m_current_msg_length(0), m_current_msg(NULL),m_pNPLRuntime(NULL), m_processed_msg_count(0) {
            if(name != NULL){
                m_name = name;
            }
        };
        virtual ~CNPLMiniState() {};

        void Init(){}

        class CCurrentMessage
        {
        public:
            CCurrentMessage(CNPLMiniState* pState, const char* msg, int nLength):m_pState(pState){
                if(m_pState)
                    m_pState->SetCurrentMessage(msg,nLength);
            }
            ~CCurrentMessage(){
                if(m_pState)
                    m_pState->SetCurrentMessage(NULL, 0);
            }
            CNPLMiniState* m_pState;
        };

    public:
        virtual const std::string& GetName() const {return m_name;};

        virtual int activate(const char * sNPLFilename, const char* sCode, int nCodeLength=0, int priority=2, int reliability=4){
            // Note: if one wants to process in the calling thread, just inherit and do following. 
            // NPLInterface::NPLObjectProxy tabMsg = NPLInterface::NPLHelper::MsgStringToNPLTable(sCode, nCodeLength);
            // OnMessageCallback(tabMsg);

            ParaEngine::Mutex::ScopedLock lock_(m_mutex);
            m_input_queue.push(NPLMiniMessage(sNPLFilename, sCode, nCodeLength));
            return NPL::NPL_OK;
        }

        virtual NPL::NPLReturnCode Activate_async(const std::string & filepath, const char * code = NULL,int nLength=0, int priority=0) {
            return (NPL::NPLReturnCode)activate(filepath.c_str(), code, nLength);
        }

        virtual NPL::NPLReturnCode ActivateLocal(const char* filepath, const char * code = NULL,int nLength=0, int priority=0) {
            return (NPL::NPLReturnCode)activate(filepath, code, nLength);
        }

        virtual NPL::NPLReturnCode Activate_async(NPL::NPLMessage_ptr& msg, int priority=0) {return NPL::NPL_OK;}

        virtual NPL::NPLReturnCode SendMessage(NPL::NPLMessage_ptr& msg, int priority=0) {return NPL::NPL_OK;}

        virtual const char* GetCurrentMsg(){return m_current_msg;}

        virtual int GetCurrentMsgLength(){return m_current_msg_length;}

        virtual NPL::INPLRuntime* GetNPLRuntime() {return m_pNPLRuntime;}

        virtual void WriteLog(const char* text, int nTextLen=0, int nLogType = 0) {}

        //
        // Timer functions
        //

        virtual bool SetTimer(int nIDEvent, float fElapse, const char* sNeuronFile) {return false;};

        virtual bool KillTimer(int nIDEvent) { return false;};

        virtual bool ChangeTimer(int nIDEvent, int dueTime, int period) { return false;};

        virtual void RegisterFile(const char* sFilename, NPL::INPLActivationFile* pFileHandler = NULL) {};

        virtual void call(const char * sNPLFilename, const char* sCode, int nCodeLength = 0){};
    public:
        virtual int Process()
        {
            int nCount = 0;
            // process as many as possible. 
            ParaEngine::Mutex::ScopedLock lock_(m_mutex);
            while(!m_input_queue.empty())
            {
                ProcessMsg(m_input_queue.front());
                m_input_queue.pop();
                ++nCount;
            }
            return nCount;
        }

        bool RegisterFileHandler(const char* filename, const NPLFileActivateHandlerCallback_t& fileCallback)
        {
            if(filename == 0)
                return false;
            std::string sFileName = filename;

            
#ifdef USE_BOOST_SIGNAL_FILE_HANDLER
            NPLFileHandlerMap_t::iterator iter = m_file_handlers_map.find(sFileName);
            if( iter == m_file_handlers_map.end() )
            {
                m_file_handlers_map[sFileName] = boost::shared_ptr<typename NPLFileActivateHandler_t>(new NPLFileActivateHandler_t());
            }
            m_file_handlers_map[sFileName]->connect(fileCallback);
#else
            m_file_handlers_map[sFileName] = fileCallback;
#endif
            return true;
        }

    protected:
        void SetCurrentMessage(const char* msg, int nLength)
        {
            m_current_msg = msg;
            m_current_msg_length = nLength;
        }

        int ProcessMsg( const NPLMiniMessage& msg )
        {
            ++ m_processed_msg_count;

            if(msg.m_type == 0)
            {
                NPLFileHandlerMap_t::iterator iter = m_file_handlers_map.find(msg.m_sFilename);
                if( iter!= m_file_handlers_map.end() )
                {
                    CCurrentMessage  push_msg(this, msg.m_sCode.c_str(), (int)(msg.m_sCode.size()));
#ifdef USE_BOOST_SIGNAL_FILE_HANDLER
                    iter->second->operator ()(ParaEngine::PluginActType_STATE, this);
#else
                    iter->second(ParaEngine::PluginActType_STATE, this);
#endif
                }
            }
            return 0;
        }

    protected:

        const char* m_current_msg;

        int m_current_msg_length;

        std::string m_name;

        ParaEngine::Mutex m_mutex;
        
        NPL::INPLRuntime * m_pNPLRuntime;

        NPLMiniMessageQueue_t m_input_queue;

        NPLFileHandlerMap_t m_file_handlers_map;

        int m_processed_msg_count;
    };

#pragma endregion CNPLMiniState

#pragma region CNPLMiniRuntime

    template <class NPL_STATE = CNPLMiniState>
    class CNPLMiniRuntimeT : public NPL::INPLRuntime
    {
    public:
        typedef boost::shared_ptr<typename NPL_STATE> NPLRuntimeState_ptr;
        typedef std::vector<NPLRuntimeState_ptr>    NPLRuntime_Temp_Pool_Type;

        struct NPLRuntimeState_PtrOps
        {
            bool operator()( const NPLRuntimeState_ptr & a, const NPLRuntimeState_ptr & b ) const
            { return a.get() < b.get(); }
        };
        typedef std::set <NPLRuntimeState_ptr, NPLRuntimeState_PtrOps>  NPLRuntime_Pool_Type;

        CNPLMiniRuntimeT(){
            Init();
        }
        ~CNPLMiniRuntimeT(){
            Cleanup();
        }
    public:

        virtual void Init(){
            if(m_runtime_state_main.get() == 0)
            {
                // SetCompressionKey(NULL, 0, 1);
                // the default "main" runtime state
                m_runtime_state_main = CreateRuntimeState("main", NPL::NPLRuntimeStateType_NPL);
            }
        };

        virtual void Run(bool bToEnd = true)
        {
            // the main runtime state is processed in the main game thread. 
            {
                // in case the structure is modified by other threads or during processing, we will first dump to a temp queue and then process from the queue. 
                ParaEngine::Mutex::ScopedLock lock_(m_mutex);
                NPLRuntime_Pool_Type::const_iterator iter, iter_end = m_runtime_states.end();
                for(iter = m_runtime_states.begin(); iter!=iter_end; ++iter)
                {
                    m_temp_rts_pool.push_back(*iter);
                }
            }

            NPLRuntime_Temp_Pool_Type::iterator itCur, itEnd = m_temp_rts_pool.end();
            for (itCur = m_temp_rts_pool.begin(); itCur != itEnd; ++itCur)
            {
                (*itCur)->Process();
            }
            m_temp_rts_pool.clear();
        };

        virtual void Cleanup(){
            m_runtime_state_main.reset();
            m_runtime_states.clear();
            m_active_state_map.clear();
        };

        virtual void SetHostMainStatesInFrameMove(bool bHostMainStatesInFrameMove){};

        virtual NPL::INPLRuntimeState* CreateState(const char* name, NPL::NPLRuntimeStateType type_=NPL::NPLRuntimeStateType_NPL){
            return CreateRuntimeState(name, type_).get();
        };

        virtual NPL::INPLRuntimeState* GetState(const char* name = NULL){
            if( name == NULL)
                return m_runtime_state_main.get();

            return GetRuntimeState(name).get();
        };

        virtual NPL::INPLRuntimeState* CreateGetState(const char* name, NPL::NPLRuntimeStateType type_=NPL::NPLRuntimeStateType_NPL){
            return CreateGetRuntimeState(name, type_).get();
        };

        virtual bool DeleteState(NPL::INPLRuntimeState* pRuntime_state){
            if(pRuntime_state)
            {
                return DeleteRuntimeState(((NPL_STATE*)pRuntime_state)->shared_from_this());
            }
            return false;
        };

        virtual NPL::INPLRuntimeState* GetMainState(){
            return m_runtime_state_main.get();
        };

        virtual bool AddToMainThread(NPL::INPLRuntimeState* runtime_state){return false;};

        virtual void SetUseCompression(bool bCompressIncoming, bool bCompressOutgoing){};

        virtual void SetCompressionKey(const byte* sKey=0, int nSize=0, int nUsePlainTextEncoding = 0){};

        virtual void SetCompressionLevel(int nLevel){};
        virtual int GetCompressionLevel(){return 0;};

        virtual void SetCompressionThreshold(int nThreshold){};
        virtual int GetCompressionThreshold(){return 0;};

        virtual void SetTCPKeepAlive(bool bEnable){};

        virtual bool IsTCPKeepAliveEnabled(){return false;};

        virtual void SetKeepAlive(bool bEnable){};
        virtual bool IsKeepAliveEnabled(){return false;};

        virtual void EnableIdleTimeout(bool bEnable){};
        virtual bool IsIdleTimeoutEnabled(){return false;};

        virtual void SetIdleTimeoutPeriod(int nMilliseconds){};
        virtual int GetIdleTimeoutPeriod(){return false;};

        virtual void StartNetServer(const char* server=NULL, const char* port=NULL){};

        virtual void StopNetServer(){};

        virtual void AddPublicFile(const std::string& filename, int nID){};

        virtual void ClearPublicFiles(){};

        virtual void GetIP(const char* nid, char* pOutput){};


        virtual void accept(const char* tid, const char* nid = NULL){};

        virtual void reject(const char* nid, int nReason = 0){};


        //
        // jabber client functions
        //

        virtual ParaEngine::INPLJabberClient* GetJabberClient(const char* sJID){return NULL;};
        virtual ParaEngine::INPLJabberClient* CreateJabberClient(const char* sJID){return NULL;};

        virtual bool CloseJabberClient(const char* sJID){return false;};

        //
        // new libcUrl interface. 
        //

        virtual bool AppendURLRequest(ParaEngine::CURLRequestTask* pUrlTask, const char* sPoolName = NULL){return false;};

        virtual bool ChangeRequestPoolSize(const char* sPoolName, int nCount){return false;};


        //
        // Downloader functions
        //

        virtual void AsyncDownload(const char* url, const char* destFolder, const char* callbackScript, const char* DownloaderName){};


        virtual void CancelDownload(const char* DownloaderName){};

        virtual int Download(const char* url, const char* destFolder, const char* callbackScript, const char* DownloaderName){return 0;};

        virtual void NPL_AddDNSRecord(const char * sDNSName, const char* sAddress){};

        virtual void NPL_SetDefaultChannel(int channel_ID){};
        virtual int NPL_GetDefaultChannel(){return 0;};

        virtual void NPL_SetChannelProperty(int channel_ID, int priority, int reliability){};
        virtual void NPL_ResetChannelProperties(){};

        virtual void NPL_GetChannelProperty(int channel_ID, int* priority, int* reliability){};

        //
        // Global activation functions
        //

        virtual int Activate(NPL::INPLRuntimeState* pRuntimeState, const char * sNeuronFile, const char * code = NULL,int nLength=0,  int channel=0, int priority=2, int reliability=3){
            if(sNeuronFile == NULL)
            {
                return (int)NPL::NPL_FailedToLoadFile;
            }

            NPLFileName FullName(sNeuronFile);

            // use Dispatcher to dispatch to a proper local runtime state or a remote one. 
            if(pRuntimeState == 0)
            {
                // default to main state. 
                return m_runtime_state_main->Activate_async(FullName.sRelativePath, code, nLength);
            }
            else 
            {
                if(FullName.sNID.empty())
                {
                    // local activation between local npl runtime state. 
                    if(!FullName.sRuntimeStateName.empty())
                    {
                        NPLRuntimeState_ptr rts = GetRuntimeState(FullName.sRuntimeStateName);
                        if(rts.get() != 0)
                        {
                            return rts->Activate_async(FullName.sRelativePath, code, nLength);
                        }
                        else
                        {
                            return -1;
                        }
                    }
                    else
                    {
                        return pRuntimeState->Activate_async(FullName.sRelativePath, code, nLength);
                    }
                }
                else
                {
                    return (int)NPL::NPL_Error;
                }
            }
        }
    public:
        NPLRuntimeState_ptr CreateRuntimeState(const std::string& name, NPL::NPLRuntimeStateType type_=NPL::NPLRuntimeStateType_NPL)
        {
            NPLRuntimeState_ptr runtimestate = GetRuntimeState(name);
            if(runtimestate.get() == 0)
            {
                runtimestate.reset(new NPL_STATE(name.c_str()));
                runtimestate->Init();
                ParaEngine::Mutex::ScopedLock lock_(m_mutex);
                m_runtime_states.insert(runtimestate);
                // assert(m_runtime_states.find(runtimestate)!= m_runtime_states.end());
                if(!name.empty())
                    m_active_state_map[name] = runtimestate;
            }
            return runtimestate;
        }

        NPLRuntimeState_ptr GetRuntimeState(const std::string& name)
        {
            if( name.empty() )
                return m_runtime_state_main;

            ParaEngine::Mutex::ScopedLock lock_(m_mutex);
            std::map<string, NPLRuntimeState_ptr>::iterator iter = m_active_state_map.find(name);

            if( iter != m_active_state_map.end())
            {
                return iter->second;
            }
            return NPLRuntimeState_ptr();
        }

        NPLRuntimeState_ptr CreateGetRuntimeState(const std::string& name, NPL::NPLRuntimeStateType type_=NPL::NPLRuntimeStateType_NPL)
        {
            NPLRuntimeState_ptr runtimestate =  GetRuntimeState(name);
            if(runtimestate.get() == 0)
            {
                // create the state and run it in the main thread. 
                runtimestate =  CreateRuntimeState(name, type_);
            }
            return runtimestate;
        }

        bool DeleteRuntimeState(NPLRuntimeState_ptr runtime_state)
        {
            if(runtime_state.get() == 0)
                return true;
            ParaEngine::Mutex::ScopedLock lock_(m_mutex);
            NPLRuntime_Pool_Type::iterator iter = m_runtime_states.find(runtime_state);
            if(iter != m_runtime_states.end())
            {
                m_runtime_states.erase(iter);
                return true;
            }
            if( ! runtime_state->GetName().empty() )
            {
                m_active_state_map.erase(runtime_state->GetName());
            }
            return false;
        }

        NPLRuntimeState_ptr GetMainRuntimeState(){
            return m_runtime_state_main;
        }
        
    protected:
        NPLRuntimeState_ptr m_runtime_state_main;

        NPLRuntime_Pool_Type m_runtime_states;

        NPLRuntime_Temp_Pool_Type m_temp_rts_pool;

        std::map<std::string, NPLRuntimeState_ptr> m_active_state_map;

        ParaEngine::Mutex m_mutex;
        ParaEngine::Semaphore m_semaphore;
    };
#pragma endregion CNPLMiniRuntime
}