3 files changed, 350 insertions, 42 deletions

diff --git a/common/storage.h b/common/storage.h
index 09365b0..77f11f9 100644
--- a/common/storage.h
+++ b/common/storage.h
@@ -51,8 +51,64 @@ public:
         int code;
     };
 
+    class Transaction
+    {
+    public:
+        ~Transaction();
+        bool commit(const std::function<void(const Storage::Error &error)> &errorHandler = std::function<void(const Storage::Error &error)>());
+        void abort();
+
+        /**
+         * Write a value
+         */
+        bool write(const QByteArray &key, const QByteArray &value, const std::function<void(const Storage::Error &error)> &errorHandler = std::function<void(const Storage::Error &error)>());
+
+        /**
+         * Remove a value
+         */
+        void remove(const QByteArray &key,
+                    const std::function<void(const Storage::Error &error)> &errorHandler);
+        /**
+        * Read values with a given key.
+        * 
+        * * An empty @param key results in a full scan
+        * * If duplicates are existing (revisions), all values are returned.
+        * * The pointers of the returned values are valid during the execution of the @param resultHandler
+        * 
+        * @return The number of values retrieved.
+        */
+        int scan(const QByteArray &k,
+                    const std::function<bool(const QByteArray &key, const QByteArray &value)> &resultHandler,
+                    const std::function<void(const Storage::Error &error)> &errorHandler = std::function<void(const Storage::Error &error)>());
+
+        Transaction(Transaction&& other) : d(other.d)
+        {
+            d = other.d;
+            other.d = nullptr;
+        } 
+        Transaction& operator=(Transaction&& other) {
+            d = other.d;
+            other.d = nullptr;
+            return *this;
+        }
+    private:
+        Transaction(Transaction& other);
+        Transaction& operator=(Transaction& other);
+        friend Storage;
+        class Private;
+        Transaction();
+        Transaction(Private*);
+        Private *d;
+    };
+
     Storage(const QString &storageRoot, const QString &name, AccessMode mode = ReadOnly, bool allowDuplicates = false);
     ~Storage();
+
+    Transaction createTransaction(AccessMode mode = ReadWrite,
+                  const std::function<void(const Storage::Error &error)> &errorHandler = std::function<void(const Storage::Error &error)>());
+
+
+    //Old API
     bool isInTransaction() const;
     bool startTransaction(AccessMode mode = ReadWrite, const std::function<void(const Storage::Error &error)> &errorHandler = std::function<void(const Storage::Error &error)>());
     bool commitTransaction(const std::function<void(const Storage::Error &error)> &errorHandler = std::function<void(const Storage::Error &error)>());
diff --git a/common/storage_lmdb.cpp b/common/storage_lmdb.cpp
index 7bbf8b5..230409f 100644
--- a/common/storage_lmdb.cpp
+++ b/common/storage_lmdb.cpp
@@ -47,6 +47,220 @@ int getErrorCode(int e)
     return -1;
 }
 
