daemon/src/core/SQLiteBundleStorage.h

Go to the documentation of this file.

/*
 * SQLiteBundleStorage.h
 *
 *  Created on: 09.01.2010
 *      Author: Myrtus
 */


#ifndef SQLITEBUNDLESTORAGE_H_
#define SQLITEBUNDLESTORAGE_H_

#include "BundleStorage.h"
#include "Component.h"
#include "EventReceiver.h"
#include "core/BundleStorage.h"
#include "core/EventReceiver.h"
#include <ibrdtn/data/MetaBundle.h>

#include <ibrcommon/thread/Thread.h>
#include <ibrcommon/thread/Conditional.h>
#include <ibrcommon/data/File.h>
#include <ibrcommon/thread/Queue.h>

#include <sqlite3.h>
#include <string>
#include <list>
#include <set>

//#define SQLITE_STORAGE_EXTENDED 1

class SQLiteBundleStorageTest;

namespace dtn
{
        namespace core
        {
                class SQLiteBundleStorage: public BundleStorage, public EventReceiver, public dtn::daemon::IndependentComponent, public ibrcommon::MutexInterface
                {
                        enum SQL_TABLES
                        {
                                SQL_TABLE_BUNDLE = 0,
                                SQL_TABLE_BLOCK = 1,
                                SQL_TABLE_ROUTING = 2,
                                SQL_TABLE_BUNDLE_ROUTING_INFO = 3,
                                SQL_TABLE_NODE_ROUTING_INFO = 4,
                                SQL_TABLE_END = 5
                        };

                        // enum of all possible statements
                        enum STORAGE_STMT
                        {
                                BUNDLE_GET_FILTER,
                                BUNDLE_GET_ID,
                                FRAGMENT_GET_ID,
                                BUNDLE_GET_FRAGMENT,

                                EXPIRE_BUNDLES,
                                EXPIRE_BUNDLE_FILENAMES,
                                EXPIRE_BUNDLE_DELETE,
                                EXPIRE_NEXT_TIMESTAMP,

                                EMPTY_CHECK,
                                COUNT_ENTRIES,

                                BUNDLE_DELETE,
                                FRAGMENT_DELETE,
                                BUNDLE_CLEAR,
                                BUNDLE_STORE,
                                BUNDLE_UPDATE_CUSTODIAN,
                                FRAGMENT_UPDATE_CUSTODIAN,

                                PROCFLAGS_SET,

                                BLOCK_GET_ID,
                                BLOCK_GET_ID_FRAGMENT,
                                BLOCK_GET,
                                BLOCK_GET_FRAGMENT,
                                BLOCK_CLEAR,
                                BLOCK_STORE,

#ifdef SQLITE_STORAGE_EXTENDED
                                ROUTING_GET,
                                ROUTING_REMOVE,
                                ROUTING_CLEAR,
                                ROUTING_STORE,

                                NODE_GET,
                                NODE_REMOVE,
                                NODE_CLEAR,
                                NODE_STORE,

                                INFO_GET,
                                INFO_REMOVE,
                                INFO_STORE,
#endif

                                VACUUM,
                                SQL_QUERIES_END
                        };

                        static const std::string _tables[SQL_TABLE_END];

                        // array of sql queries
                        static const std::string _sql_queries[SQL_QUERIES_END];

                        // array of the db structure as sql
                        static const std::string _db_structure[10];

                public:
                        friend class ::SQLiteBundleStorageTest;

                        class SQLBundleQuery
                        {
                        public:
                                SQLBundleQuery();
                                virtual ~SQLBundleQuery() = 0;

                                virtual const std::string getWhere() const = 0;

                                virtual size_t bind(sqlite3_stmt*, size_t offset) const
                                {
                                        return offset;
                                }

                                sqlite3_stmt *_statement;
                        };

                        class SQLiteQueryException : public ibrcommon::Exception
                        {
                        public:
                                SQLiteQueryException(string what = "Unable to execute Querry.") throw(): Exception(what)
                                {
                                }
                        };

                        SQLiteBundleStorage(const ibrcommon::File &path, const size_t &size);

                        virtual ~SQLiteBundleStorage();

                        void openDatabase(const ibrcommon::File &path);

                        void store(const dtn::data::Bundle &bundle);

                        dtn::data::Bundle get(const dtn::data::BundleID &id);

                        const std::list<dtn::data::MetaBundle> get(BundleFilterCallback &cb);

#ifdef SQLITE_STORAGE_EXTENDED

                        void storeBundleRoutingInfo(const data::BundleID &BundleID, const int &key, const string &routingInfo);

                        void storeNodeRoutingInfo(const data::EID &nodel, const int &key, const std::string &routingInfo);

                        void storeRoutingInfo(const int &key, const std::string &routingInfo);

                        std::string getBundleRoutingInfo(const data::BundleID &bundleID, const int &key);

                        std::string getNodeRoutingInfo(const data::EID &eid, const int &key);

                        std::string getRoutingInfo(const int &key);

                        void removeBundleRoutingInfo(const data::BundleID &bundleID, const int &key);

                        void removeNodeRoutingInfo(const data::EID &eid, const int &key);

                        void removeRoutingInfo(const int &key);
#endif

                        void remove(const dtn::data::BundleID &id);

                        void clear();

                        void clearAll();

                        bool empty();

                        unsigned int count();

                        void releaseCustody(const dtn::data::EID &custodian, const dtn::data::BundleID &id);

#ifdef SQLITE_STORAGE_EXTENDED

