platform/include/pion/platform/ReactionEngine.hpp

// ------------------------------------------------------------------------
// Pion is a development platform for building Reactors that process Events
// ------------------------------------------------------------------------
// Copyright (C) 2007-2008 Atomic Labs, Inc.  (http://www.atomiclabs.com)
//
// Pion is free software: you can redistribute it and/or modify it under the
// terms of the GNU Affero General Public License as published by the Free
// Software Foundation, either version 3 of the License, or (at your option)
// any later version.
//
// Pion is distributed in the hope that it will be useful, but WITHOUT ANY
// WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
// FOR A PARTICULAR PURPOSE.  See the GNU Affero General Public License for
// more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with Pion.  If not, see <http://www.gnu.org/licenses/>.
//

#ifndef __PION_REACTIONENGINE_HEADER__
#define __PION_REACTIONENGINE_HEADER__

#include <string>
#include <libxml/tree.h>
#include <boost/bind.hpp>
#include <boost/function/function0.hpp>
#include <pion/PionConfig.hpp>
#include <pion/PionException.hpp>
#include <pion/platform/Event.hpp>
#include <pion/platform/Reactor.hpp>
#include <pion/platform/PluginConfig.hpp>
#include <pion/platform/ReactionScheduler.hpp>

namespace pion {        // begin namespace pion
namespace platform {    // begin namespace platform (Pion Platform Library)

// forward declarations
class VocabularyManager;
class CodecFactory;
class ProtocolFactory;
class DatabaseManager;
    
class PION_PLATFORM_API ReactionEngine :
    public PluginConfig<Reactor>
{
public:

    class ReactorNotFoundException : public PionException {
    public:
        ReactorNotFoundException(const std::string& reactor_id)
            : PionException("No reactors found for identifier: ", reactor_id) {}
    };
    
    class ConnectionNotFoundException : public PionException {
    public:
        ConnectionNotFoundException(const std::string& connection_id)
            : PionException("No connections found for identifier: ", connection_id) {}
    };

    class WorkspaceNotFoundException : public PionException {
    public:
        WorkspaceNotFoundException(const std::string& workspace_id)
            : PionException("No Workspace found for identifier: ", workspace_id) {}
    };

    class RemoveNonEmptyWorkspaceException : public PionException {
    public:
        RemoveNonEmptyWorkspaceException(const std::string& workspace_id)
            : PionException("The Workspace specified for removal was not empty: ", workspace_id) {}
    };

    class EmptyConnectionIdException : public PionException {
    public:
        EmptyConnectionIdException(const std::string& config_file)
            : PionException("Configuration file includes a connection with an empty identifier: ", config_file) {}
    };

    class BadConnectionTypeException : public PionException {
    public:
        BadConnectionTypeException(const std::string& connection_id)
            : PionException("Bad connection type in configuration file: ", connection_id) {}
    };
    
    class EmptyFromException : public PionException {
    public:
        EmptyFromException(const std::string& connection_id)
            : PionException("Reactor configuration has a connection with empty From element: ", connection_id) {}
    };
    
    class EmptyToException : public PionException {
    public:
        EmptyToException(const std::string& connection_id)
            : PionException("Reactor configuration has a connection with empty To element: ", connection_id) {}
    };
    
    class AddConnectionConfigException : public PionException {
    public:
        AddConnectionConfigException(const std::string& connection)
            : PionException("Unable to add a Connection to the Reactor configuration file: ", connection) {}
    };
    
    class RemoveConnectionConfigException : public PionException {
    public:
        RemoveConnectionConfigException(const std::string& connection)
            : PionException("Unable to remove a Connection from the Reactor configuration file: ", connection) {}
    };

    class BadConnectionConfigException : public std::exception {
    public:
        virtual const char* what() const throw() {
            return "New Reactor connection configuration is invalid";
        }
    };

    class BadWorkspaceConfigException : public std::exception {
    public:
        virtual const char* what() const throw() {
            return "New Reactor Workspace configuration is invalid";
        }
    };

    class AddWorkspaceConfigException : public PionException {
    public:
        AddWorkspaceConfigException()
            : PionException("Unable to add a Workspace to the Reactor configuration file") {}
    };

    class SetWorkspaceConfigException : public PionException {
    public:
        SetWorkspaceConfigException()
            : PionException("Error setting the configuration for a Workspace") {}
    };

    class UpdateConfigOptionException : public PionException {
    public:
        UpdateConfigOptionException(const std::string& reactor_id)
            : PionException("updateConfigOption failed for Reactor with identifier ", reactor_id) {}
    };


    ReactionEngine(VocabularyManager& vocab_mgr,
                   CodecFactory& codec_factory,
                   ProtocolFactory& protocol_factory,
                   DatabaseManager& database_mgr);
    
    virtual ~ReactionEngine() { shutdown();  }
    
    virtual void openConfigFile(void);

    void clearReactorStats(const std::string& reactor_id);

    void start(void);
    
    void stop(void);
    
    void shutdown(void);

    void clearStats(void);

    void updateCodecs(void);

    void updateDatabases(void);

    void updateProtocols(void);

    void restartReactorsThatShouldBeRunning(void);
    
    void startReactor(const std::string& reactor_id);

    void stopReactor(const std::string& reactor_id);

    void setReactorConfig(const std::string& reactor_id,
                          const xmlNodePtr config_ptr);
    
    void setReactorLocation(const std::string& reactor_id,
                            const xmlNodePtr config_ptr);

    std::string addReactor(const xmlNodePtr config_ptr);
    
    void removeReactor(const std::string& reactor_id);
    
    Reactor *addTempConnectionIn(const std::string& reactor_id, 
                                 const std::string& connection_id,
                                 const std::string& connection_info,
                                 boost::function0<void> removed_handler);
    
    void addTempConnectionOut(const std::string& reactor_id, 
                              const std::string& connection_id,
                              const std::string& connection_info,
                              Reactor::EventHandler connection_handler);
    
    void removeTempConnection(const std::string& connection_id);

    std::string addReactorConnection(const std::string& from_id, const std::string& to_id);
    
    std::string addReactorConnection(const xmlNodePtr config_ptr);
    
    void removeReactorConnection(const std::string& from_id, const std::string& to_id);
    
    void removeReactorConnection(const std::string& connection_id);

    std::string addWorkspace(const char* content_buf, std::size_t content_length);

    void removeWorkspace(const std::string& workspace_id);

    void removeReactorsFromWorkspace(const std::string& workspace_id);

    void setWorkspaceConfig(const std::string& workspace_id, const char* content_buf, std::size_t content_length);

    void writeStatsXML(std::ostream& out, const std::string& reactor_id = "", const bool details = false);

    void writeConnectionsXML(std::ostream& out, const std::string& only_id) const;

    inline void writeConnectionsXML(std::ostream& out) const {
        std::string empty_only_id;
        writeConnectionsXML(out, empty_only_id);
    }

    bool writeWorkspaceXML(std::ostream& out, const std::string& workspace_id) const;

    void writeWorkspacesXML(std::ostream& out) const;

    bool writeWorkspaceLimitedConfigXML(std::ostream& out, const std::string& workspace_id) const;

    bool hasWorkspace(const std::string& workspace_id) const;

    bool creationAllowed(xmlNodePtr permission_config_ptr, xmlNodePtr config_ptr) const;

    bool updateAllowed(xmlNodePtr permission_config_ptr, const std::string& id, xmlNodePtr config_ptr) const;

    bool removalAllowed(xmlNodePtr permission_config_ptr, const std::string& id) const;

    bool accessAllowed(xmlNodePtr permission_config_ptr, const std::string& reactor_id) const;

    std::string getPermissionType(void) const { return REACTORS_PERMISSION_TYPE; }

    static xmlNodePtr createReactorConfig(const char *buf, std::size_t len) {
        return ConfigManager::createResourceConfig(Reactor::REACTOR_ELEMENT_NAME, buf, len);
    }

    static xmlNodePtr createConnectionConfig(const char *buf, std::size_t len) {
        return ConfigManager::createResourceConfig(CONNECTION_ELEMENT_NAME, buf, len);
    }

    inline void send(const std::string& reactor_id, EventPtr& e) {
        Reactor *reactor_ptr = m_plugins.get(reactor_id);
        if (reactor_ptr == NULL)
            throw ReactorNotFoundException(reactor_id);
        m_scheduler.post(boost::bind<void>(boost::ref(*reactor_ptr), e));
    }

    template <typename F>
    inline boost::signals::connection subscribe(const std::string& reactor_id, const std::string& signal_id, F f) {
        Reactor *reactor_ptr = m_plugins.get(reactor_id);
        if (reactor_ptr == NULL)
            throw ReactorNotFoundException(reactor_id);
        return reactor_ptr->subscribe(signal_id, f);
    }

    inline void query(const std::string& reactor_id, std::ostream& out,
        const Reactor::QueryBranches& branches, const Reactor::QueryParams& qp)
    {
        if (branches.size() < 3) {
            // query request for /query/<reactor_id>
            writeStatsXML(out, reactor_id, true);
        } else {
            Reactor *reactor_ptr = m_plugins.get(reactor_id);
            if (reactor_ptr == NULL)
                throw ReactorNotFoundException(reactor_id);
            reactor_ptr->query(out, branches, qp);
        }
    }

    inline boost::uint64_t getTotalOperations(void) const {
        return m_plugins.getStatistic(boost::bind(&Reactor::getEventsIn, _1));
    }
    
    inline boost::uint64_t getEventsIn(const std::string& reactor_id) const {
        return m_plugins.getStatistic(reactor_id, boost::bind(&Reactor::getEventsIn, _1));
    }
    
    inline boost::uint64_t getEventsOut(const std::string& reactor_id) const {
        return m_plugins.getStatistic(reactor_id, boost::bind(&Reactor::getEventsOut, _1));
    }
    
    inline std::size_t getEventsQueued(void) const { return m_scheduler.getQueueSize(); }

    inline bool isRunning(const std::string& reactor_id) const {
        const Reactor *reactor_ptr = m_plugins.get(reactor_id);
        if (reactor_ptr == NULL)
            throw ReactorNotFoundException(reactor_id);
        return reactor_ptr->isRunning();
    }

    template<typename WorkFunction>
    inline void post(WorkFunction work_func) { m_scheduler.post(work_func); }
    
    inline boost::uint32_t getNumThreads(void) const { return m_scheduler.getNumThreads(); }
    
    inline void setNumThreads(const boost::uint32_t n) { m_scheduler.setNumThreads(n); }
    
    inline bool getMultithreadBranches(void) const { return m_multithread_branches; }
    
    inline void setMultithreadBranches(bool b) { m_multithread_branches = b; }

    inline bool isRunning(void) const { return m_is_running; }  


private:
    
    struct TempConnection {
        TempConnection(bool output_connection,
                       const std::string& reactor_id,
                       const std::string& connection_id,
                       const std::string& connection_info,
                       boost::function0<void> removed_handler)
            : m_output_connection(output_connection), m_reactor_id(reactor_id),
            m_connection_id(connection_id), m_connection_info(connection_info),
            m_removed_handler(removed_handler)
        {}
        
        ~TempConnection() {}
        
        const bool                      m_output_connection;
        const std::string               m_reactor_id;
        const std::string               m_connection_id;
        const std::string               m_connection_info;
        boost::function0<void>          m_removed_handler;
    };
    
    typedef std::list<TempConnection>   TempConnectionList;
    
    struct ReactorConnection {
        ReactorConnection(const std::string& connection_id,
                          const std::string& from_id,
                          const std::string& to_id)
            : m_connection_id(connection_id), m_from_id(from_id), m_to_id(to_id)
        {}
        
        ~ReactorConnection() {}
        
        const std::string               m_connection_id;
        const std::string               m_from_id;
        const std::string               m_to_id;
    };
    
    typedef std::list<ReactorConnection>    ReactorConnectionList;

    
    virtual void addPluginNoLock(const std::string& plugin_id,
                                 const std::string& plugin_name,
                                 const xmlNodePtr config_ptr)
    {
        try {
            Reactor *reactor_ptr = m_plugins.load(plugin_id, plugin_name);
            reactor_ptr->setId(plugin_id);
            reactor_ptr->setScheduler(m_scheduler);
            reactor_ptr->setMultithreadBranches(m_multithread_branches);
            reactor_ptr->setCodecFactory(m_codec_factory);
            reactor_ptr->setProtocolFactory(m_protocol_factory);
            reactor_ptr->setDatabaseManager(m_database_mgr);
            reactor_ptr->setReactionEngine(*this);
            if (config_ptr != NULL) {
                VocabularyPtr vocab_ptr(m_vocab_mgr.getVocabulary());
                reactor_ptr->setConfig(*vocab_ptr, config_ptr);
            }
            try {
                reactor_ptr->startOutRunning(config_ptr, true);
            } catch (std::exception& e) {
                // log but don't propagate exceptions from startOutRunning()
                PION_LOG_ERROR(m_logger, e.what());
            }
        } catch (PionPlugin::PluginNotFoundException&) {
            throw;
        } catch (std::exception& e) {
            throw PluginException(e.what());
        }
    }

    static inline std::string getConnectionAsText(const std::string& from_id,
                                                  const std::string& to_id)
    {
        std::string result(from_id);
        result += " -> ";
        result += to_id;
        return result;
    }
    
    void addConnectionNoLock(const std::string& connection_id,
                             const std::string& from_id,
                             const std::string& to_id);
    
    void removeConnectionNoLock(const std::string& reactor_id,
                                const std::string& connection_id);

    void removeConnectionConfigNoLock(const std::string& from_id,
                                      const std::string& to_id);
    
    void stopNoLock(void);

    void setWorkspaceConfig(xmlNodePtr workspace_node_ptr, const char* content_buf, std::size_t content_length);

    
    static const boost::uint32_t    DEFAULT_NUM_THREADS;
    
    static const std::string        DEFAULT_CONFIG_FILE;

    static const std::string        CONNECTION_ELEMENT_NAME;
    
    static const std::string        TYPE_ELEMENT_NAME;
    
    static const std::string        FROM_ELEMENT_NAME;
    
    static const std::string        TO_ELEMENT_NAME;
    
    static const std::string        TOTAL_OPS_ELEMENT_NAME;
    
    static const std::string        EVENTS_QUEUED_ELEMENT_NAME;

    static const std::string        CONNECTION_TYPE_REACTOR;

    static const std::string        CONNECTION_TYPE_INPUT;

    static const std::string        CONNECTION_TYPE_OUTPUT;

    static const std::string        REACTORS_PERMISSION_TYPE;

    static const std::string        UNRESTRICTED_ELEMENT_NAME;

    static const std::string        WORKSPACE_QUALIFIER_ELEMENT_NAME;

    ReactionScheduler               m_scheduler;

    CodecFactory &                  m_codec_factory;

    ProtocolFactory &               m_protocol_factory;

    DatabaseManager &               m_database_mgr;
    
    TempConnectionList              m_temp_connections;
    
    ReactorConnectionList           m_reactor_connections;

    boost::signals::scoped_connection   m_codec_connection;
    
    boost::signals::scoped_connection   m_db_connection;

    boost::signals::scoped_connection   m_protocol_connection;

    bool                            m_is_running;

    bool                            m_multithread_branches;
};


}   // end namespace platform
}   // end namespace pion

#endif

---