+
+
+class Storage::Transaction::Private
+{
+public:
+    Private(MDB_txn *_txn, MDB_dbi _dbi, bool _allowDuplicates, const std::function<void(const Storage::Error &error)> &_defaultErrorHandler, const QString &_name)
+        : transaction(_txn),
+        dbi(_dbi),
+        allowDuplicates(_allowDuplicates),
+        defaultErrorHandler(_defaultErrorHandler),
+        name(_name),
+        implicitCommit(false),
+        error(false)
+    {
+
+    }
+    ~Private()
+    {
+
+    }
+
+    MDB_txn *transaction;
+    MDB_dbi dbi;
+    bool allowDuplicates;
+    std::function<void(const Storage::Error &error)> defaultErrorHandler;
+    QString name;
+    bool implicitCommit;
+    bool error;
+};
+
+Storage::Transaction::Transaction()
+    : d(0)
+{
+
+}
+
+Storage::Transaction::Transaction(Transaction::Private *prv)
+    : d(prv)
+{
+
+}
+
+Storage::Transaction::~Transaction()
+{
+    if (d && d->transaction) {
+        if (d->implicitCommit && !d->error) {
+            commit();
+        } else {
+            mdb_txn_abort(d->transaction);
+        }
+    }
+    delete d;
+}
+
+bool Storage::Transaction::commit(const std::function<void(const Storage::Error &error)> &errorHandler)
+{
+    if (!d) {
+        return false;
+    }
+
+    const int rc = mdb_txn_commit(d->transaction);
+    if (rc) {
+        mdb_txn_abort(d->transaction);
+        Error error(d->name.toLatin1(), ErrorCodes::GenericError, "Error during transaction commit: " + QByteArray(mdb_strerror(rc)));
+        errorHandler ? errorHandler(error) : d->defaultErrorHandler(error);
+    }
+    d->transaction = nullptr;
+
+    return !rc;
+}
+
+void Storage::Transaction::abort()
+{
+    if (!d || !d->transaction) {
+        return;
+    }
+
+    mdb_txn_abort(d->transaction);
+    d->transaction = nullptr;
+}
+
+bool Storage::Transaction::write(const QByteArray &sKey, const QByteArray &sValue, const std::function<void(const Storage::Error &error)> &errorHandler)
+{
+    if (!d || !d->transaction) {
+        return false;
+    }
+    const void *keyPtr = sKey.data();
+    const size_t keySize = sKey.size();
+    const void *valuePtr = sValue.data();
+    const size_t valueSize = sValue.size();
+
+    if (!keyPtr || keySize == 0) {
+        Error error(d->name.toLatin1(), ErrorCodes::GenericError, "Tried to write empty key.");
+        errorHandler ? errorHandler(error) : d->defaultErrorHandler(error);
+        return false;
+    }
+
+    int rc;
+    MDB_val key, data;
+    key.mv_size = keySize;
+    key.mv_data = const_cast<void*>(keyPtr);
+    data.mv_size = valueSize;
+    data.mv_data = const_cast<void*>(valuePtr);
+    rc = mdb_put(d->transaction, d->dbi, &key, &data, 0);
+
+    if (rc) {
+        d->error = true;
+        Error error(d->name.toLatin1(), ErrorCodes::GenericError, "mdb_put: " + QByteArray(mdb_strerror(rc)));
+        errorHandler ? errorHandler(error) : d->defaultErrorHandler(error);
+    } else {
+        d->implicitCommit = true;
+    }
+
+    return !rc;
+}
+
+int Storage::Transaction::scan(const QByteArray &k,
+                  const std::function<bool(const QByteArray &key, const QByteArray &value)> &resultHandler,
+                  const std::function<void(const Storage::Error &error)> &errorHandler)
+{
+    if (!d || !d->transaction) {
+        Error error(d->name.toLatin1(), ErrorCodes::NotOpen, "Not open");
+        errorHandler ? errorHandler(error) : d->defaultErrorHandler(error);
+        return 0;
+    }
+
+    int rc;
+    MDB_val key;
+    MDB_val data;
+    MDB_cursor *cursor;
+
+    key.mv_data = (void*)k.constData();
+    key.mv_size = k.size();
+
+    rc = mdb_cursor_open(d->transaction, d->dbi, &cursor);
+    if (rc) {
+        Error error(d->name.toLatin1(), getErrorCode(rc), QByteArray("Error during mdb_cursor open: ") + QByteArray(mdb_strerror(rc)));
+        errorHandler ? errorHandler(error) : d->defaultErrorHandler(error);
+        return 0;
+    }
+
+    int numberOfRetrievedValues = 0;
+
+    if (k.isEmpty() || d->allowDuplicates) {
+        if ((rc = mdb_cursor_get(cursor, &key, &data, d->allowDuplicates ? MDB_SET_RANGE : MDB_FIRST)) == 0) {
+            numberOfRetrievedValues++;
+            if (resultHandler(QByteArray::fromRawData((char*)key.mv_data, key.mv_size), QByteArray::fromRawData((char*)data.mv_data, data.mv_size))) {
+                while ((rc = mdb_cursor_get(cursor, &key, &data, d->allowDuplicates ? MDB_NEXT_DUP : MDB_NEXT)) == 0) {
+                    numberOfRetrievedValues++;
+                    if (!resultHandler(QByteArray::fromRawData((char*)key.mv_data, key.mv_size), QByteArray::fromRawData((char*)data.mv_data, data.mv_size))) {
+                        break;
+                    }
+                }
+            }
+        }
+
+        //We never find the last value
+        if (rc == MDB_NOTFOUND) {
+            rc = 0;
+        }
+    } else {
+        if ((rc = mdb_cursor_get(cursor, &key, &data, MDB_SET)) == 0) {
+            numberOfRetrievedValues++;
+            resultHandler(QByteArray::fromRawData((char*)key.mv_data, key.mv_size), QByteArray::fromRawData((char*)data.mv_data, data.mv_size));
+        }
+    }
+
+    mdb_cursor_close(cursor);
+
+    if (rc) {
+        Error error(d->name.toLatin1(), getErrorCode(rc), QByteArray("Key: ") + k + " : " + QByteArray(mdb_strerror(rc)));
+        errorHandler ? errorHandler(error) : d->defaultErrorHandler(error);
+    }
+
+    return numberOfRetrievedValues;
+}
+
+void Storage::Transaction::remove(const QByteArray &k,
+                     const std::function<void(const Storage::Error &error)> &errorHandler)
+{
+    if (!d) {
+        Error error(d->name.toLatin1(), ErrorCodes::GenericError, "Not open");
+        errorHandler ? errorHandler(error) : d->defaultErrorHandler(error);
+        return;
+    }
+
+    int rc;
+    MDB_val key;
+    key.mv_size = k.size();
+    key.mv_data = const_cast<void*>(static_cast<const void*>(k.data()));
+    rc = mdb_del(d->transaction, d->dbi, &key, 0);
+
+    if (rc) {
+        d->error = true;
+        Error error(d->name.toLatin1(), ErrorCodes::GenericError, QString("Error on mdb_del: %1 %2").arg(rc).arg(mdb_strerror(rc)).toLatin1());
+        errorHandler ? errorHandler(error) : d->defaultErrorHandler(error);
+    } else {
+        d->implicitCommit = true;
+    }
+
+    return;
+}
+
+
+
+
+
+
+
+
+
+
+
+
 class Storage::Private
 {
 public:
@@ -81,33 +295,38 @@ Storage::Private::Private(const QString &s, const QString &n, AccessMode m, bool
       allowDuplicates(duplicates)
 {
     const QString fullPath(storageRoot + '/' + name);
-    QDir dir;
-    dir.mkpath(storageRoot);
-    dir.mkdir(fullPath);
-
-    //Ensure the environment is only created once
-    QMutexLocker locker(&sMutex);
-
-    int rc = 0;
-    /*
-     * It seems we can only ever have one environment open in the process. 
-     * Otherwise multi-threading breaks.
-     */
-    env = sEnvironments.value(fullPath);
-    if (!env) {
-        if ((rc = mdb_env_create(&env))) {
-            // TODO: handle error
-            std::cerr << "mdb_env_create: " << rc << " " << mdb_strerror(rc) << std::endl;
-        } else {
-            if ((rc = mdb_env_open(env, fullPath.toStdString().data(), mode == ReadOnly ? MDB_RDONLY : 0 , 0664))) {
-                std::cerr << "mdb_env_open: " << rc << " " << mdb_strerror(rc) << std::endl;
-                mdb_env_close(env);
-                env = 0;
+    QFileInfo dirInfo(fullPath);
+    if (!dirInfo.exists() && mode == ReadWrite) {
+        QDir().mkpath(fullPath);
+        dirInfo.refresh();
+    }
+    if (mode == ReadWrite && !dirInfo.permission(QFile::WriteOwner)) {
+        qCritical() << fullPath << "does not have write permissions. Aborting";
+    } else if (dirInfo.exists()) {
+        //Ensure the environment is only created once
+        QMutexLocker locker(&sMutex);
+
+        /*
+        * It seems we can only ever have one environment open in the process. 
+        * Otherwise multi-threading breaks.
+        */
+        env = sEnvironments.value(fullPath);
+        if (!env) {
+            int rc = 0;
+            if ((rc = mdb_env_create(&env))) {
+                // TODO: handle error
+                std::cerr << "mdb_env_create: " << rc << " " << mdb_strerror(rc) << std::endl;
             } else {
-                //FIXME: dynamic resize
-                const size_t dbSize = (size_t)10485760 * (size_t)100 * (size_t)80; //10MB * 800
-                mdb_env_set_mapsize(env, dbSize);
-                sEnvironments.insert(fullPath, env);
+                if ((rc = mdb_env_open(env, fullPath.toStdString().data(), mode == ReadOnly ? MDB_RDONLY : 0 , 0664))) {
+                    std::cerr << "mdb_env_open: " << rc << " " << mdb_strerror(rc) << std::endl;
+                    mdb_env_close(env);
+                    env = 0;
+                } else {
+                    //FIXME: dynamic resize
+                    const size_t dbSize = (size_t)10485760 * (size_t)8000; //1MB * 8000
+                    mdb_env_set_mapsize(env, dbSize);
+                    sEnvironments.insert(fullPath, env);
+                }
             }
         }
     }
@@ -142,9 +361,40 @@ bool Storage::exists() const
     return (d->env != 0);
 }
 
-bool Storage::isInTransaction() const
+Storage::Transaction Storage::createTransaction(AccessMode type, const std::function<void(const Storage::Error &error)> &errorHandlerArg)
 {
-    return d->transaction;
+    auto errorHandler = errorHandlerArg ? errorHandlerArg : defaultErrorHandler();
+    if (!d->env) {
+        errorHandler(Error(d->name.toLatin1(), ErrorCodes::GenericError, "Missing database environment"));
+        return Transaction();
+    }
+
+    bool requestedRead = type == ReadOnly;
+
+    if (d->mode == ReadOnly && !requestedRead) {
+        errorHandler(Error(d->name.toLatin1(), ErrorCodes::GenericError, "Requested read/write transaction in read-only mode."));
+        return Transaction();
+    }
+
+    int rc;
+    MDB_txn *txn;
+    rc = mdb_txn_begin(d->env, NULL, requestedRead ? MDB_RDONLY : 0, &txn);
+    if (!rc) {
+        //TODO: Move opening of dbi into Transaction for different named databases
+        MDB_dbi dbi;
+        rc = mdb_dbi_open(txn, NULL, d->allowDuplicates ? MDB_DUPSORT : 0, &dbi);
+        if (rc) {
+            errorHandler(Error(d->name.toLatin1(), ErrorCodes::GenericError, "Error while opening transaction: " + QByteArray(mdb_strerror(rc))));
+            return Transaction();
+        }
+        return Transaction(new Transaction::Private(txn, dbi, d->allowDuplicates, defaultErrorHandler(), d->name));
+    } else {
+        if (rc) {
+            errorHandler(Error(d->name.toLatin1(), ErrorCodes::GenericError, "Error while beginning transaction: " + QByteArray(mdb_strerror(rc))));
+            return Transaction();
+        }
+    }
+    return Transaction();
 }
 
 bool Storage::startTransaction(AccessMode type,
diff --git a/tests/storagetest.cpp b/tests/storagetest.cpp
index b088670..b71a767 100644
--- a/tests/storagetest.cpp
+++ b/tests/storagetest.cpp
@@ -20,17 +20,18 @@ private:
     void populate(int count)
     {
         Akonadi2::Storage storage(testDataPath, dbName, Akonadi2::Storage::ReadWrite);
+        auto transaction = storage.createTransaction(Akonadi2::Storage::ReadWrite);
         for (int i = 0; i < count; i++) {
             //This should perhaps become an implementation detail of the db?
             if (i % 10000 == 0) {
                 if (i > 0) {
-                    storage.commitTransaction();
+                    transaction.commit();
+                    transaction = std::move(storage.createTransaction(Akonadi2::Storage::ReadWrite));
                 }
-                storage.startTransaction();
             }
-            storage.write(keyPrefix + QByteArray::number(i), keyPrefix + QByteArray::number(i));
+            transaction.write(keyPrefix + QByteArray::number(i), keyPrefix + QByteArray::number(i));
         }
-        storage.commitTransaction();
+        transaction.commit();
     }
 
     bool verify(Akonadi2::Storage &storage, int i)
@@ -38,8 +39,8 @@ private:
         bool success = true;
         bool keyMatch = true;
         const auto reference = keyPrefix + QByteArray::number(i);
-        storage.scan(keyPrefix + QByteArray::number(i),
-            [&keyMatch, &reference](const QByteArray &value) -> bool {
+        storage.createTransaction(Akonadi2::Storage::ReadOnly).scan(keyPrefix + QByteArray::number(i),
+            [&keyMatch, &reference](const QByteArray &key, const QByteArray &value) -> bool {
                 if (value != reference) {
                     qDebug() << "Mismatch while reading";
                     keyMatch = false;
@@ -102,8 +103,8 @@ private Q_SLOTS:
         {
             int hit = 0;
             Akonadi2::Storage store(testDataPath, dbName);
-            store.scan("", [&](void *keyValue, int keySize, void *dataValue, int dataSize) -> bool {
-                if (std::string(static_cast<char*>(keyValue), keySize) == "key50") {
+            store.createTransaction(Akonadi2::Storage::ReadOnly).scan("", [&](const QByteArray &key, const QByteArray &value) -> bool {
+                if (key == "key50") {
                     hit++;
                 }
                 return true;
@@ -116,8 +117,8 @@ private Q_SLOTS:
             int hit = 0;
             bool foundInvalidValue = false;
             Akonadi2::Storage store(testDataPath, dbName);
-            store.scan("key50", [&](void *keyValue, int keySize, void *dataValue, int dataSize) -> bool {
-                if (std::string(static_cast<char*>(keyValue), keySize) != "key50") {
+            store.createTransaction(Akonadi2::Storage::ReadOnly).scan("key50", [&](const QByteArray &key, const QByteArray &value) -> bool {
+                if (key != "key50") {
                     foundInvalidValue = true;
                 }
                 hit++;
@@ -128,12 +129,13 @@ private Q_SLOTS:
         }
     }
 
-    void testTurnReadToWrite()
+    void testNestedOperations()
     {
         populate(3);
         Akonadi2::Storage store(testDataPath, dbName, Akonadi2::Storage::ReadWrite);
-        store.scan("key1", [&](void *keyValue, int keySize, void *dataValue, int dataSize) -> bool {
-            store.remove(QByteArray::fromRawData(static_cast<const char*>(keyValue), keySize), [](const Akonadi2::Storage::Error &) {
+        auto transaction = store.createTransaction(Akonadi2::Storage::ReadWrite);
+        transaction.scan("key1", [&](const QByteArray &key, const QByteArray &value) -> bool {
+            transaction.remove(key, [](const Akonadi2::Storage::Error &) {
                 QVERIFY(false);
             });
             return false;
@@ -145,7 +147,7 @@ private Q_SLOTS:
         bool gotResult = false;
         bool gotError = false;
         Akonadi2::Storage store(testDataPath, dbName, Akonadi2::Storage::ReadWrite);
-        int numValues = store.scan("", [&](void *keyValue, int keySize, void *dataValue, int dataSize) -> bool {
+        int numValues = store.createTransaction(Akonadi2::Storage::ReadOnly).scan("", [&](const QByteArray &key, const QByteArray &value) -> bool {
             gotResult = true;
             return false;
         },