                        void setPriority(const int priority,const dtn::data::BundleID &id);

                        list<data::Block> getBlock(const data::BundleID &bundleID,const char &blocktype);

                        int occupiedSpace();
#endif

                        void raiseEvent(const Event *evt);

                        dtn::data::MetaBundle remove(const ibrcommon::BloomFilter&) { return dtn::data::MetaBundle(); };

                        virtual void trylock() throw (ibrcommon::MutexException);

                        virtual void enter() throw (ibrcommon::MutexException);

                        virtual void leave() throw (ibrcommon::MutexException);

                protected:
                        virtual void componentRun();
                        virtual void componentUp();
                        virtual void componentDown();
                        bool __cancellation();

                private:
                        enum Position
                        {
                                FIRST_FRAGMENT  = 0,
                                LAST_FRAGMENT   = 1,
                                BOTH_FRAGMENTS  = 2
                        };

                        class AutoResetLock
                        {
                        public:
                                AutoResetLock(ibrcommon::Mutex &mutex, sqlite3_stmt *st);
                                ~AutoResetLock();

                        private:
                                ibrcommon::MutexLock _lock;
                                sqlite3_stmt *_st;
                        };

                        class Task
                        {
                        public:
                                virtual ~Task() {};
                                virtual void run(SQLiteBundleStorage &storage) = 0;
                        };

                        class BlockingTask : public Task
                        {
                        public:
                                BlockingTask() : _done(false), _abort(false) {};
                                virtual ~BlockingTask() {};
                                virtual void run(SQLiteBundleStorage &storage) = 0;

                                void wait()
                                {
                                        ibrcommon::MutexLock l(_cond);
                                        while (!_done && !_abort) _cond.wait();

                                        if (_abort) throw ibrcommon::Exception("Task aborted");
                                }

                                void abort()
                                {
                                        ibrcommon::MutexLock l(_cond);
                                        _abort = true;
                                        _cond.signal(true);
                                }

                                void done()
                                {
                                        ibrcommon::MutexLock l(_cond);
                                        _done = true;
                                        _cond.signal(true);
                                }

                        private:
                                ibrcommon::Conditional _cond;
                                bool _done;
                                bool _abort;
                        };

                        class TaskRemove : public Task
                        {
                        public:
                                TaskRemove(const dtn::data::BundleID &id)
                                 : _id(id) { };

                                virtual ~TaskRemove() {};
                                virtual void run(SQLiteBundleStorage &storage);

                        private:
                                const dtn::data::BundleID _id;
                        };

                        class TaskIdle : public Task
                        {
                        public:
                                TaskIdle() { };

                                virtual ~TaskIdle() {};
                                virtual void run(SQLiteBundleStorage &storage);

                                static ibrcommon::Mutex _mutex;
                                static bool _idle;
                        };

                        class TaskExpire : public Task
                        {
                        public:
                                TaskExpire(size_t timestamp)
                                : _timestamp(timestamp) { };

                                virtual ~TaskExpire() {};
                                virtual void run(SQLiteBundleStorage &storage);

                        private:
                                size_t _timestamp;
                        };

                        dtn::data::MetaBundle get_meta(const dtn::data::BundleID &id);

                        void get(sqlite3_stmt *st, dtn::data::MetaBundle &bundle, size_t offset = 0);
                        void get(sqlite3_stmt *st, dtn::data::Bundle &bundle, size_t offset = 0);

#ifdef SQLITE_STORAGE_EXTENDED

                        void storeFragment(const dtn::data::Bundle &bundle);
#endif

                        sqlite3_stmt* prepare(const std::string &sqlquery);

                        int prepareBundle(list<std::string> &filenames, dtn::data::Bundle &bundle);

                        int store_blocks(const data::Bundle &Bundle);

                        void get_blocks(dtn::data::Bundle &bundle, const dtn::data::BundleID &id);

                        void check_consistency();

                        void check_fragments(set<string> &payloadFiles);

                        void check_bundles(set<string> &blockFiles);

                        void update_expire_time();

                        void update_custodian(const dtn::data::BundleID &b, const dtn::data::EID &custodian);

                        void new_expire_time(size_t ttl);
                        void reset_expire_time();
                        size_t get_expire_time();

                        void set_bundleid(sqlite3_stmt *st, const dtn::data::BundleID &id, size_t offset = 0) const;
                        void get_bundleid(sqlite3_stmt *st, dtn::data::BundleID &id, size_t offset = 0) const;

                        virtual const std::string getName() const;

                        ibrcommon::File dbPath;
                        ibrcommon::File dbFile;
                        ibrcommon::File _blockPath;

                        int dbSize;

                        // holds the database handle
                        sqlite3 *_database;

                        // contains all jobs to do
                        ibrcommon::Queue<Task*> _tasks;

                        ibrcommon::Mutex _next_expiration_lock;
                        size_t _next_expiration;

                        // array of statements
                        sqlite3_stmt* _statements[SQL_QUERIES_END];

                        // array of locks for each statement
                        ibrcommon::Mutex _locks[SQL_QUERIES_END];

                        void add_deletion(const dtn::data::BundleID &id);
                        void remove_deletion(const dtn::data::BundleID &id);
                        bool contains_deletion(const dtn::data::BundleID &id);

                        // set of bundles to delete
                        ibrcommon::Mutex _deletion_mutex;
                        std::set<dtn::data::BundleID> _deletion_list;
                };
        }
}

#endif /* SQLITEBUNDLESTORAGE_H_ */