platform/include/pion/platform/DatabaseInserter.hpp

// ------------------------------------------------------------------------
// Pion is a development platform for building Reactors that process Events
// ------------------------------------------------------------------------
// Copyright (C) 2007-2009 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_DATABASINSERTER_HEADER__
#define __PION_DATABASINSERTER_HEADER__

#include <vector>
#include <boost/scoped_ptr.hpp>
#include <boost/thread/thread.hpp>
#include <boost/thread/condition.hpp>
#include <pion/PionConfig.hpp>
#include <pion/PionException.hpp>
#include <pion/PionLogger.hpp>
#include <pion/platform/Event.hpp>
#include <pion/platform/RuleChain.hpp>
#include <pion/platform/Query.hpp>
#include <pion/platform/Database.hpp>
#include <pion/platform/DatabaseManager.hpp>


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


class PION_PLATFORM_API DatabaseInserter
{
public:

    class MissingDatabaseManagerException : public std::exception {
    public:
        virtual const char* what() const throw() {
            return "DatabaseInserter is missing the DatabaseManager";
        }
    };

    class EmptyDatabaseException : public std::exception {
    public:
        virtual const char* what() const throw() {
            return "DatabaseInserter configuration is missing a required Database parameter";
        }
    };

    class EmptyTableException : public std::exception {
    public:
        virtual const char* what() const throw() {
            return "DatabaseInserter configuration is missing a required Table parameter";
        }
    };

    class NoFieldsException : public std::exception {
    public:
        virtual const char* what() const throw() {
            return "DatabaseInserter configuration must contain at least one field mapping";
        }
    };

    class EmptyFieldException : public std::exception {
    public:
        virtual const char* what() const throw() {
            return "DatabaseInserter configuration includes an empty field name";
        }
    };

    class EmptyTermException : public std::exception {
    public:
        virtual const char* what() const throw() {
            return "DatabaseInserter configuration is missing a term identifier";
        }
    };

    class UnknownTermException : public PionException {
    public:
        UnknownTermException(const std::string& term_id)
            : PionException("DatabaseInserter configuration maps field to an unknown term: ", term_id) {}
    };

    class IllegalCharactersException: public PionException {
    public:
        IllegalCharactersException(const std::string& field_name)
            : PionException("DatabaseInserter configuration has a field name with illegal characters: ", field_name) {}
    };

    class DuplicateColumnName: public PionException {
    public:
        DuplicateColumnName(const std::string& field_name)
            : PionException("DatabaseInserter configuration has a duplicate field name: ", field_name) {}
    };

    class NoUniqueKeyFound: public PionException {
    public:
        NoUniqueKeyFound(void)
            : PionException("DatabaseInserter configuration has MaxAge, but there is no Unique indexed column") {}
    };

    class MissingEventTime: public PionException {
    public:
        MissingEventTime(const std::string& element )
            : PionException("DatabaseInserter configuration has MaxAge, but the age term is missing: " + element) {}
    };

    DatabaseInserter(void) :
        m_logger(PION_GET_LOGGER("pion.platform.DatabaseInserter")),
        m_database_mgr_ptr(NULL),
        m_event_queue_ptr(new EventQueue), 
        m_queue_max(DEFAULT_QUEUE_SIZE), m_queue_timeout(DEFAULT_QUEUE_TIMEOUT),
        m_is_running(false), m_partition(0), m_wipe(false), m_max_age(0), m_last_time(0), m_table_size(0)
    {}

    virtual ~DatabaseInserter() { stop(); }

    inline void setDatabaseManager(DatabaseManager& mgr) { m_database_mgr_ptr = & mgr; }
    
    void setConfig(const Vocabulary& v, const xmlNodePtr config_ptr);

    void updateVocabulary(const Vocabulary& v);

    void updateDatabases(void);

    void insert(const EventPtr& e);

    void start(void);

    void stop(void);

    std::size_t getEventsQueued(void) const;

    std::size_t getKeyCacheSize(void) const;

    const std::string& getDatabaseId(void) const { return m_database_id; }

    DatabasePtr getDatabasePtr(void) { return m_database_ptr; }

    const std::string& getTableName(void) const { return m_table_name; }

    bool setTableName(const std::string& name)
    {
        if (m_is_running) return false;
        m_table_name = name;
        return true;
    }

    bool setPartition(unsigned partition)
    {
        if (m_is_running) return false;
        m_partition = partition;
        return true;
    }

    bool setWipe(bool wipe)
    {
        if (m_is_running) return false;
        m_wipe = wipe;
        return true;
    }

    boost::uint32_t getRotate(void) const
    {
        return m_cache_size ? (m_cache_consumption * 100UL / m_cache_size) : 0;
    }

    boost::uint64_t getTableSize(void) const { return m_table_size; }

    Query::FieldMap getFieldMap(void) const { return m_field_map; }

    Query::IndexMap getIndexMap(void) const { return m_index_map; }

    inline bool isRunning(void) const { return m_is_running; }
    
    inline void setLogger(PionLogger log_ptr) { m_logger = log_ptr; }

    inline PionLogger getLogger(void) { return m_logger; }

    inline bool tableExists(void)
    {
        if (!m_database_ptr)
            m_database_ptr = getDatabaseManager().getDatabase(m_database_id);
        return m_database_ptr->tableExists(m_table_name, m_partition);
    }


private:

    typedef std::vector<EventPtr>   EventQueue;


    DatabaseManager& getDatabaseManager(void);

    void insertEvents(void);

    bool checkEventQueue(boost::scoped_ptr<EventQueue>& insert_queue_ptr);


    static const boost::uint32_t            DEFAULT_QUEUE_SIZE;

    static const boost::uint32_t            DEFAULT_QUEUE_TIMEOUT;

    static const std::string                DATABASE_ELEMENT_NAME;

    static const std::string                TABLE_ELEMENT_NAME;

    static const std::string                FIELD_ELEMENT_NAME;

    static const std::string                QUEUE_SIZE_ELEMENT_NAME;

    static const std::string                QUEUE_TIMEOUT_ELEMENT_NAME;

    static const std::string                TERM_ATTRIBUTE_NAME;

    static const char *                     CHARSET_FOR_TABLES;

    static const std::string                INDEX_ATTRIBUTE_NAME;

    static const std::string                SQL_ATTRIBUTE_NAME;

    static const std::string                IGNORE_INSERT_ELEMENT_NAME;

    static const std::string                DEFAULT_IGNORE;

    static const std::string                MAX_KEY_AGE_ELEMENT_NAME;

    static const std::string                EVENT_AGE_ELEMENT_NAME;

    static const boost::uint32_t            DEFAULT_MAX_AGE;

    PionLogger                              m_logger;

    DatabaseManager *                       m_database_mgr_ptr;

    std::string                             m_database_id;

    std::string                             m_table_name;

    Query::FieldMap                         m_field_map;

    Query::IndexMap                         m_index_map;

    DatabasePtr                             m_database_ptr;
    
    QueryPtr                                m_insert_query_ptr;

    QueryPtr                                m_begin_transaction_ptr;

    QueryPtr                                m_commit_transaction_ptr;

    boost::scoped_ptr<EventQueue>           m_event_queue_ptr;

    boost::uint32_t                         m_queue_max;

    boost::uint32_t                         m_queue_timeout;

    mutable boost::mutex                    m_queue_mutex;

    boost::condition                        m_wakeup_worker;

    boost::condition                        m_swapped_queue;

    boost::scoped_ptr<boost::thread>        m_thread;

    bool                                    m_ignore_insert;
    
    volatile bool                           m_is_running;

    unsigned                                m_partition;

    bool                                    m_wipe;

    pion::platform::RuleChain               m_rules;

    typedef PION_HASH_MAP<pion::platform::Event::BlobType, boost::uint32_t, PION_HASH(pion::platform::Event::BlobType)> KeyHash;
    KeyHash                                 m_keys;

    Vocabulary::TermRef                     m_key_term_ref;

    boost::uint32_t                         m_max_age;

    Vocabulary::TermRef                     m_timestamp_term_ref;

    boost::uint32_t                         m_last_time;

    boost::uint64_t                         m_cache_consumption;

    boost::uint32_t                         m_cache_overhead;

    std::vector<Vocabulary::Term>           m_cache_terms;

    boost::uint64_t                         m_cache_rows;

    boost::uint64_t                         m_cache_size;

    boost::uint64_t                         m_table_size;
};


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

#endif

---