52 files changed, 1114 insertions, 668 deletions
diff --git a/common/CMakeLists.txt b/common/CMakeLists.txt
index a80ef95..fe72605 100644
--- a/common/CMakeLists.txt
+++ b/common/CMakeLists.txt
@@ -1,41 +1,45 @@
 include_directories(${CMAKE_CURRENT_BINARY_DIR})
 include_directories(domain)
-project(sinkcommon)
+project(sink)
-ecm_setup_version("0.1" VARIABLE_PREFIX SinkCommon
+ecm_setup_version("0.1" VARIABLE_PREFIX Sink
-                        VERSION_HEADER "${CMAKE_CURRENT_BINARY_DIR}/sinkcommon_version.h"
+                        VERSION_HEADER "${CMAKE_CURRENT_BINARY_DIR}/sink_version.h"
-                        PACKAGE_VERSION_FILE "${CMAKE_CURRENT_BINARY_DIR}/SinkCommonConfigVersion.cmake"
+                        PACKAGE_VERSION_FILE "${CMAKE_CURRENT_BINARY_DIR}/SinkConfigVersion.cmake"
                        SOVERSION 0
 )
 ########### CMake Config Files ###########
-set(CMAKECONFIG_INSTALL_DIR "${KDE_INSTALL_CMAKEPACKAGEDIR}/SinkCommon")
+set(CMAKECONFIG_INSTALL_DIR "${KDE_INSTALL_CMAKEPACKAGEDIR}/Sink")
 ecm_configure_package_config_file(
-  "${CMAKE_CURRENT_SOURCE_DIR}/SinkCommonConfig.cmake.in"
+  "${CMAKE_CURRENT_SOURCE_DIR}/SinkConfig.cmake.in"
-  "${CMAKE_CURRENT_BINARY_DIR}/SinkCommonConfig.cmake"
+  "${CMAKE_CURRENT_BINARY_DIR}/SinkConfig.cmake"
  INSTALL_DESTINATION  ${CMAKECONFIG_INSTALL_DIR}
 )
 install(FILES
-  "${CMAKE_CURRENT_BINARY_DIR}/SinkCommonConfig.cmake"
+  "${CMAKE_CURRENT_BINARY_DIR}/SinkConfig.cmake"
-  "${CMAKE_CURRENT_BINARY_DIR}/SinkCommonConfigVersion.cmake"
+  "${CMAKE_CURRENT_BINARY_DIR}/SinkConfigVersion.cmake"
  DESTINATION "${CMAKECONFIG_INSTALL_DIR}"
  COMPONENT Devel
 )
-install(EXPORT SinkCommonTargets DESTINATION "${CMAKECONFIG_INSTALL_DIR}" FILE SinkCommonTargets.cmake)
+add_definitions("-fvisibility=hidden")
+install(EXPORT SinkTargets DESTINATION "${CMAKECONFIG_INSTALL_DIR}" FILE SinkTargets.cmake)
 set(storage_SRCS storage_lmdb.cpp)
 set(storage_LIBS lmdb)
 set(command_SRCS
+    store.cpp
+    notifier.cpp
+    resourcecontrol.cpp
    modelresult.cpp
    definitions.cpp
    log.cpp
    entitybuffer.cpp
-    clientapi.cpp
    facadefactory.cpp
    commands.cpp
    facade.cpp
@@ -85,7 +89,7 @@ generate_flatbuffers(
    queuedcommand
 )
-generate_export_header(${PROJECT_NAME} BASE_NAME SinkCommon EXPORT_FILE_NAME sinkcommon_export.h)
+generate_export_header(${PROJECT_NAME} BASE_NAME Sink EXPORT_FILE_NAME sink_export.h)
 SET_TARGET_PROPERTIES(${PROJECT_NAME}
    PROPERTIES LINKER_LANGUAGE CXX
    VERSION "0.1"
@@ -95,15 +99,20 @@ SET_TARGET_PROPERTIES(${PROJECT_NAME}
 qt5_use_modules(${PROJECT_NAME} Network)
 target_link_libraries(${PROJECT_NAME} ${storage_LIBS} KF5::Async)
 install(TARGETS ${PROJECT_NAME}
-    EXPORT SinkCommonTargets ${KDE_INSTALL_TARGETS_DEFAULT_ARGS} ${LIBRARY_NAMELINK} )
+    EXPORT SinkTargets ${KDE_INSTALL_TARGETS_DEFAULT_ARGS} ${LIBRARY_NAMELINK} )
+add_clang_static_analysis(${PROJECT_NAME})
 install(FILES
-    clientapi.h
+    store.h
+    notifier.h
+    resourcecontrol.h
    domain/applicationdomaintype.h
    query.h
    inspection.h
    notification.h
    bufferadaptor.h
+    ${CMAKE_CURRENT_BINARY_DIR}/sink_export.h
    DESTINATION ${INCLUDE_INSTALL_DIR}/${PROJECT_NAME} COMPONENT Devel
 )
diff --git a/common/SinkCommonConfig.cmake.in b/common/SinkConfig.cmake.in
index 6c2a1c2..930b2db 100644
--- a/common/SinkCommonConfig.cmake.in
+++ b/common/SinkConfig.cmake.in
@@ -2,4 +2,4 @@
 find_dependency(KF5Mime "@KMIME_LIB_VERSION@")
-include("${CMAKE_CURRENT_LIST_DIR}/SinkCommonTargets.cmake")
+include("${CMAKE_CURRENT_LIST_DIR}/SinkTargets.cmake")
diff --git a/common/clientapi.h b/common/clientapi.h
deleted file mode 100644
index 64c4f64..0000000
--- a/common/clientapi.h
+++ /dev/null
@@ -1,140 +0,0 @@
-/*
- * Copyright (C) 2014 Christian Mollekopf <chrigi_1@fastmail.fm>
- *
- * This library is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Lesser General Public
- * License as published by the Free Software Foundation; either
- * version 2.1 of the License, or (at your option) version 3, or any
- * later version accepted by the membership of KDE e.V. (or its
- * successor approved by the membership of KDE e.V.), which shall
- * act as a proxy defined in Section 6 of version 3 of the license.
- *
- * This library 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
- * Lesser General Public License for more details.
- *
- * You should have received a copy of the GNU Lesser General Public
- * License along with this library.  If not, see <http://www.gnu.org/licenses/>.
- */
-#pragma once
-#include <QString>
-#include <QSharedPointer>
-#include <Async/Async>
-#include "query.h"
-#include "inspection.h"
-#include "applicationdomaintype.h"
-class QAbstractItemModel;
-namespace Sink {
-class ResourceAccess;
-class Notification;
-/**
- * Store interface used in the client API.
- */
-class Store {
-public:
-    static QString storageLocation();
-    static QByteArray resourceName(const QByteArray &instanceIdentifier);
-    enum Roles {
-        DomainObjectRole = Qt::UserRole + 1, //Must be the same as in ModelResult
-        ChildrenFetchedRole,
-        DomainObjectBaseRole
-    };
-    /**
-     * Asynchronusly load a dataset with tree structure information
-     */
-    template <class DomainType>
-    static QSharedPointer<QAbstractItemModel> loadModel(Query query);
-    /**
-     * Create a new entity.
-     */
-    template <class DomainType>
-    static KAsync::Job<void> create(const DomainType &domainObject);
-    /**
-     * Modify an entity.
-     * 
-     * This includes moving etc. since these are also simple settings on a property.
-     */
-    template <class DomainType>
-    static KAsync::Job<void> modify(const DomainType &domainObject);
-    /**
-     * Remove an entity.
-     */
-    template <class DomainType>
-    static KAsync::Job<void> remove(const DomainType &domainObject);
-    /**
-     * Synchronize data to local cache.
-     */
-    static KAsync::Job<void> synchronize(const Sink::Query &query);
-    /**
-     * Shutdown resource.
-     */
-    static KAsync::Job<void> shutdown(const QByteArray &resourceIdentifier);
-    /**
-     * Start resource.
-     * 
-     * The resource is ready for operation once this command completes.
-     * This command is only necessary if a resource was shutdown previously,
-     * otherwise the resource process will automatically start as necessary.
-     */
-    static KAsync::Job<void> start(const QByteArray &resourceIdentifier);
-    /**
-     * Flushes any pending messages to disk
-     */
-    static KAsync::Job<void> flushMessageQueue(const QByteArrayList &resourceIdentifier);
-    /**
-     * Flushes any pending messages that haven't been replayed to the source.
-     */
-    static KAsync::Job<void> flushReplayQueue(const QByteArrayList &resourceIdentifier);
-    /**
-     * Removes a resource from disk.
-     */
-    static void removeFromDisk(const QByteArray &resourceIdentifier);
-    template <class DomainType>
-    static KAsync::Job<DomainType> fetchOne(const Sink::Query &query);
-    template <class DomainType>
-    static KAsync::Job<QList<typename DomainType::Ptr> > fetchAll(const Sink::Query &query);
-    template <class DomainType>
-    static KAsync::Job<QList<typename DomainType::Ptr> > fetch(const Sink::Query &query, int minimumAmount = 0);
-};
-namespace Resources {
-    template <class DomainType>
-    KAsync::Job<void> inspect(const Inspection &inspectionCommand);
-}
-class Notifier {
-public:
-    Notifier(const QSharedPointer<ResourceAccess> &resourceAccess);
-    // Notifier(const QByteArray &resource);
-    // Notifier(const QByteArrayList &resource);
-    void registerHandler(std::function<void(const Notification &)>);
-private:
-    class Private;
-    QScopedPointer<Private> d;
-};
-}
diff --git a/common/commands.cpp b/common/commands.cpp
index 8b915f0..5d38afa 100644
--- a/common/commands.cpp
+++ b/common/commands.cpp
@@ -61,6 +61,8 @@ QByteArray name(int commandId)
            return "RevisionReplayed";
        case InspectionCommand:
            return "Inspection";
+        case RemoveFromDiskCommand:
+            return "RemoveFromDisk";
        case CustomCommand:
            return "Custom";
    };
diff --git a/common/commands.h b/common/commands.h
index 19c827e..64abd76 100644
--- a/common/commands.h
+++ b/common/commands.h
@@ -20,7 +20,7 @@
 #pragma once
-#include <sinkcommon_export.h>
+#include "sink_export.h"
 #include <flatbuffers/flatbuffers.h>
 #include <QByteArray>
@@ -48,16 +48,17 @@ enum CommandIds {
    PingCommand,
    RevisionReplayedCommand,
    InspectionCommand,
+    RemoveFromDiskCommand,
    CustomCommand = 0xffff
 };
 QByteArray name(int commandId);
-int SINKCOMMON_EXPORT headerSize();
+int SINK_EXPORT headerSize();
-void SINKCOMMON_EXPORT write(QIODevice *device, int messageId, int commandId);
+void SINK_EXPORT write(QIODevice *device, int messageId, int commandId);
-void SINKCOMMON_EXPORT write(QIODevice *device, int messageId, int commandId, const char *buffer, uint size);
+void SINK_EXPORT write(QIODevice *device, int messageId, int commandId, const char *buffer, uint size);
-void SINKCOMMON_EXPORT write(QIODevice *device, int messageId, int commandId, flatbuffers::FlatBufferBuilder &fbb);
+void SINK_EXPORT write(QIODevice *device, int messageId, int commandId, flatbuffers::FlatBufferBuilder &fbb);
 }
diff --git a/common/commands/notification.fbs b/common/commands/notification.fbs
index ff01fc5..8750ff5 100644
--- a/common/commands/notification.fbs
+++ b/common/commands/notification.fbs
@@ -1,6 +1,6 @@
 namespace Sink.Commands;
-enum NotificationType : byte { Shutdown = 1, Status, Warning, Progress, Inspection }
+enum NotificationType : byte { Shutdown = 1, Status, Warning, Progress, Inspection, RevisionUpdate }
 enum NotificationCode : byte { Success = 0, Failure = 1, UserCode }
 table Notification {
diff --git a/common/definitions.h b/common/definitions.h
index 5834f01..029d3f8 100644
--- a/common/definitions.h
+++ b/common/definitions.h
@@ -20,10 +20,11 @@
 #pragma once
+#include "sink_export.h"
 #include <QString>
 #include <QByteArray>
 namespace Sink {
-    QString storageLocation();
+    QString SINK_EXPORT storageLocation();
-    QByteArray resourceName(const QByteArray &instanceIdentifier);
+    QByteArray SINK_EXPORT resourceName(const QByteArray &instanceIdentifier);
 }
diff --git a/common/domain/applicationdomaintype.cpp b/common/domain/applicationdomaintype.cpp
index b0433be..df10327 100644
--- a/common/domain/applicationdomaintype.cpp
+++ b/common/domain/applicationdomaintype.cpp
@@ -64,6 +64,12 @@ ApplicationDomainType::~ApplicationDomainType()
 {
 }
+bool ApplicationDomainType::hasProperty(const QByteArray &key) const
+{
+    Q_ASSERT(mAdaptor);
+    return mAdaptor->availableProperties().contains(key);
+}
 QVariant ApplicationDomainType::getProperty(const QByteArray &key) const
 {
    Q_ASSERT(mAdaptor);
@@ -76,13 +82,18 @@ QVariant ApplicationDomainType::getProperty(const QByteArray &key) const
 void ApplicationDomainType::setProperty(const QByteArray &key, const QVariant &value)
 {
    Q_ASSERT(mAdaptor);
-    mChangeSet.insert(key, value);
+    mChangeSet.insert(key);
    mAdaptor->setProperty(key, value);
 }
+void ApplicationDomainType::setChangedProperties(const QSet<QByteArray> &changeset)
+{
+    mChangeSet = changeset;
+}
 QByteArrayList ApplicationDomainType::changedProperties() const
 {
-    return mChangeSet.keys();
+    return mChangeSet.toList();
 }
 qint64 ApplicationDomainType::revision() const
@@ -100,6 +111,36 @@ QByteArray ApplicationDomainType::identifier() const
    return mIdentifier;
 }
+Entity::~Entity()
+{
+}
+Event::~Event()
+{
+}
+Todo::~Todo()
+{
+}
+Mail::~Mail()
+{
+}
+Folder::~Folder()
+{
+}
+SinkResource::~SinkResource()
+{
+}
 template<>
 QByteArray getTypeName<Event>()
 {
diff --git a/common/domain/applicationdomaintype.h b/common/domain/applicationdomaintype.h
index 63f030c..32d8999 100644
--- a/common/domain/applicationdomaintype.h
+++ b/common/domain/applicationdomaintype.h
@@ -19,9 +19,11 @@
 */
 #pragma once
+#include "sink_export.h"
 #include <QSharedPointer>
 #include <QVariant>
 #include <QByteArray>
+#include <QDebug>
 #include "bufferadaptor.h"
 namespace Sink {
@@ -35,7 +37,7 @@ namespace ApplicationDomain {
 *
 * ApplicationDomainTypes don't adhere to any standard and are meant to be extended frequently (hence the non-typesafe interface).
 */
-class ApplicationDomainType {
+class SINK_EXPORT ApplicationDomainType {
 public:
    typedef QSharedPointer<ApplicationDomainType> Ptr;
@@ -55,16 +57,19 @@ public:
    virtual ~ApplicationDomainType();
-    virtual QVariant getProperty(const QByteArray &key) const;
+    bool hasProperty(const QByteArray &key) const;
-    virtual void setProperty(const QByteArray &key, const QVariant &value);
+    QVariant getProperty(const QByteArray &key) const;
-    virtual QByteArrayList changedProperties() const;
+    void setProperty(const QByteArray &key, const QVariant &value);
+    void setChangedProperties(const QSet<QByteArray> &changeset);
+    QByteArrayList changedProperties() const;
    qint64 revision() const;
    QByteArray resourceInstanceIdentifier() const;
    QByteArray identifier() const;
 private:
+    friend QDebug operator<<(QDebug, const ApplicationDomainType &);
    QSharedPointer<BufferAdaptor> mAdaptor;
-    QHash<QByteArray, QVariant> mChangeSet;
+    QSet<QByteArray> mChangeSet;
    /*
     * Each domain object needs to store the resource, identifier, revision triple so we can link back to the storage location.
     */
@@ -82,34 +87,50 @@ inline bool operator==(const ApplicationDomainType& lhs, const ApplicationDomain
            && lhs.resourceInstanceIdentifier() == rhs.resourceInstanceIdentifier();
 }
-struct Entity : public ApplicationDomainType {
+inline QDebug operator<< (QDebug d, const ApplicationDomainType &type)
+{
+    d << "ApplicationDomainType(\n";
+    for (const auto &property : type.mAdaptor->availableProperties()) {
+        d << " " << property << "\t" << type.getProperty(property) << "\n";
+    }
+    d << ")";
+    return d;
+}
+struct SINK_EXPORT Entity : public ApplicationDomainType {
    typedef QSharedPointer<Entity> Ptr;
    using ApplicationDomainType::ApplicationDomainType;
+    virtual ~Entity();
 };
-struct Event : public Entity {
+struct SINK_EXPORT Event : public Entity {
    typedef QSharedPointer<Event> Ptr;
    using Entity::Entity;
+    virtual ~Event();
 };
-struct Todo : public Entity {
+struct SINK_EXPORT Todo : public Entity {
    typedef QSharedPointer<Todo> Ptr;
    using Entity::Entity;
+    virtual ~Todo();
 };
-struct Calendar : public Entity {
+struct SINK_EXPORT Calendar : public Entity {
    typedef QSharedPointer<Calendar> Ptr;
    using Entity::Entity;
+    virtual ~Calendar();
 };
-struct Mail : public Entity {
+struct SINK_EXPORT Mail : public Entity {
    typedef QSharedPointer<Mail> Ptr;
    using Entity::Entity;
+    virtual ~Mail();
 };
-struct Folder : public Entity {
+struct SINK_EXPORT Folder : public Entity {
    typedef QSharedPointer<Folder> Ptr;
    using Entity::Entity;
+    virtual ~Folder();
 };
 /**
@@ -118,9 +139,10 @@ struct Folder : public Entity {
 * This type is used for configuration of resources,
 * and for creating and removing resource instances.
 */
-struct SinkResource : public ApplicationDomainType {
+struct SINK_EXPORT SinkResource : public ApplicationDomainType {
    typedef QSharedPointer<SinkResource> Ptr;
    using ApplicationDomainType::ApplicationDomainType;
+    virtual ~SinkResource();
 };
 /**
@@ -129,22 +151,22 @@ struct SinkResource : public ApplicationDomainType {
 * Do not store these types to disk, they may change over time.
 */
 template<class DomainType>
-QByteArray getTypeName();
+QByteArray SINK_EXPORT getTypeName();
 template<>
-QByteArray getTypeName<Event>();
+QByteArray SINK_EXPORT getTypeName<Event>();
 template<>
-QByteArray getTypeName<Todo>();
+QByteArray SINK_EXPORT getTypeName<Todo>();
 template<>
-QByteArray getTypeName<SinkResource>();
+QByteArray SINK_EXPORT getTypeName<SinkResource>();
 template<>
-QByteArray getTypeName<Mail>();
+QByteArray SINK_EXPORT getTypeName<Mail>();
 template<>
-QByteArray getTypeName<Folder>();
+QByteArray SINK_EXPORT getTypeName<Folder>();
 /**
 * Type implementation.
@@ -153,7 +175,7 @@ QByteArray getTypeName<Folder>();
 * Contains all non-resource specific, but type-specific code.
 */
 template<typename DomainType>
-class TypeImplementation;
+class SINK_EXPORT TypeImplementation;
 }
 }
diff --git a/common/domain/event.cpp b/common/domain/event.cpp
index 9f81eb8..4210125 100644
--- a/common/domain/event.cpp
+++ b/common/domain/event.cpp
@@ -69,6 +69,7 @@ QSharedPointer<ReadPropertyMapper<TypeImplementation<Event>::Buffer> > TypeImple
 {
    auto propertyMapper = QSharedPointer<ReadPropertyMapper<Buffer> >::create();
    propertyMapper->addMapping<QString, Buffer>("summary", &Buffer::summary);
+    propertyMapper->addMapping<QString, Buffer>("description", &Buffer::description);
    propertyMapper->addMapping<QString, Buffer>("uid", &Buffer::uid);
    propertyMapper->addMapping<QByteArray, Buffer>("attachment", &Buffer::attachment);
    return propertyMapper;
@@ -78,6 +79,7 @@ QSharedPointer<WritePropertyMapper<TypeImplementation<Event>::BufferBuilder> > T
 {
    auto propertyMapper = QSharedPointer<WritePropertyMapper<BufferBuilder> >::create();
    propertyMapper->addMapping<QString>("summary", &BufferBuilder::add_summary);
+    propertyMapper->addMapping<QString>("description", &BufferBuilder::add_description);
    propertyMapper->addMapping<QString>("uid", &BufferBuilder::add_uid);
    propertyMapper->addMapping<QByteArray>("attachment", &BufferBuilder::add_attachment);
    return propertyMapper;
diff --git a/common/domainadaptor.h b/common/domainadaptor.h
index d43fad7..893c8d0 100644
--- a/common/domainadaptor.h
+++ b/common/domainadaptor.h
@@ -19,6 +19,7 @@
 #pragma once
+#include "sink_export.h"
 #include <QVariant>
 #include <QByteArray>
 #include <functional>
@@ -121,7 +122,7 @@ public:
 * This is required by the facade the read the value, and by the pipeline preprocessors to access the domain values in a generic way.
 */
 template<typename DomainType, typename ResourceBuffer, typename ResourceBuilder>
-class DomainTypeAdaptorFactory : public DomainTypeAdaptorFactoryInterface
+class SINK_EXPORT DomainTypeAdaptorFactory : public DomainTypeAdaptorFactoryInterface
 {
    typedef typename Sink::ApplicationDomain::TypeImplementation<DomainType>::Buffer LocalBuffer;
    typedef typename Sink::ApplicationDomain::TypeImplementation<DomainType>::BufferBuilder LocalBuilder;
diff --git a/common/domaintypeadaptorfactoryinterface.h b/common/domaintypeadaptorfactoryinterface.h
index d974bbf..4d0ce04 100644
--- a/common/domaintypeadaptorfactoryinterface.h
+++ b/common/domaintypeadaptorfactoryinterface.h
@@ -38,5 +38,11 @@ public:
    typedef QSharedPointer<DomainTypeAdaptorFactoryInterface> Ptr;
    virtual ~DomainTypeAdaptorFactoryInterface() {};
    virtual QSharedPointer<Sink::ApplicationDomain::BufferAdaptor> createAdaptor(const Sink::Entity &entity) = 0;
+    /*
+     * Creates a buffer from @param domainType
+     *
+     * Note that this only serialized parameters that are part of ApplicationDomainType::changedProperties()
+     */
    virtual void createBuffer(const Sink::ApplicationDomain::ApplicationDomainType &domainType, flatbuffers::FlatBufferBuilder &fbb, void const *metadataData = 0, size_t metadataSize = 0) = 0;
 };
diff --git a/common/entitybuffer.cpp b/common/entitybuffer.cpp
index 0e5435a..b4a5cb2 100644
--- a/common/entitybuffer.cpp
+++ b/common/entitybuffer.cpp
@@ -18,6 +18,12 @@ EntityBuffer::EntityBuffer(const void *dataValue, int dataSize)
    }
 }
+EntityBuffer::EntityBuffer(const QByteArray &data)
+    : EntityBuffer(data.constData(), data.size())
+{
+}
 bool EntityBuffer::isValid() const
 {
    return mEntity;
@@ -25,6 +31,7 @@ bool EntityBuffer::isValid() const
 const Sink::Entity &EntityBuffer::entity()
 {
+    Q_ASSERT(mEntity);
    return *mEntity;
 }
diff --git a/common/entitybuffer.h b/common/entitybuffer.h
index c9c2453..474a619 100644
--- a/common/entitybuffer.h
+++ b/common/entitybuffer.h
@@ -1,14 +1,17 @@
 #pragma once
+#include "sink_export.h"
 #include <functional>
 #include <flatbuffers/flatbuffers.h>
+#include <QByteArray>
 namespace Sink {
 struct Entity;
-class EntityBuffer {
+class SINK_EXPORT EntityBuffer {
 public:
    EntityBuffer(const void *dataValue, int size);
+    EntityBuffer(const QByteArray &data);
    const uint8_t *resourceBuffer();
    const uint8_t *metadataBuffer();
    const uint8_t *localBuffer();
diff --git a/common/facade.cpp b/common/facade.cpp
index 8cb776c..1b91ce4 100644
--- a/common/facade.cpp
+++ b/common/facade.cpp
@@ -29,6 +29,14 @@
 using namespace Sink;
+#undef DEBUG_AREA
+#define DEBUG_AREA "client.facade"
+/**
+ * A factory for resource access instances that caches the instance for some time.
+ * 
+ * This avoids constantly recreating connections, and should allow a single process to have one connection per resource.
+ */
 class ResourceAccessFactory {
 public:
    static ResourceAccessFactory &instance()
@@ -140,6 +148,7 @@ QPair<KAsync::Job<void>, typename ResultEmitter<typename DomainType::Ptr>::Ptr>
 {
    //The runner lives for the lifetime of the query
    auto runner = new QueryRunner<DomainType>(query, mResourceAccess, mResourceInstanceIdentifier, mDomainTypeAdaptorFactory, bufferTypeForDomainType());
+    runner->setResultTransformation(mResultTransformation);
    return qMakePair(KAsync::null<void>(), runner->emitter());
 }
diff --git a/common/facade.h b/common/facade.h
index c25464f..a85b1de 100644
--- a/common/facade.h
+++ b/common/facade.h
@@ -19,6 +19,7 @@
 #pragma once
+#include "sink_export.h"
 #include "facadeinterface.h"
 #include <QByteArray>
@@ -43,7 +44,7 @@ namespace Sink {
 * Additionally a resource only has to provide a synchronizer plugin to execute the synchronization
 */
 template <typename DomainType>
-class GenericFacade: public Sink::StoreFacade<DomainType>
+class SINK_EXPORT GenericFacade: public Sink::StoreFacade<DomainType>
 {
 public:
    /**
@@ -59,9 +60,10 @@ public:
    KAsync::Job<void> create(const DomainType &domainObject) Q_DECL_OVERRIDE;
    KAsync::Job<void> modify(const DomainType &domainObject) Q_DECL_OVERRIDE;
    KAsync::Job<void> remove(const DomainType &domainObject) Q_DECL_OVERRIDE;
-    QPair<KAsync::Job<void>, typename ResultEmitter<typename DomainType::Ptr>::Ptr> load(const Sink::Query &query) Q_DECL_OVERRIDE;
+    virtual QPair<KAsync::Job<void>, typename ResultEmitter<typename DomainType::Ptr>::Ptr> load(const Sink::Query &query) Q_DECL_OVERRIDE;
 protected:
+    std::function<void(Sink::ApplicationDomain::ApplicationDomainType &domainObject)> mResultTransformation;
    //TODO use one resource access instance per application & per resource
    QSharedPointer<Sink::ResourceAccessInterface> mResourceAccess;
    DomainTypeAdaptorFactoryInterface::Ptr mDomainTypeAdaptorFactory;
diff --git a/common/facadefactory.h b/common/facadefactory.h
index 3dca63b..ef2a3f9 100644
--- a/common/facadefactory.h
+++ b/common/facadefactory.h
@@ -20,6 +20,7 @@
 #pragma once
+#include "sink_export.h"
 #include <QByteArray>
 #include <QDebug>
 #include <QMutex>
@@ -37,7 +38,7 @@ namespace Sink {
 *
 * If we were to provide default implementations for certain capabilities. Here would be the place to do so.
 */
-class FacadeFactory {
+class SINK_EXPORT FacadeFactory {
 public:
    typedef std::function<std::shared_ptr<void>(const QByteArray &)> FactoryFunction;
diff --git a/common/genericresource.cpp b/common/genericresource.cpp
index c7326d3..2688df0 100644
--- a/common/genericresource.cpp
+++ b/common/genericresource.cpp
@@ -17,11 +17,17 @@
 #include <QUuid>
 #include <QDataStream>
+#include <QTime>
 static int sBatchSize = 100;
+//This interval directly affects the roundtrip time of single commands
+static int sCommitInterval = 10;
 using namespace Sink;
+#undef DEBUG_AREA
+#define DEBUG_AREA "resource.changereplay"
 /**
 * Replays changes from the storage one by one.
 *
@@ -87,7 +93,7 @@ public Q_SLOTS:
                const auto uid = Storage::getUidFromRevision(mainStoreTransaction, revision);
                const auto type = Storage::getTypeFromRevision(mainStoreTransaction, revision);
                const auto key = Storage::assembleKey(uid, revision);
-                mainStoreTransaction.openDatabase(type + ".main").scan(key, [&lastReplayedRevision, type, this](const QByteArray &key, const QByteArray &value) -> bool {
+                Storage::mainDatabase(mainStoreTransaction, type).scan(key, [&lastReplayedRevision, type, this](const QByteArray &key, const QByteArray &value) -> bool {
                    mReplayFunction(type, key, value).exec();
                    //TODO make for loop async, and pass to async replay function together with type
                    Trace() << "Replaying " << key;
@@ -110,6 +116,9 @@ private:
    ReplayFunction mReplayFunction;
 };
+#undef DEBUG_AREA
+#define DEBUG_AREA "resource.commandprocessor"
 /**
 * Drives the pipeline using the output from all command queues
 */
@@ -197,7 +206,6 @@ private slots:
            default:
                return KAsync::error<qint64>(-1, "Unhandled command");
        }
-        return KAsync::null<qint64>();
    }
    KAsync::Job<qint64, qint64> processQueuedCommand(const QByteArray &data)
@@ -226,15 +234,17 @@ private slots:
    //Process all messages of this queue
    KAsync::Job<void> processQueue(MessageQueue *queue)
    {
+        auto time = QSharedPointer<QTime>::create();
        return KAsync::start<void>([this](){
            mPipeline->startTransaction();
        }).then(KAsync::dowhile(
            [queue]() { return !queue->isEmpty(); },
-            [this, queue](KAsync::Future<void> &future) {
+            [this, queue, time](KAsync::Future<void> &future) {
-                queue->dequeueBatch(sBatchSize, [this](const QByteArray &data) {
+                queue->dequeueBatch(sBatchSize, [this, time](const QByteArray &data) {
-                        return KAsync::start<void>([this, data](KAsync::Future<void> &future) {
+                        time->start();
-                            processQueuedCommand(data).then<void, qint64>([&future, this](qint64 createdRevision) {
+                        return KAsync::start<void>([this, data, time](KAsync::Future<void> &future) {
-                                Trace() << "Created revision " << createdRevision;
+                            processQueuedCommand(data).then<void, qint64>([&future, this, time](qint64 createdRevision) {
+                                Trace() << "Created revision " << createdRevision << ". Processing took: " << Log::TraceTime(time->elapsed());
                                future.setFinished();
                            }).exec();
                        });
@@ -256,21 +266,27 @@ private slots:
    KAsync::Job<void> processPipeline()
    {
+        auto time = QSharedPointer<QTime>::create();
+        time->start();
        mPipeline->startTransaction();
        Trace() << "Cleaning up from " << mPipeline->cleanedUpRevision() + 1 << " to " << mLowerBoundRevision;
        for (qint64 revision = mPipeline->cleanedUpRevision() + 1; revision <= mLowerBoundRevision; revision++) {
            mPipeline->cleanupRevision(revision);
        }
        mPipeline->commit();
+        Trace() << "Cleanup done." << Log::TraceTime(time->elapsed());
        //Go through all message queues
        auto it = QSharedPointer<QListIterator<MessageQueue*> >::create(mCommandQueues);
        return KAsync::dowhile(
            [it]() { return it->hasNext(); },
            [it, this](KAsync::Future<void> &future) {
+                auto time = QSharedPointer<QTime>::create();
+                time->start();
                auto queue = it->next();
-                processQueue(queue).then<void>([&future]() {
+                processQueue(queue).then<void>([&future, time]() {
-                    Trace() << "Queue processed";
+                    Trace() << "Queue processed." << Log::TraceTime(time->elapsed());
                    future.setFinished();
                }).exec();
            }
@@ -287,6 +303,8 @@ private:
    InspectionFunction mInspect;
 };
+#undef DEBUG_AREA
+#define DEBUG_AREA "resource"
 GenericResource::GenericResource(const QByteArray &resourceInstanceIdentifier, const QSharedPointer<Pipeline> &pipeline)
    : Sink::Resource(),
@@ -313,12 +331,14 @@ GenericResource::GenericResource(const QByteArray &resourceInstanceIdentifier, c
            QVariant expectedValue;
            s >> expectedValue;
            inspect(inspectionType, inspectionId, domainType, entityId, property, expectedValue).then<void>([=]() {
+                Log_area("resource.inspection") << "Inspection was successful: " << inspectionType << inspectionId << entityId;
                Sink::Notification n;
                n.type = Sink::Commands::NotificationType_Inspection;
                n.id = inspectionId;
                n.code = Sink::Commands::NotificationCode_Success;
                emit notify(n);
            }, [=](int code, const QString &message) {
+                Log() << "Inspection failed: "<< inspectionType << inspectionId << entityId << message;
                Sink::Notification n;
                n.type = Sink::Commands::NotificationType_Inspection;
                n.message = message;
@@ -341,7 +361,7 @@ GenericResource::GenericResource(const QByteArray &resourceInstanceIdentifier, c
    mClientLowerBoundRevision = mPipeline->cleanedUpRevision();
    mProcessor->setOldestUsedRevision(mSourceChangeReplay->getLastReplayedRevision());
-    mCommitQueueTimer.setInterval(100);
+    mCommitQueueTimer.setInterval(sCommitInterval);
    mCommitQueueTimer.setSingleShot(true);
    QObject::connect(&mCommitQueueTimer, &QTimer::timeout, &mUserQueue, &MessageQueue::commit);
 }
@@ -381,13 +401,18 @@ KAsync::Job<void> GenericResource::replay(Sink::Storage &synchronizationStore, c
    return KAsync::null<void>();
 }
+void GenericResource::removeDataFromDisk()
+{
+    removeFromDisk(mResourceInstanceIdentifier);
+}
 void GenericResource::removeFromDisk(const QByteArray &instanceIdentifier)
 {
-    Warning() << "Removing from generic resource";
    Sink::Storage(Sink::storageLocation(), instanceIdentifier, Sink::Storage::ReadWrite).removeFromDisk();
    Sink::Storage(Sink::storageLocation(), instanceIdentifier + ".userqueue", Sink::Storage::ReadWrite).removeFromDisk();
    Sink::Storage(Sink::storageLocation(), instanceIdentifier + ".synchronizerqueue", Sink::Storage::ReadWrite).removeFromDisk();
    Sink::Storage(Sink::storageLocation(), instanceIdentifier + ".changereplay", Sink::Storage::ReadWrite).removeFromDisk();
+    Sink::Storage(Sink::storageLocation(), instanceIdentifier + ".synchronization", Sink::Storage::ReadWrite).removeFromDisk();
 }
 qint64 GenericResource::diskUsage(const QByteArray &instanceIdentifier)
@@ -556,38 +581,39 @@ void GenericResource::deleteEntity(const QByteArray &sinkId, qint64 revision, co
 void GenericResource::recordRemoteId(const QByteArray &bufferType, const QByteArray &localId, const QByteArray &remoteId, Sink::Storage::Transaction &transaction)
 {
-    Index index("rid.mapping." + bufferType, transaction);
+    Index("rid.mapping." + bufferType, transaction).add(remoteId, localId);;
-    Index localIndex("localid.mapping." + bufferType, transaction);
+    Index("localid.mapping." + bufferType, transaction).add(localId, remoteId);
-    index.add(remoteId, localId);
-    localIndex.add(localId, remoteId);
 }
 void GenericResource::removeRemoteId(const QByteArray &bufferType, const QByteArray &localId, const QByteArray &remoteId, Sink::Storage::Transaction &transaction)
 {
-    Index index("rid.mapping." + bufferType, transaction);
+    Index("rid.mapping." + bufferType, transaction).remove(remoteId, localId);
-    Index localIndex("localid.mapping." + bufferType, transaction);
+    Index("localid.mapping." + bufferType, transaction).remove(localId, remoteId);
-    index.remove(remoteId, localId);
+}
-    localIndex.remove(localId, remoteId);
+void GenericResource::updateRemoteId(const QByteArray &bufferType, const QByteArray &localId, const QByteArray &remoteId, Sink::Storage::Transaction &transaction)
+{
+    const auto oldRemoteId = Index("localid.mapping." + bufferType, transaction).lookup(localId);
+    removeRemoteId(bufferType, localId, oldRemoteId, transaction);
+    recordRemoteId(bufferType, localId, remoteId, transaction);
 }
 QByteArray GenericResource::resolveRemoteId(const QByteArray &bufferType, const QByteArray &remoteId, Sink::Storage::Transaction &transaction)
 {
    //Lookup local id for remote id, or insert a new pair otherwise
    Index index("rid.mapping." + bufferType, transaction);
-    Index localIndex("localid.mapping." + bufferType, transaction);
    QByteArray sinkId = index.lookup(remoteId);
    if (sinkId.isEmpty()) {
        sinkId = QUuid::createUuid().toString().toUtf8();
        index.add(remoteId, sinkId);
-        localIndex.add(sinkId, remoteId);
+        Index("localid.mapping." + bufferType, transaction).add(sinkId, remoteId);
    }
    return sinkId;
 }
 QByteArray GenericResource::resolveLocalId(const QByteArray &bufferType, const QByteArray &localId, Sink::Storage::Transaction &transaction)
 {
-    Index index("localid.mapping." + bufferType, transaction);
+    QByteArray remoteId = Index("localid.mapping." + bufferType, transaction).lookup(localId);
-    QByteArray remoteId = index.lookup(localId);
    if (remoteId.isEmpty()) {
        Warning() << "Couldn't find the remote id for " << localId;
        return QByteArray();
@@ -633,7 +659,7 @@ static QSharedPointer<Sink::ApplicationDomain::BufferAdaptor> getLatest(const Si
 void GenericResource::createOrModify(Sink::Storage::Transaction &transaction, Sink::Storage::Transaction &synchronizationTransaction, DomainTypeAdaptorFactoryInterface &adaptorFactory, const QByteArray &bufferType, const QByteArray &remoteId, const Sink::ApplicationDomain::ApplicationDomainType &entity)
 {
-    auto mainDatabase = transaction.openDatabase(bufferType + ".main");
+    auto mainDatabase = Storage::mainDatabase(transaction, bufferType);
    const auto sinkId = resolveRemoteId(bufferType, remoteId, synchronizationTransaction);
    const auto found = mainDatabase.contains(sinkId);
    if (!found) {
@@ -663,4 +689,7 @@ void GenericResource::createOrModify(Sink::Storage::Transaction &transaction, Si
 }
+#pragma clang diagnostic push
+#pragma clang diagnostic ignored "-Wundefined-reinterpret-cast"
 #include "genericresource.moc"
+#pragma clang diagnostic pop
diff --git a/common/genericresource.h b/common/genericresource.h
index 4ae2645..ceb6b61 100644
--- a/common/genericresource.h
+++ b/common/genericresource.h
@@ -19,7 +19,7 @@
 */
 #pragma once
-#include <sinkcommon_export.h>
+#include "sink_export.h"
 #include <resource.h>
 #include <messagequeue.h>
 #include <flatbuffers/flatbuffers.h>
@@ -37,7 +37,7 @@ class Preprocessor;
 /**
 * Generic Resource implementation.
 */
-class SINKCOMMON_EXPORT GenericResource : public Resource
+class SINK_EXPORT GenericResource : public Resource
 {
 public:
    GenericResource(const QByteArray &resourceInstanceIdentifier, const QSharedPointer<Pipeline> &pipeline = QSharedPointer<Pipeline>());
@@ -52,6 +52,7 @@ public:
    int error() const;
+    void removeDataFromDisk() Q_DECL_OVERRIDE;
    static void removeFromDisk(const QByteArray &instanceIdentifier);
    static qint64 diskUsage(const QByteArray &instanceIdentifier);
@@ -74,6 +75,7 @@ protected:
     */
    void recordRemoteId(const QByteArray &bufferType, const QByteArray &localId, const QByteArray &remoteId, Sink::Storage::Transaction &transaction);
    void removeRemoteId(const QByteArray &bufferType, const QByteArray &localId, const QByteArray &remoteId, Sink::Storage::Transaction &transaction);
+    void updateRemoteId(const QByteArray &bufferType, const QByteArray &localId, const QByteArray &remoteId, Sink::Storage::Transaction &transaction);
    /**
     * Tries to find a local id for the remote id, and creates a new local id otherwise.
diff --git a/common/index.h b/common/index.h
index 20213b2..1a5b250 100644
--- a/common/index.h
+++ b/common/index.h
@@ -1,5 +1,6 @@
 #pragma once
+#include "sink_export.h"
 #include <string>
 #include <functional>
 #include <QString>
@@ -8,7 +9,7 @@
 /**
 * An index for value pairs.
 */
-class Index
+class SINK_EXPORT Index
 {
 public:
    enum ErrorCodes {
diff --git a/common/inspection.h b/common/inspection.h
index b9f6bf3..d85eab6 100644
--- a/common/inspection.h
+++ b/common/inspection.h
@@ -24,7 +24,7 @@
 #include "applicationdomaintype.h"
 namespace Sink {
-    namespace Resources {
+    namespace ResourceControl {
 struct Inspection {
    static Inspection PropertyInspection(const Sink::ApplicationDomain::Entity &entity, const QByteArray &property, const QVariant &expectedValue)
diff --git a/common/listener.cpp b/common/listener.cpp
index 13ebbbb..ed6f305 100644
--- a/common/listener.cpp
+++ b/common/listener.cpp
@@ -39,6 +39,9 @@
 #include <QTime>
 #include <QDataStream>
+#undef DEBUG_AREA
+#define DEBUG_AREA "resource.communication"
 Listener::Listener(const QByteArray &resourceInstanceIdentifier, QObject *parent)
    : QObject(parent),
      m_server(new QLocalServer(this)),
@@ -234,7 +237,7 @@ void Listener::processCommand(int commandId, uint messageId, const QByteArray &c
                    job = job.then<void>(loadResource()->processAllMessages());
                }
                job.then<void>([callback, timer]() {
-                    Trace() << "Sync took " << timer->elapsed();
+                    Trace() << "Sync took " << Sink::Log::TraceTime(timer->elapsed());
                    callback(true);
                }).exec();
                return;
@@ -272,6 +275,14 @@ void Listener::processCommand(int commandId, uint messageId, const QByteArray &c
            loadResource()->setLowerBoundRevision(lowerBoundRevision());
        }
            break;
+        case Sink::Commands::RemoveFromDiskCommand: {
+            Log() << QString("\tReceived a remove from disk command from %1").arg(client.name);
+            m_resource->removeDataFromDisk();
+            delete m_resource;
+            m_resource = nullptr;
+            loadResource()->setLowerBoundRevision(0);
+        }
+            break;
        default:
            if (commandId > Sink::Commands::CustomCommand) {
                Log() << QString("\tReceived custom command from %1: ").arg(client.name) << commandId;
@@ -424,3 +435,7 @@ Sink::Resource *Listener::loadResource()
    return m_resource;
 }
+#pragma clang diagnostic push
+#pragma clang diagnostic ignored "-Wundefined-reinterpret-cast"
+#include "moc_listener.cpp"
+#pragma clang diagnostic pop
diff --git a/common/listener.h b/common/listener.h
index e17f315..5bcc93e 100644
--- a/common/listener.h
+++ b/common/listener.h
@@ -19,6 +19,7 @@
 #pragma once
+#include "sink_export.h"
 #include <QObject>
 #include <QPointer>
@@ -56,7 +57,7 @@ public:
    qint64 currentRevision;
 };
-class Listener : public QObject
+class SINK_EXPORT Listener : public QObject
 {
    Q_OBJECT
diff --git a/common/listmodelresult.h b/common/listmodelresult.h
deleted file mode 100644
index 71a0d09..0000000
--- a/common/listmodelresult.h
+++ /dev/null
@@ -1,125 +0,0 @@
-/*
- * Copyright (C) 2014 Christian Mollekopf <chrigi_1@fastmail.fm>
- *
- * This library is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Lesser General Public
- * License as published by the Free Software Foundation; either
- * version 2.1 of the License, or (at your option) version 3, or any
- * later version accepted by the membership of KDE e.V. (or its
- * successor approved by the membership of KDE e.V.), which shall
- * act as a proxy defined in Section 6 of version 3 of the license.
- *
- * This library 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
- * Lesser General Public License for more details.
- *
- * You should have received a copy of the GNU Lesser General Public
- * License along with this library.  If not, see <http://www.gnu.org/licenses/>.
- */
-#pragma once
-#include <QAbstractListModel>
-#include <QDebug>
-#include "resultprovider.h"
-enum Roles {
-    DomainObjectRole = Qt::UserRole + 1
-};
-template<class T>
-class ListModelResult : public QAbstractListModel
-{
-public:
-    ListModelResult(const QList<QByteArray> &propertyColumns)
-        :QAbstractListModel(),
-        mPropertyColumns(propertyColumns)
-    {
-    }
-    ListModelResult(const QSharedPointer<Sink::ResultEmitter<T> > &emitter, const QList<QByteArray> &propertyColumns)
-        :QAbstractListModel(),
-        mPropertyColumns(propertyColumns)
-    {
-        setEmitter(emitter);
-    }
-    void setEmitter(const QSharedPointer<Sink::ResultEmitter<T> > &emitter)
-    {
-        beginResetModel();
-        mEntities.clear();
-        mEmitter = emitter;
-        emitter->onAdded([this](const T &value) {
-            const auto keys = mEntities.keys();
-            int index = 0;
-            for (; index < keys.size(); index++) {
-                if (value->identifier() < keys.at(index)) {
-                    break;
-                }
-            }
-            beginInsertRows(QModelIndex(), index, index);
-            mEntities.insert(value->identifier(), value);
-            endInsertRows();
-        });
-        emitter->onModified([this](const T &value) {
-            auto i = mEntities.keys().indexOf(value->identifier());
-            mEntities.remove(value->identifier());
-            mEntities.insert(value->identifier(), value);
-            auto idx = index(i, 0, QModelIndex());
-            emit dataChanged(idx, idx);
-        });
-        emitter->onRemoved([this](const T &value) {
-            auto index = mEntities.keys().indexOf(value->identifier());
-            beginRemoveRows(QModelIndex(), index, index);
-            mEntities.remove(value->identifier());
-            endRemoveRows();
-        });
-        emitter->onInitialResultSetComplete([this]() {
-        });
-        emitter->onComplete([this]() {
-            mEmitter.clear();
-        });
-        emitter->onClear([this]() {
-            beginResetModel();
-            mEntities.clear();
-            endResetModel();
-        });
-        endResetModel();
-    }
-    int rowCount(const QModelIndex &parent = QModelIndex()) const
-    {
-        return mEntities.size();
-    }
-    int columnCount(const QModelIndex &parent = QModelIndex()) const
-    {
-        return mPropertyColumns.size();
-    }
-    virtual QVariant data(const QModelIndex &index, int role = Qt::DisplayRole) const
-    {
-        if (index.row() >= mEntities.size()) {
-            qWarning() << "Out of bounds access";
-            return QVariant();
-        }
-        if (role == Qt::DisplayRole) {
-            if (index.column() < mPropertyColumns.size()) {
-                auto entity = mEntities.value(mEntities.keys().at(index.row()));
-                return entity->getProperty(mPropertyColumns.at(index.column())).toString();
-            }
-        }
-        if (role == DomainObjectRole) {
-            return QVariant::fromValue(mEntities.value(mEntities.keys().at(index.row())));
-        }
-        return QVariant();
-    }
-private:
-    QSharedPointer<Sink::ResultEmitter<T> > mEmitter;
-    QMap<QByteArray, T> mEntities;
-    QList<QByteArray> mPropertyColumns;
-};
diff --git a/common/log.cpp b/common/log.cpp
index 45bbec1..96c6f82 100644
--- a/common/log.cpp
+++ b/common/log.cpp
@@ -3,11 +3,19 @@
 #include <QString>
 #include <QIODevice>
 #include <QCoreApplication>
+#include <QSettings>
+#include <QStandardPaths>
+#include <QSharedPointer>
 #include <iostream>
 #include <unistd.h>
 using namespace Sink::Log;
+static QSharedPointer<QSettings> config()
+{
+    return QSharedPointer<QSettings>::create(QStandardPaths::writableLocation(QStandardPaths::GenericDataLocation) + "/sink/log.ini", QSettings::IniFormat);
+}
 class DebugStream: public QIODevice
 {
 public:
@@ -17,24 +25,24 @@ public:
    {
        open(WriteOnly);
    }
-    virtual ~DebugStream(){};
+    virtual ~DebugStream();
    bool isSequential() const { return true; }
    qint64 readData(char *, qint64) { return 0; /* eof */ }
    qint64 readLineData(char *, qint64) { return 0; /* eof */ }
    qint64 writeData(const char *data, qint64 len)
    {
-        const QByteArray buf = QByteArray::fromRawData(data, len);
+        std::cout << data << std::endl;
-        // if (!qgetenv("IMAP_TRACE").isEmpty()) {
-            // qt_message_output(QtDebugMsg, buf.trimmed().constData());
-            std::cout << buf.trimmed().constData() << std::endl;
-        // }
        return len;
    }
 private:
    Q_DISABLE_COPY(DebugStream)
 };
+//Virtual method anchor
+DebugStream::~DebugStream()
+{}
 class NullStream: public QIODevice
 {
 public:
@@ -43,7 +51,7 @@ public:
    {
        open(WriteOnly);
    }
-    virtual ~NullStream(){};
+    virtual ~NullStream();
    bool isSequential() const { return true; }
    qint64 readData(char *, qint64) { return 0; /* eof */ }
@@ -56,6 +64,10 @@ private:
    Q_DISABLE_COPY(NullStream)
 };
+//Virtual method anchor
+NullStream::~NullStream()
+{}
    /*
     * ANSI color codes:
     * 0: reset colors/style
@@ -130,31 +142,95 @@ DebugLevel Sink::Log::debugLevelFromName(const QByteArray &name)
 void Sink::Log::setDebugOutputLevel(DebugLevel debugLevel)
 {
-    qputenv("SINKDEBUGLEVEL", debugLevelName(debugLevel));
+    config()->setValue("level", debugLevel);
 }
 Sink::Log::DebugLevel Sink::Log::debugOutputLevel()
 {
-    return debugLevelFromName(qgetenv("SINKDEBUGLEVEL"));
+    return static_cast<Sink::Log::DebugLevel>(config()->value("level", Sink::Log::Log).toInt());
+}
+void Sink::Log::setDebugOutputFilter(FilterType type, const QByteArrayList &filter)
+{
+    switch (type) {
+        case ApplicationName:
+            config()->setValue("applicationfilter", QVariant::fromValue(filter));
+            break;
+        case Area:
+            config()->setValue("areafilter", QVariant::fromValue(filter));
+            break;
+    }
+}
+QByteArrayList Sink::Log::debugOutputFilter(FilterType type)
+{
+    switch (type) {
+        case ApplicationName:
+            return config()->value("applicationfilter").value<QByteArrayList>();
+        case Area:
+            return config()->value("areafilter").value<QByteArrayList>();
+    }
+}
+void Sink::Log::setDebugOutputFields(const QByteArrayList &output)
+{
+    config()->setValue("outputfields", QVariant::fromValue(output));
+}
+QByteArrayList Sink::Log::debugOutputFields()
+{
+    return config()->value("outputfields").value<QByteArrayList>();
+}
+static QByteArray getProgramName()
+{
+    if (QCoreApplication::instance()) {
+        return QCoreApplication::instance()->applicationName().toLocal8Bit();
+    } else {
+        return "<unknown program name>";
+    }
+}
+static bool containsItemStartingWith(const QByteArray &pattern, const QByteArrayList &list)
+{
+    for (const auto &item : list) {
+        if (pattern.startsWith(item)) {
+            return true;
+        }
+    }
+    return false;
+}
+static bool caseInsensitiveContains(const QByteArray &pattern, const QByteArrayList &list)
+{
+    for (const auto &item : list) {
+        if (item.toLower() == pattern) {
+            return true;
+        }
+    }
+    return false;
 }
 QDebug Sink::Log::debugStream(DebugLevel debugLevel, int line, const char* file, const char* function, const char* debugArea)
 {
-    DebugLevel debugOutputLevel = debugLevelFromName(qgetenv("SINKDEBUGLEVEL"));
+    static NullStream nullstream;
-    if (debugLevel < debugOutputLevel) {
+    if (debugLevel < debugOutputLevel()) {
-        static NullStream stream;
+        return QDebug(&nullstream);
-        return QDebug(&stream);
    }
-    static DebugStream stream;
+    auto areas = debugOutputFilter(Sink::Log::Area);
-    QDebug debug(&stream);
+    if (debugArea && !areas.isEmpty()) {
+        if (!containsItemStartingWith(debugArea, areas)) {
+            return QDebug(&nullstream);
+        }
+    }
+    static QByteArray programName = getProgramName();
-    static QByteArray programName;
+    auto filter = debugOutputFilter(Sink::Log::ApplicationName);
-    if (programName.isEmpty()) {
+    if (!filter.isEmpty() && !filter.contains(programName)) {
-        if (QCoreApplication::instance())
+        if (!containsItemStartingWith(programName, filter)) {
-            programName = QCoreApplication::instance()->applicationName().toLocal8Bit();
+            return QDebug(&nullstream);
-        else
+        }
-            programName = "<unknown program name>";
    }
    QString prefix;
@@ -175,29 +251,54 @@ QDebug Sink::Log::debugStream(DebugLevel debugLevel, int line, const char* file,
            prefix = "Error:  ";
            prefixColorCode = ANSI_Colors::Red;
            break;
-        default:
-            break;
    };
-    bool showLocation = false;
+    auto debugOutput = debugOutputFields();
-    bool showProgram = true;
+    bool showLocation = debugOutput.isEmpty() ? false : caseInsensitiveContains("location", debugOutput);
+    bool showFunction = debugOutput.isEmpty() ? false : caseInsensitiveContains("function", debugOutput);
+    bool showProgram = debugOutput.isEmpty() ? false : caseInsensitiveContains("application", debugOutput);
+    bool useColor = true;
+    bool multiline = false;
    const QString resetColor = colorCommand(ANSI_Colors::Reset);
    QString output;
-    output += colorCommand(QList<int>() << ANSI_Colors::Bold << prefixColorCode) + prefix + resetColor;
+    if (useColor) {
+        output += colorCommand(QList<int>() << ANSI_Colors::Bold << prefixColorCode);
+    }
+    output += prefix;
+    if (useColor) {
+        output += resetColor;
+    }
    if (showProgram) {
-        output += QString(" %1(%2)").arg(QString::fromLatin1(programName)).arg(unsigned(getpid()));
+        int width = 10;
+        output += QString(" %1(%2)").arg(QString::fromLatin1(programName).leftJustified(width, ' ', true)).arg(unsigned(getpid())).rightJustified(width + 8, ' ');
    }
    if (debugArea) {
-        output += colorCommand(QList<int>() << ANSI_Colors::Bold << prefixColorCode) + QString(" %1 ").arg(QString::fromLatin1(debugArea)) + resetColor;
+        if (useColor) {
+            output += colorCommand(QList<int>() << ANSI_Colors::Bold << prefixColorCode);
+        }
+        output += QString(" %1 ").arg(QString::fromLatin1(debugArea).leftJustified(25, ' ', true));
+        if (useColor) {
+            output += resetColor;
+        }
+    }
+    if (showFunction) {
+        output += QString(" %3").arg(QString::fromLatin1(function).leftJustified(25, ' ', true));
    }
    if (showLocation) {
-        output += QString(" %3").arg(function);
+        const auto filename = QString::fromLatin1(file).split('/').last();
-        output += QString("%1:%2").arg(file).arg(line);
+        output += QString(" %1:%2").arg(filename.right(25)).arg(QString::number(line).leftJustified(4, ' ')).leftJustified(30, ' ', true);
+    }
+    if (multiline) {
+        output += "\n  ";
    }
    output += ":";
-    debug << output;
+    static DebugStream stream;
+    QDebug debug(&stream);
+    debug.noquote().nospace() << output;
-    return debug;
+    return debug.space().quote();
 }
diff --git a/common/log.h b/common/log.h
index 483f16f..415c7f7 100644
--- a/common/log.h
+++ b/common/log.h
@@ -1,5 +1,6 @@
 #pragma once
+#include "sink_export.h"
 #include <QDebug>
 namespace Sink {
@@ -12,19 +13,75 @@ enum DebugLevel {
    Error
 };
-QByteArray debugLevelName(DebugLevel debugLevel);
+QByteArray SINK_EXPORT debugLevelName(DebugLevel debugLevel);
-DebugLevel debugLevelFromName(const QByteArray &name);
+DebugLevel SINK_EXPORT debugLevelFromName(const QByteArray &name);
-void setDebugOutputLevel(DebugLevel);
+/**
-DebugLevel debugOutputLevel();
+ * Sets the debug output level.
+ *
+ * Everything below is ignored.
+ */
+void SINK_EXPORT setDebugOutputLevel(DebugLevel);
+DebugLevel SINK_EXPORT debugOutputLevel();
-QDebug debugStream(DebugLevel debugLevel, int line, const char* file, const char* function, const char* debugArea = 0);
+enum FilterType {
+    Area,
+    ApplicationName
+};
+/**
+ * Sets a debug output filter.
+ * 
+ * Everything that is not matching the filter is ignored.
+ * An empty filter matches everything.
+ *
+ * Note: In case of resources the application name is the identifier.
+ */
+void SINK_EXPORT setDebugOutputFilter(FilterType, const QByteArrayList &filter);
+QByteArrayList SINK_EXPORT debugOutputFilter(FilterType type);
+/**
+ * Set the debug output fields.
+ *
+ * Currently supported are:
+ * * Name: Application name used for filter.
+ * * Function: The function name:
+ * * Location: The source code location.
+ *
+ * These are additional items to the default ones (level, area, message).
+ */
+void SINK_EXPORT setDebugOutputFields(const QByteArrayList &filter);
+QByteArrayList SINK_EXPORT debugOutputFields();
+QDebug SINK_EXPORT debugStream(DebugLevel debugLevel, int line, const char* file, const char* function, const char* debugArea = 0);
+struct SINK_EXPORT TraceTime
+{
+    TraceTime(int i) : time(i){};
+    const int time;
+};
+inline QDebug SINK_EXPORT operator<<(QDebug d, const TraceTime &time)
+{
+    d << time.time << "[ms]";
+    return d;
+}
 }
 }
-#define Trace() Sink::Log::debugStream(Sink::Log::DebugLevel::Trace, __LINE__, __FILE__, Q_FUNC_INFO)
+#define DEBUG_AREA nullptr
-#define Log() Sink::Log::debugStream(Sink::Log::DebugLevel::Log, __LINE__, __FILE__, Q_FUNC_INFO)
-#define Warning() Sink::Log::debugStream(Sink::Log::DebugLevel::Warning, __LINE__, __FILE__, Q_FUNC_INFO)
+#define Trace_() Sink::Log::debugStream(Sink::Log::DebugLevel::Trace, __LINE__, __FILE__, Q_FUNC_INFO)
+#define Log_() Sink::Log::debugStream(Sink::Log::DebugLevel::Log, __LINE__, __FILE__, Q_FUNC_INFO)
+#define Trace_area(AREA) Sink::Log::debugStream(Sink::Log::DebugLevel::Trace, __LINE__, __FILE__, Q_FUNC_INFO, AREA)
+#define Log_area(AREA) Sink::Log::debugStream(Sink::Log::DebugLevel::Log, __LINE__, __FILE__, Q_FUNC_INFO, AREA)
+#define Warning_area(AREA) Sink::Log::debugStream(Sink::Log::DebugLevel::Warning, __LINE__, __FILE__, Q_FUNC_INFO, AREA)
+#define Error_area(AREA) Sink::Log::debugStream(Sink::Log::DebugLevel::Error, __LINE__, __FILE__, Q_FUNC_INFO, AREA)
+#define Trace() Sink::Log::debugStream(Sink::Log::DebugLevel::Trace, __LINE__, __FILE__, Q_FUNC_INFO, DEBUG_AREA)
+#define Log() Sink::Log::debugStream(Sink::Log::DebugLevel::Log, __LINE__, __FILE__, Q_FUNC_INFO, DEBUG_AREA)
+#define Warning() Sink::Log::debugStream(Sink::Log::DebugLevel::Warning, __LINE__, __FILE__, Q_FUNC_INFO, DEBUG_AREA)
 //FIXME Error clashes with Storage::Error and MessageQueue::Error
-#define ErrorMsg() Sink::Log::debugStream(Sink::Log::DebugLevel::Error, __LINE__, __FILE__, Q_FUNC_INFO)
+#define ErrorMsg() Sink::Log::debugStream(Sink::Log::DebugLevel::Error, __LINE__, __FILE__, Q_FUNC_INFO, DEBUG_AREA)
diff --git a/common/messagequeue.cpp b/common/messagequeue.cpp
index 1055922..73198a5 100644
--- a/common/messagequeue.cpp
+++ b/common/messagequeue.cpp
@@ -174,3 +174,7 @@ bool MessageQueue::isEmpty()
    return count == 0;
 }
+#pragma clang diagnostic push
+#pragma clang diagnostic ignored "-Wundefined-reinterpret-cast"
+#include "moc_messagequeue.cpp"
+#pragma clang diagnostic pop
diff --git a/common/messagequeue.h b/common/messagequeue.h
index 3206388..9399055 100644
--- a/common/messagequeue.h
+++ b/common/messagequeue.h
@@ -1,5 +1,6 @@
 #pragma once
+#include "sink_export.h"
 #include <QObject>
 #include <QByteArrayList>
 #include <string>
@@ -11,7 +12,7 @@
 /**
 * A persistent FIFO message queue.
 */
-class MessageQueue : public QObject
+class SINK_EXPORT MessageQueue : public QObject
 {
    Q_OBJECT
 public:
diff --git a/common/modelresult.cpp b/common/modelresult.cpp
index 3a9fb95..f28c665 100644
--- a/common/modelresult.cpp
+++ b/common/modelresult.cpp
@@ -24,6 +24,9 @@
 #include "domain/folder.h"
 #include "log.h"
+#undef DEBUG_AREA
+#define DEBUG_AREA "client.modelresult"
 static  uint qHash(const Sink::ApplicationDomain::ApplicationDomainType &type)
 {
    Q_ASSERT(!type.resourceInstanceIdentifier().isEmpty());
diff --git a/common/notification.h b/common/notification.h
index ae24bd2..cf69a99 100644
--- a/common/notification.h
+++ b/common/notification.h
@@ -19,7 +19,7 @@
 */
 #pragma once
-#include <sinkcommon_export.h>
+#include "sink_export.h"
 #include <QString>
 namespace Sink
@@ -28,7 +28,7 @@ namespace Sink
 /**
 * A notification
 */
-class SINKCOMMON_EXPORT Notification
+class SINK_EXPORT Notification
 {
 public:
    QByteArray id;
diff --git a/common/notifier.cpp b/common/notifier.cpp
new file mode 100644
index 0000000..e4248df
--- /dev/null
+++ b/common/notifier.cpp
@@ -0,0 +1,69 @@
+/*
+ * Copyright (C) 2015 Christian Mollekopf <chrigi_1@fastmail.fm>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) version 3, or any
+ * later version accepted by the membership of KDE e.V. (or its
+ * successor approved by the membership of KDE e.V.), which shall
+ * act as a proxy defined in Section 6 of version 3 of the license.
+ *
+ * This library 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
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library.  If not, see <http://www.gnu.org/licenses/>.
+ */
+#include "notifier.h"
+#include <functional>
+#include "resourceaccess.h"
+#include "log.h"
+using namespace Sink;
+class Sink::Notifier::Private {
+public:
+    Private()
+        : context(new QObject)
+    {
+    }
+    QList<QSharedPointer<ResourceAccess> > resourceAccess;
+    QList<std::function<void(const Notification &)> > handler;
+    QSharedPointer<QObject> context;
+};
+Notifier::Notifier(const QSharedPointer<ResourceAccess> &resourceAccess)
+    : d(new Sink::Notifier::Private)
+{
+    QObject::connect(resourceAccess.data(), &ResourceAccess::notification, d->context.data(), [this](const Notification &notification) {
+        for (const auto &handler : d->handler) {
+            handler(notification);
+        }
+    });
+    d->resourceAccess << resourceAccess;
+}
+Notifier::Notifier(const QByteArray &instanceIdentifier)
+    : d(new Sink::Notifier::Private)
+{
+    auto resourceAccess = Sink::ResourceAccess::Ptr::create(instanceIdentifier);
+    resourceAccess->open();
+    QObject::connect(resourceAccess.data(), &ResourceAccess::notification, d->context.data(), [this](const Notification &notification) {
+        for (const auto &handler : d->handler) {
+            handler(notification);
+        }
+    });
+    d->resourceAccess << resourceAccess;
+}
+void Notifier::registerHandler(std::function<void(const Notification &)> handler)
+{
+    d->handler << handler;
+}
diff --git a/common/listmodelresult.cpp b/common/notifier.h
index 6ef1c5f..d16a311 100644
--- a/common/listmodelresult.cpp
+++ b/common/notifier.h
@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2014 Christian Mollekopf <chrigi_1@fastmail.fm>
+ * Copyright (C) 2015 Christian Mollekopf <chrigi_1@fastmail.fm>
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
@@ -18,4 +18,29 @@
 * License along with this library.  If not, see <http://www.gnu.org/licenses/>.
 */
-#include "listmodelresult.h"
+#pragma once
+#include "sink_export.h"
+#include <QByteArray>
+#include <QSharedPointer>
+#include <Async/Async>
+class QAbstractItemModel;
+namespace Sink {
+class ResourceAccess;
+class Notification;
+class SINK_EXPORT Notifier {
+public:
+    Notifier(const QSharedPointer<ResourceAccess> &resourceAccess);
+    Notifier(const QByteArray &resourceInstanceIdentifier);
+    void registerHandler(std::function<void(const Notification &)>);
+private:
+    class Private;
+    QSharedPointer<Private> d;
+};
+}
diff --git a/common/pipeline.cpp b/common/pipeline.cpp
index 401c26d..93d8236 100644
--- a/common/pipeline.cpp
+++ b/common/pipeline.cpp
@@ -25,6 +25,7 @@
 #include <QVector>
 #include <QUuid>
 #include <QDebug>
+#include <QTime>
 #include "entity_generated.h"
 #include "metadata_generated.h"
 #include "createentity_generated.h"
@@ -36,6 +37,9 @@
 #include "definitions.h"
 #include "bufferutils.h"
+#undef DEBUG_AREA
+#define DEBUG_AREA "resource.pipeline"
 namespace Sink
 {
@@ -53,8 +57,24 @@ public:
    QHash<QString, QVector<Preprocessor *> > processors;
    QHash<QString, DomainTypeAdaptorFactoryInterface::Ptr> adaptorFactory;
    bool revisionChanged;
+    void storeNewRevision(qint64 newRevision, const flatbuffers::FlatBufferBuilder &fbb, const QByteArray &bufferType, const QByteArray &uid);
+    QTime transactionTime;
+    int transactionItemCount;
 };
+void Pipeline::Private::storeNewRevision(qint64 newRevision, const flatbuffers::FlatBufferBuilder &fbb, const QByteArray &bufferType, const QByteArray &uid)
+{
+    Storage::mainDatabase(transaction, bufferType).write(Storage::assembleKey(uid, newRevision), BufferUtils::extractBuffer(fbb),
+        [uid, newRevision](const Storage::Error &error) {
+            Warning() << "Failed to write entity" << uid << newRevision;
+        }
+    );
+    revisionChanged = true;
+    Storage::setMaxRevision(transaction, newRevision);
+    Storage::recordRevision(transaction, newRevision, uid, bufferType);
+}
 Pipeline::Pipeline(const QString &resourceName, QObject *parent)
    : QObject(parent),
      d(new Private(resourceName))
@@ -86,7 +106,10 @@ void Pipeline::startTransaction()
    if (d->transaction) {
        return;
    }
-    d->transaction = std::move(storage().createTransaction(Sink::Storage::ReadWrite));
+    Trace() << "Starting transaction.";
+    d->transactionTime.start();
+    d->transactionItemCount = 0;
+    d->transaction = std::move(storage().createTransaction(Storage::ReadWrite));
 }
 void Pipeline::commit()
@@ -96,8 +119,9 @@ void Pipeline::commit()
    // for (auto processor : d->processors[bufferType]) {
    //     processor->finalize();
    // }
-    const auto revision = Sink::Storage::maxRevision(d->transaction);
+    const auto revision = Storage::maxRevision(d->transaction);
-    Trace() << "Committing " << revision;
+    const auto elapsed = d->transactionTime.elapsed();
+    Trace() << "Committing revision: " << revision << ":" << d->transactionItemCount << " items in: " << Log::TraceTime(elapsed) << " " << (double)elapsed/(double)qMax(d->transactionItemCount, 1) << "[ms/item]";
    if (d->transaction) {
        d->transaction.commit();
    }
@@ -118,41 +142,30 @@ Storage &Pipeline::storage() const
    return d->storage;
 }
-void Pipeline::storeNewRevision(qint64 newRevision, const flatbuffers::FlatBufferBuilder &fbb, const QByteArray &bufferType, const QByteArray &uid)
-{
-    d->transaction.openDatabase(bufferType + ".main").write(Sink::Storage::assembleKey(uid, newRevision), BufferUtils::extractBuffer(fbb),
-        [](const Sink::Storage::Error &error) {
-            Warning() << "Failed to write entity";
-        }
-    );
-    d->revisionChanged = true;
-    Sink::Storage::setMaxRevision(d->transaction, newRevision);
-    Sink::Storage::recordRevision(d->transaction, newRevision, uid, bufferType);
-}
 KAsync::Job<qint64> Pipeline::newEntity(void const *command, size_t size)
 {
    Trace() << "Pipeline: New Entity";
+    d->transactionItemCount++;
    {
        flatbuffers::Verifier verifyer(reinterpret_cast<const uint8_t *>(command), size);
-        if (!Sink::Commands::VerifyCreateEntityBuffer(verifyer)) {
+        if (!Commands::VerifyCreateEntityBuffer(verifyer)) {
            Warning() << "invalid buffer, not a create entity buffer";
            return KAsync::error<qint64>(0);
        }
    }
-    auto createEntity = Sink::Commands::GetCreateEntity(command);
+    auto createEntity = Commands::GetCreateEntity(command);
    const bool replayToSource = createEntity->replayToSource();
    const QByteArray bufferType = QByteArray(reinterpret_cast<char const*>(createEntity->domainType()->Data()), createEntity->domainType()->size());
    {
        flatbuffers::Verifier verifyer(reinterpret_cast<const uint8_t *>(createEntity->delta()->Data()), createEntity->delta()->size());
-        if (!Sink::VerifyEntityBuffer(verifyer)) {
+        if (!VerifyEntityBuffer(verifyer)) {
            Warning() << "invalid buffer, not an entity buffer";
            return KAsync::error<qint64>(0);
        }
    }
-    auto entity = Sink::GetEntity(createEntity->delta()->Data());
+    auto entity = GetEntity(createEntity->delta()->Data());
    if (!entity->resource()->size() && !entity->local()->size()) {
        Warning() << "No local and no resource buffer while trying to create entity.";
        return KAsync::error<qint64>(0);
@@ -161,7 +174,7 @@ KAsync::Job<qint64> Pipeline::newEntity(void const *command, size_t size)
    QByteArray key;
    if (createEntity->entityId()) {
        key = QByteArray(reinterpret_cast<char const*>(createEntity->entityId()->Data()), createEntity->entityId()->size());
-        if (d->transaction.openDatabase(bufferType + ".main").contains(key)) {
+        if (Storage::mainDatabase(d->transaction, bufferType).contains(key)) {
            ErrorMsg() << "An entity with this id already exists: " << key;
            return KAsync::error<qint64>(0);
        }
@@ -171,21 +184,21 @@ KAsync::Job<qint64> Pipeline::newEntity(void const *command, size_t size)
        key = QUuid::createUuid().toString().toUtf8();
    }
    Q_ASSERT(!key.isEmpty());
-    const qint64 newRevision = Sink::Storage::maxRevision(d->transaction) + 1;
+    const qint64 newRevision = Storage::maxRevision(d->transaction) + 1;
    //Add metadata buffer
    flatbuffers::FlatBufferBuilder metadataFbb;
-    auto metadataBuilder = Sink::MetadataBuilder(metadataFbb);
+    auto metadataBuilder = MetadataBuilder(metadataFbb);
    metadataBuilder.add_revision(newRevision);
-    metadataBuilder.add_operation(Sink::Operation_Creation);
+    metadataBuilder.add_operation(Operation_Creation);
    metadataBuilder.add_replayToSource(replayToSource);
    auto metadataBuffer = metadataBuilder.Finish();
-    Sink::FinishMetadataBuffer(metadataFbb, metadataBuffer);
+    FinishMetadataBuffer(metadataFbb, metadataBuffer);
    flatbuffers::FlatBufferBuilder fbb;
    EntityBuffer::assembleEntityBuffer(fbb, metadataFbb.GetBufferPointer(), metadataFbb.GetSize(), entity->resource()->Data(), entity->resource()->size(), entity->local()->Data(), entity->local()->size());
-    storeNewRevision(newRevision, fbb, bufferType, key);
+    d->storeNewRevision(newRevision, fbb, bufferType, key);
    auto adaptorFactory = d->adaptorFactory.value(bufferType);
    if (!adaptorFactory) {
@@ -194,14 +207,14 @@ KAsync::Job<qint64> Pipeline::newEntity(void const *command, size_t size)
    }
    Log() << "Pipeline: wrote entity: " << key << newRevision << bufferType;
-    d->transaction.openDatabase(bufferType + ".main").scan(Sink::Storage::assembleKey(key, newRevision), [this, bufferType, newRevision, adaptorFactory, key](const QByteArray &, const QByteArray &value) -> bool {
+    Storage::mainDatabase(d->transaction, bufferType).scan(Storage::assembleKey(key, newRevision), [this, bufferType, newRevision, adaptorFactory, key](const QByteArray &, const QByteArray &value) -> bool {
-        auto entity = Sink::GetEntity(value);
+        auto entity = GetEntity(value);
        auto adaptor = adaptorFactory->createAdaptor(*entity);
        for (auto processor : d->processors[bufferType]) {
            processor->newEntity(key, newRevision, *adaptor, d->transaction);
        }
        return false;
-    }, [this](const Sink::Storage::Error &error) {
+    }, [this](const Storage::Error &error) {
        ErrorMsg() << "Failed to find value in pipeline: " << error.message;
    });
    return KAsync::start<qint64>([newRevision](){
@@ -212,17 +225,18 @@ KAsync::Job<qint64> Pipeline::newEntity(void const *command, size_t size)
 KAsync::Job<qint64> Pipeline::modifiedEntity(void const *command, size_t size)
 {
    Trace() << "Pipeline: Modified Entity";
+    d->transactionItemCount++;
-    const qint64 newRevision = Sink::Storage::maxRevision(d->transaction) + 1;
+    const qint64 newRevision = Storage::maxRevision(d->transaction) + 1;
    {
        flatbuffers::Verifier verifyer(reinterpret_cast<const uint8_t *>(command), size);
-        if (!Sink::Commands::VerifyModifyEntityBuffer(verifyer)) {
+        if (!Commands::VerifyModifyEntityBuffer(verifyer)) {
            Warning() << "invalid buffer, not a modify entity buffer";
            return KAsync::error<qint64>(0);
        }
    }
-    auto modifyEntity = Sink::Commands::GetModifyEntity(command);
+    auto modifyEntity = Commands::GetModifyEntity(command);
    Q_ASSERT(modifyEntity);
    const qint64 baseRevision = modifyEntity->revision();
@@ -236,7 +250,7 @@ KAsync::Job<qint64> Pipeline::modifiedEntity(void const *command, size_t size)
    }
    {
        flatbuffers::Verifier verifyer(reinterpret_cast<const uint8_t *>(modifyEntity->delta()->Data()), modifyEntity->delta()->size());
-        if (!Sink::VerifyEntityBuffer(verifyer)) {
+        if (!VerifyEntityBuffer(verifyer)) {
            Warning() << "invalid buffer, not an entity buffer";
            return KAsync::error<qint64>(0);
        }
@@ -249,13 +263,13 @@ KAsync::Job<qint64> Pipeline::modifiedEntity(void const *command, size_t size)
        return KAsync::error<qint64>(0);
    }
-    auto diffEntity = Sink::GetEntity(modifyEntity->delta()->Data());
+    auto diffEntity = GetEntity(modifyEntity->delta()->Data());
    Q_ASSERT(diffEntity);
    auto diff = adaptorFactory->createAdaptor(*diffEntity);
-    QSharedPointer<Sink::ApplicationDomain::BufferAdaptor> current;
+    QSharedPointer<ApplicationDomain::BufferAdaptor> current;
-    d->transaction.openDatabase(bufferType + ".main").findLatest(key, [&current, adaptorFactory](const QByteArray &key, const QByteArray &data) -> bool {
+    Storage::mainDatabase(d->transaction, bufferType).findLatest(key, [&current, adaptorFactory](const QByteArray &key, const QByteArray &data) -> bool {
-        Sink::EntityBuffer buffer(const_cast<const char *>(data.data()), data.size());
+        EntityBuffer buffer(const_cast<const char *>(data.data()), data.size());
        if (!buffer.isValid()) {
            Warning() << "Read invalid buffer from disk";
        } else {
@@ -273,15 +287,22 @@ KAsync::Job<qint64> Pipeline::modifiedEntity(void const *command, size_t size)
    }
    //resource and uid don't matter at this point
-    const Sink::ApplicationDomain::ApplicationDomainType existingObject("", "", newRevision, current);
+    const ApplicationDomain::ApplicationDomainType existingObject("", "", newRevision, current);
-    auto newObject = Sink::ApplicationDomain::ApplicationDomainType::getInMemoryRepresentation<Sink::ApplicationDomain::ApplicationDomainType>(existingObject);
+    auto newObject = ApplicationDomain::ApplicationDomainType::getInMemoryRepresentation<ApplicationDomain::ApplicationDomainType>(existingObject);
    //Apply diff
    //FIXME only apply the properties that are available in the buffer
    Trace() << "Applying changed properties: " << diff->availableProperties();
+    QSet<QByteArray> changeset;
    for (const auto &property : diff->availableProperties()) {
-        newObject->setProperty(property, diff->getProperty(property));
+        changeset << property;
+        const auto value = diff->getProperty(property);
+        if (value.isValid()) {
+            newObject->setProperty(property, value);
+        }
    }
+    //Altough we only set some properties, we want all to be serialized
+    newObject->setChangedProperties(changeset);
    //Remove deletions
    if (modifyEntity->deletions()) {
@@ -292,26 +313,29 @@ KAsync::Job<qint64> Pipeline::modifiedEntity(void const *command, size_t size)
    //Add metadata buffer
    flatbuffers::FlatBufferBuilder metadataFbb;
-    auto metadataBuilder = Sink::MetadataBuilder(metadataFbb);
+    auto metadataBuilder = MetadataBuilder(metadataFbb);
    metadataBuilder.add_revision(newRevision);
-    metadataBuilder.add_operation(Sink::Operation_Modification);
+    metadataBuilder.add_operation(Operation_Modification);
    metadataBuilder.add_replayToSource(replayToSource);
    auto metadataBuffer = metadataBuilder.Finish();
-    Sink::FinishMetadataBuffer(metadataFbb, metadataBuffer);
+    FinishMetadataBuffer(metadataFbb, metadataBuffer);
    flatbuffers::FlatBufferBuilder fbb;
    adaptorFactory->createBuffer(*newObject, fbb, metadataFbb.GetBufferPointer(), metadataFbb.GetSize());
-    storeNewRevision(newRevision, fbb, bufferType, key);
+    d->storeNewRevision(newRevision, fbb, bufferType, key);
    Log() << "Pipeline: modified entity: " << key << newRevision << bufferType;
-    d->transaction.openDatabase(bufferType + ".main").scan(Sink::Storage::assembleKey(key, newRevision), [this, bufferType, newRevision, adaptorFactory, current, key](const QByteArray &, const QByteArray &value) -> bool {
+    Storage::mainDatabase(d->transaction, bufferType).scan(Storage::assembleKey(key, newRevision), [this, bufferType, newRevision, adaptorFactory, current, key](const QByteArray &k, const QByteArray &value) -> bool {
-        auto entity = Sink::GetEntity(value);
+        if (value.isEmpty()) {
+            ErrorMsg() << "Read buffer is empty.";
+        }
+        auto entity = GetEntity(value.data());
        auto newEntity = adaptorFactory->createAdaptor(*entity);
        for (auto processor : d->processors[bufferType]) {
            processor->modifiedEntity(key, newRevision, *current, *newEntity, d->transaction);
        }
        return false;
-    }, [this](const Sink::Storage::Error &error) {
+    }, [this](const Storage::Error &error) {
        ErrorMsg() << "Failed to find value in pipeline: " << error.message;
    });
    return KAsync::start<qint64>([newRevision](){
@@ -322,15 +346,16 @@ KAsync::Job<qint64> Pipeline::modifiedEntity(void const *command, size_t size)
 KAsync::Job<qint64> Pipeline::deletedEntity(void const *command, size_t size)
 {
    Trace() << "Pipeline: Deleted Entity";
+    d->transactionItemCount++;
    {
        flatbuffers::Verifier verifyer(reinterpret_cast<const uint8_t *>(command), size);
-        if (!Sink::Commands::VerifyDeleteEntityBuffer(verifyer)) {
+        if (!Commands::VerifyDeleteEntityBuffer(verifyer)) {
            Warning() << "invalid buffer, not a delete entity buffer";
            return KAsync::error<qint64>(0);
        }
    }
-    auto deleteEntity = Sink::Commands::GetDeleteEntity(command);
+    auto deleteEntity = Commands::GetDeleteEntity(command);
    const bool replayToSource = deleteEntity->replayToSource();
    const QByteArray bufferType = QByteArray(reinterpret_cast<char const*>(deleteEntity->domainType()->Data()), deleteEntity->domainType()->size());
@@ -338,13 +363,12 @@ KAsync::Job<qint64> Pipeline::deletedEntity(void const *command, size_t size)
    bool found = false;
    bool alreadyRemoved = false;
-    d->transaction.openDatabase(bufferType + ".main").findLatest(key, [&found, &alreadyRemoved](const QByteArray &key, const QByteArray &data) -> bool {
+    Storage::mainDatabase(d->transaction, bufferType).findLatest(key, [&found, &alreadyRemoved](const QByteArray &key, const QByteArray &data) -> bool {
-        Sink::EntityBuffer buffer(const_cast<const char *>(data.data()), data.size());
+        auto entity = GetEntity(data.data());
-        auto entity = Sink::GetEntity(data.data());
        if (entity && entity->metadata()) {
-            auto metadata = Sink::GetMetadata(entity->metadata()->Data());
+            auto metadata = GetMetadata(entity->metadata()->Data());
            found = true;
-            if (metadata->operation() == Sink::Operation_Removal) {
+            if (metadata->operation() == Operation_Removal) {
                alreadyRemoved = true;
            }
@@ -364,16 +388,16 @@ KAsync::Job<qint64> Pipeline::deletedEntity(void const *command, size_t size)
        return KAsync::error<qint64>(0);
    }
-    const qint64 newRevision = Sink::Storage::maxRevision(d->transaction) + 1;
+    const qint64 newRevision = Storage::maxRevision(d->transaction) + 1;
    //Add metadata buffer
    flatbuffers::FlatBufferBuilder metadataFbb;
-    auto metadataBuilder = Sink::MetadataBuilder(metadataFbb);
+    auto metadataBuilder = MetadataBuilder(metadataFbb);
    metadataBuilder.add_revision(newRevision);
-    metadataBuilder.add_operation(Sink::Operation_Removal);
+    metadataBuilder.add_operation(Operation_Removal);
    metadataBuilder.add_replayToSource(replayToSource);
    auto metadataBuffer = metadataBuilder.Finish();
-    Sink::FinishMetadataBuffer(metadataFbb, metadataBuffer);
+    FinishMetadataBuffer(metadataFbb, metadataBuffer);
    flatbuffers::FlatBufferBuilder fbb;
    EntityBuffer::assembleEntityBuffer(fbb, metadataFbb.GetBufferPointer(), metadataFbb.GetSize(), 0, 0, 0, 0);
@@ -384,20 +408,20 @@ KAsync::Job<qint64> Pipeline::deletedEntity(void const *command, size_t size)
        return KAsync::error<qint64>(0);
    }
-    QSharedPointer<Sink::ApplicationDomain::BufferAdaptor> current;
+    QSharedPointer<ApplicationDomain::BufferAdaptor> current;
-    d->transaction.openDatabase(bufferType + ".main").findLatest(key, [this, bufferType, newRevision, adaptorFactory, key, &current](const QByteArray &, const QByteArray &data) -> bool {
+    Storage::mainDatabase(d->transaction, bufferType).findLatest(key, [this, bufferType, newRevision, adaptorFactory, key, &current](const QByteArray &, const QByteArray &data) -> bool {
-        Sink::EntityBuffer buffer(const_cast<const char *>(data.data()), data.size());
+        EntityBuffer buffer(const_cast<const char *>(data.data()), data.size());
        if (!buffer.isValid()) {
            Warning() << "Read invalid buffer from disk";
        } else {
            current = adaptorFactory->createAdaptor(buffer.entity());
        }
        return false;
-    }, [this](const Sink::Storage::Error &error) {
+    }, [this](const Storage::Error &error) {
        ErrorMsg() << "Failed to find value in pipeline: " << error.message;
    });
-    storeNewRevision(newRevision, fbb, bufferType, key);
+    d->storeNewRevision(newRevision, fbb, bufferType, key);
    Log() << "Pipeline: deleted entity: "<< newRevision;
    for (auto processor : d->processors[bufferType]) {
@@ -411,33 +435,33 @@ KAsync::Job<qint64> Pipeline::deletedEntity(void const *command, size_t size)
 void Pipeline::cleanupRevision(qint64 revision)
 {
-    const auto uid = Sink::Storage::getUidFromRevision(d->transaction, revision);
+    const auto uid = Storage::getUidFromRevision(d->transaction, revision);
-    const auto bufferType = Sink::Storage::getTypeFromRevision(d->transaction, revision);
+    const auto bufferType = Storage::getTypeFromRevision(d->transaction, revision);
    Trace() << "Cleaning up revision " << revision << uid << bufferType;
-    d->transaction.openDatabase(bufferType + ".main").scan(uid, [&](const QByteArray &key, const QByteArray &data) -> bool {
+    Storage::mainDatabase(d->transaction, bufferType).scan(uid, [&](const QByteArray &key, const QByteArray &data) -> bool {
-        Sink::EntityBuffer buffer(const_cast<const char *>(data.data()), data.size());
+        EntityBuffer buffer(const_cast<const char *>(data.data()), data.size());
        if (!buffer.isValid()) {
            Warning() << "Read invalid buffer from disk";
        } else {
-            const auto metadata =  flatbuffers::GetRoot<Sink::Metadata>(buffer.metadataBuffer());
+            const auto metadata =  flatbuffers::GetRoot<Metadata>(buffer.metadataBuffer());
            const qint64 rev = metadata->revision();
            //Remove old revisions, and the current if the entity has already been removed
-            if (rev < revision || metadata->operation() == Sink::Operation_Removal) {
+            if (rev < revision || metadata->operation() == Operation_Removal) {
-                Sink::Storage::removeRevision(d->transaction, rev);
+                Storage::removeRevision(d->transaction, rev);
-                d->transaction.openDatabase(bufferType + ".main").remove(key);
+                Storage::mainDatabase(d->transaction, bufferType).remove(key);
            }
        }
        return true;
-    }, [](const Sink::Storage::Error &error) {
+    }, [](const Storage::Error &error) {
        Warning() << "Error while reading: " << error.message;
    }, true);
-    Sink::Storage::setCleanedUpRevision(d->transaction, revision);
+    Storage::setCleanedUpRevision(d->transaction, revision);
 }
 qint64 Pipeline::cleanedUpRevision()
 {
-    return Sink::Storage::cleanedUpRevision(d->transaction);
+    return Storage::cleanedUpRevision(d->transaction);
 }
 Preprocessor::Preprocessor()
@@ -459,3 +483,7 @@ void Preprocessor::finalize()
 } // namespace Sink
+#pragma clang diagnostic push
+#pragma clang diagnostic ignored "-Wundefined-reinterpret-cast"
+#include "moc_pipeline.cpp"
+#pragma clang diagnostic pop
diff --git a/common/pipeline.h b/common/pipeline.h
index 60a5fa5..096771f 100644
--- a/common/pipeline.h
+++ b/common/pipeline.h
@@ -25,7 +25,7 @@
 #include <QSharedDataPointer>
 #include <QObject>
-#include <sinkcommon_export.h>
+#include "sink_export.h"
 #include <storage.h>
 #include <Async/Async>
@@ -37,7 +37,7 @@ namespace Sink
 class Preprocessor;
-class SINKCOMMON_EXPORT Pipeline : public QObject
+class SINK_EXPORT Pipeline : public QObject
 {
    Q_OBJECT
@@ -73,13 +73,11 @@ Q_SIGNALS:
    void revisionUpdated(qint64);
 private:
-    void storeNewRevision(qint64 newRevision, const flatbuffers::FlatBufferBuilder &fbb, const QByteArray &bufferType, const QByteArray &uid);
    class Private;
    Private * const d;
 };
-class SINKCOMMON_EXPORT Preprocessor
+class SINK_EXPORT Preprocessor
 {
 public:
    Preprocessor();
diff --git a/common/propertymapper.h b/common/propertymapper.h
index efde72c..57202ab 100644
--- a/common/propertymapper.h
+++ b/common/propertymapper.h
@@ -19,6 +19,7 @@
 #pragma once
+#include "sink_export.h"
 #include <QVariant>
 #include <QByteArray>
 #include <functional>
@@ -28,17 +29,17 @@
 * Defines how to convert qt primitives to flatbuffer ones
 */
 template <class T>
-flatbuffers::uoffset_t variantToProperty(const QVariant &, flatbuffers::FlatBufferBuilder &fbb);
+flatbuffers::uoffset_t SINK_EXPORT variantToProperty(const QVariant &, flatbuffers::FlatBufferBuilder &fbb);
 /**
 * Defines how to convert flatbuffer primitives to qt ones
 */
 template <typename T>
-QVariant propertyToVariant(const flatbuffers::String *);
+QVariant SINK_EXPORT propertyToVariant(const flatbuffers::String *);
 template <typename T>
-QVariant propertyToVariant(uint8_t);
+QVariant SINK_EXPORT propertyToVariant(uint8_t);
 template <typename T>
-QVariant propertyToVariant(const flatbuffers::Vector<uint8_t> *);
+QVariant SINK_EXPORT propertyToVariant(const flatbuffers::Vector<uint8_t> *);
 /**
@@ -52,6 +53,8 @@ template<typename BufferType>
 class ReadPropertyMapper
 {
 public:
+    virtual ~ReadPropertyMapper(){};
    virtual QVariant getProperty(const QByteArray &key, BufferType const *buffer) const
    {
        if (mReadAccessors.contains(key)) {
@@ -106,6 +109,8 @@ template<typename BufferBuilder>
 class WritePropertyMapper
 {
 public:
+    virtual ~WritePropertyMapper(){};
    virtual void setProperty(const QByteArray &key, const QVariant &value, QList<std::function<void(BufferBuilder &)> > &builderCalls, flatbuffers::FlatBufferBuilder &fbb) const
    {
        if (mWriteAccessors.contains(key)) {
diff --git a/common/queryrunner.cpp b/common/queryrunner.cpp
index 25d69b1..22682d3 100644
--- a/common/queryrunner.cpp
+++ b/common/queryrunner.cpp
@@ -26,6 +26,9 @@
 #include "domainadaptor.h"
 #include "asyncutils.h"
+#undef DEBUG_AREA
+#define DEBUG_AREA "client.queryrunner"
 using namespace Sink;
 /*
@@ -38,14 +41,14 @@ template<typename DomainType>
 class QueryWorker : public QObject
 {
 public:
-    QueryWorker(const Sink::Query &query, const QByteArray &instanceIdentifier, const DomainTypeAdaptorFactoryInterface::Ptr &, const QByteArray &bufferType);
+    QueryWorker(const Sink::Query &query, const QByteArray &instanceIdentifier, const DomainTypeAdaptorFactoryInterface::Ptr &, const QByteArray &bufferType, const QueryRunnerBase::ResultTransformation &transformation);
    virtual ~QueryWorker();
    qint64 executeIncrementalQuery(const Sink::Query &query, Sink::ResultProviderInterface<typename DomainType::Ptr> &resultProvider);
    qint64 executeInitialQuery(const Sink::Query &query, const typename DomainType::Ptr &parent, Sink::ResultProviderInterface<typename DomainType::Ptr> &resultProvider);
 private:
-    static void replaySet(ResultSet &resultSet, Sink::ResultProviderInterface<typename DomainType::Ptr> &resultProvider, const QList<QByteArray> &properties);
+    void replaySet(ResultSet &resultSet, Sink::ResultProviderInterface<typename DomainType::Ptr> &resultProvider, const QList<QByteArray> &properties);
    void readEntity(const Sink::Storage::NamedDatabase &db, const QByteArray &key, const std::function<void(const Sink::ApplicationDomain::ApplicationDomainType::Ptr &, Sink::Operation)> &resultCallback);
@@ -57,6 +60,7 @@ private:
    qint64 load(const Sink::Query &query, const std::function<ResultSet(Sink::Storage::Transaction &, QSet<QByteArray> &)> &baseSetRetriever, Sink::ResultProviderInterface<typename DomainType::Ptr> &resultProvider, bool initialQuery);
 private:
+    QueryRunnerBase::ResultTransformation mResultTransformation;
    DomainTypeAdaptorFactoryInterface::Ptr mDomainTypeAdaptorFactory;
    QByteArray mResourceInstanceIdentifier;
    QByteArray mBufferType;
@@ -72,11 +76,11 @@ QueryRunner<DomainType>::QueryRunner(const Sink::Query &query, const Sink::Resou
 {
    Trace() << "Starting query";
    //We delegate loading of initial data to the result provider, os it can decide for itself what it needs to load.
-    mResultProvider->setFetcher([this, query, instanceIdentifier, factory, bufferType](const typename DomainType::Ptr &parent) {
+    mResultProvider->setFetcher([=](const typename DomainType::Ptr &parent) {
        Trace() << "Running fetcher";
        auto resultProvider = mResultProvider;
-        async::run<qint64>([query, instanceIdentifier, factory, bufferType, parent, resultProvider]() -> qint64 {
+        async::run<qint64>([=]() -> qint64 {
-                QueryWorker<DomainType> worker(query, instanceIdentifier, factory, bufferType);
+                QueryWorker<DomainType> worker(query, instanceIdentifier, factory, bufferType, mResultTransformation);
                const qint64 newRevision = worker.executeInitialQuery(query, parent, *resultProvider);
                return newRevision;
            })
@@ -91,18 +95,17 @@ QueryRunner<DomainType>::QueryRunner(const Sink::Query &query, const Sink::Resou
    // In case of a live query we keep the runner for as long alive as the result provider exists
    if (query.liveQuery) {
        //Incremental updates are always loaded directly, leaving it up to the result to discard the changes if they are not interesting
-        setQuery([this, query, instanceIdentifier, factory, bufferType] () -> KAsync::Job<void> {
+        setQuery([=] () -> KAsync::Job<void> {
            auto resultProvider = mResultProvider;
-            return async::run<qint64>([query, instanceIdentifier, factory, bufferType, resultProvider]() -> qint64 {
+            return async::run<qint64>([=]() -> qint64 {
-                    QueryWorker<DomainType> worker(query, instanceIdentifier, factory, bufferType);
+                    QueryWorker<DomainType> worker(query, instanceIdentifier, factory, bufferType, mResultTransformation);
                    const qint64 newRevision = worker.executeIncrementalQuery(query, *resultProvider);
                    return newRevision;
                })
                .template then<void, qint64>([query, this](qint64 newRevision) {
                    //Only send the revision replayed information if we're connected to the resource, there's no need to start the resource otherwise.
                    mResourceAccess->sendRevisionReplayedCommand(newRevision);
-                })
+                });
-                .template then<void>([](){});
        });
        //Ensure the connection is open, if it wasn't already opened
        //TODO If we are not connected already, we have to check for the latest revision once connected, otherwise we could miss some updates
@@ -118,6 +121,12 @@ QueryRunner<DomainType>::~QueryRunner()
 }
 template<class DomainType>
+void QueryRunner<DomainType>::setResultTransformation(const ResultTransformation &transformation)
+{
+    mResultTransformation = transformation;
+}
+template<class DomainType>
 typename Sink::ResultEmitter<typename DomainType::Ptr>::Ptr QueryRunner<DomainType>::emitter()
 {
    return mResultProvider->emitter();
@@ -129,7 +138,7 @@ static inline ResultSet fullScan(const Sink::Storage::Transaction &transaction,
 {
    //TODO use a result set with an iterator, to read values on demand
    QVector<QByteArray> keys;
-    transaction.openDatabase(bufferType + ".main").scan(QByteArray(), [&](const QByteArray &key, const QByteArray &value) -> bool {
+    Storage::mainDatabase(transaction, bufferType).scan(QByteArray(), [&](const QByteArray &key, const QByteArray &value) -> bool {
        //Skip internals
        if (Sink::Storage::isInternalKey(key)) {
            return true;
@@ -147,8 +156,9 @@ static inline ResultSet fullScan(const Sink::Storage::Transaction &transaction,
 template<class DomainType>
-QueryWorker<DomainType>::QueryWorker(const Sink::Query &query, const QByteArray &instanceIdentifier, const DomainTypeAdaptorFactoryInterface::Ptr &factory, const QByteArray &bufferType)
+QueryWorker<DomainType>::QueryWorker(const Sink::Query &query, const QByteArray &instanceIdentifier, const DomainTypeAdaptorFactoryInterface::Ptr &factory, const QByteArray &bufferType, const QueryRunnerBase::ResultTransformation &transformation)
    : QObject(),
+    mResultTransformation(transformation),
    mDomainTypeAdaptorFactory(factory),
    mResourceInstanceIdentifier(instanceIdentifier),
    mBufferType(bufferType),
@@ -167,20 +177,25 @@ template<class DomainType>
 void QueryWorker<DomainType>::replaySet(ResultSet &resultSet, Sink::ResultProviderInterface<typename DomainType::Ptr> &resultProvider, const QList<QByteArray> &properties)
 {
    int counter = 0;
-    while (resultSet.next([&resultProvider, &counter, &properties](const Sink::ApplicationDomain::ApplicationDomainType::Ptr &value, Sink::Operation operation) -> bool {
+    while (resultSet.next([this, &resultProvider, &counter, &properties](const Sink::ApplicationDomain::ApplicationDomainType::Ptr &value, Sink::Operation operation) -> bool {
+        //FIXME allow maildir resource to set the mimeMessage property
+        auto valueCopy = Sink::ApplicationDomain::ApplicationDomainType::getInMemoryRepresentation<DomainType>(*value, properties).template staticCast<DomainType>();
+        if (mResultTransformation) {
+            mResultTransformation(*valueCopy);
+        }
        counter++;
        switch (operation) {
        case Sink::Operation_Creation:
            // Trace() << "Got creation";
-            resultProvider.add(Sink::ApplicationDomain::ApplicationDomainType::getInMemoryRepresentation<DomainType>(*value, properties).template staticCast<DomainType>());
+            resultProvider.add(valueCopy);
            break;
        case Sink::Operation_Modification:
            // Trace() << "Got modification";
-            resultProvider.modify(Sink::ApplicationDomain::ApplicationDomainType::getInMemoryRepresentation<DomainType>(*value, properties).template staticCast<DomainType>());
+            resultProvider.modify(valueCopy);
            break;
        case Sink::Operation_Removal:
            // Trace() << "Got removal";
-            resultProvider.remove(Sink::ApplicationDomain::ApplicationDomainType::getInMemoryRepresentation<DomainType>(*value, properties).template staticCast<DomainType>());
+            resultProvider.remove(valueCopy);
            break;
        }
        return true;
@@ -319,7 +334,7 @@ qint64 QueryWorker<DomainType>::load(const Sink::Query &query, const std::functi
        Warning() << "Error during query: " << error.store << error.message;
    });
    auto transaction = storage.createTransaction(Sink::Storage::ReadOnly);
-    auto db = transaction.openDatabase(mBufferType + ".main");
+    auto db = Storage::mainDatabase(transaction, mBufferType);
    QSet<QByteArray> remainingFilters;
    auto resultSet = baseSetRetriever(transaction, remainingFilters);
diff --git a/common/queryrunner.h b/common/queryrunner.h
index 0ee6a81..04e4587 100644
--- a/common/queryrunner.h
+++ b/common/queryrunner.h
@@ -32,6 +32,9 @@
 class QueryRunnerBase : public QObject
 {
    Q_OBJECT
+public:
+    typedef std::function<void(Sink::ApplicationDomain::ApplicationDomainType &domainObject)> ResultTransformation;
 protected:
    typedef std::function<KAsync::Job<void>()> QueryFunction;
@@ -43,7 +46,6 @@ protected:
        queryFunction = query;
    }
 protected slots:
    /**
     * Rerun query with new revision
@@ -82,10 +84,17 @@ public:
    QueryRunner(const Sink::Query &query, const Sink::ResourceAccessInterface::Ptr &, const QByteArray &instanceIdentifier, const DomainTypeAdaptorFactoryInterface::Ptr &, const QByteArray &bufferType);
    virtual ~QueryRunner();
+    /**
+     * Allows you to run a transformation on every result.
+     * This transformation is executed in the query thread.
+     */
+    void setResultTransformation(const ResultTransformation &transformation);
    typename Sink::ResultEmitter<typename DomainType::Ptr>::Ptr emitter();
 private:
    QSharedPointer<Sink::ResourceAccessInterface> mResourceAccess;
    QSharedPointer<Sink::ResultProvider<typename DomainType::Ptr> > mResultProvider;
+    ResultTransformation mResultTransformation;
 };
diff --git a/common/resource.cpp b/common/resource.cpp
index 8c448a8..5cbb2f2 100644
--- a/common/resource.cpp
+++ b/common/resource.cpp
@@ -34,7 +34,7 @@ Resource::Resource()
    : QObject(),
    d(0)
 {
+    Q_UNUSED(d);
 }
 Resource::~Resource()
@@ -63,6 +63,11 @@ void Resource::setLowerBoundRevision(qint64 revision)
    Q_UNUSED(revision)
 }
+void Resource::removeDataFromDisk()
+{
+}
 class ResourceFactory::Private
 {
 public:
@@ -75,7 +80,7 @@ ResourceFactory::ResourceFactory(QObject *parent)
    : QObject(parent),
      d(0)
 {
+    Q_UNUSED(d);
 }
 ResourceFactory::~ResourceFactory()
@@ -129,3 +134,9 @@ ResourceFactory *ResourceFactory::load(const QString &resourceName)
 }
 } // namespace Sink
+//Ignore warning I don't know how to fix in a moc file
+#pragma clang diagnostic push
+#pragma clang diagnostic ignored "-Wundefined-reinterpret-cast"
+#include "moc_resource.cpp"
+#pragma clang diagnostic pop
diff --git a/common/resource.h b/common/resource.h
index 30d6c46..ab30cb9 100644
--- a/common/resource.h
+++ b/common/resource.h
@@ -19,7 +19,7 @@
 */
 #pragma once
-#include <sinkcommon_export.h>
+#include "sink_export.h"
 #include <Async/Async>
 #include "notification.h"
@@ -31,7 +31,7 @@ class FacadeFactory;
 /**
 * Resource interface
 */
-class SINKCOMMON_EXPORT Resource : public QObject
+class SINK_EXPORT Resource : public QObject
 {
    Q_OBJECT
 public:
@@ -55,6 +55,11 @@ public:
     */
    virtual void setLowerBoundRevision(qint64 revision);
+    /**
+     * Remove the data from disk
+     */
+    virtual void removeDataFromDisk();
 Q_SIGNALS:
    void revisionUpdated(qint64);
    void notify(Notification);
@@ -67,7 +72,7 @@ private:
 /**
 * Factory interface for resource to implement.
 */
-class ResourceFactory : public QObject
+class SINK_EXPORT ResourceFactory : public QObject
 {
 public:
    static ResourceFactory *load(const QString &resourceName);
diff --git a/common/resourceaccess.cpp b/common/resourceaccess.cpp
index 0c435c9..80d60e8 100644
--- a/common/resourceaccess.cpp
+++ b/common/resourceaccess.cpp
@@ -41,11 +41,13 @@
 #include <QProcess>
 #include <QDataStream>
 #include <QBuffer>
+#include <QTime>
 #undef Trace
-#define Trace() Sink::Log::debugStream(Sink::Log::DebugLevel::Trace, __LINE__, __FILE__, Q_FUNC_INFO, "ResourceAccess")
+#define TracePrivate() Trace_area("client.communication." + resourceInstanceIdentifier)
+#define Trace() Trace_area("client.communication." + d->resourceInstanceIdentifier)
 #undef Log
-#define Log(IDENTIFIER) Sink::Log::debugStream(Sink::Log::DebugLevel::Log, __LINE__, __FILE__, Q_FUNC_INFO, "ResourceAccess("+IDENTIFIER+")")
+#define Log() Log_area("client.communication." + d->resourceInstanceIdentifier)
 static void queuedInvoke(const std::function<void()> &f, QObject *context = 0)
 {
@@ -168,45 +170,48 @@ KAsync::Job<void> ResourceAccess::Private::tryToConnect()
        return !socket;
    },
    [this, counter](KAsync::Future<void> &future) {
-        Trace() << "Loop";
+        TracePrivate() << "Loop";
-        KAsync::wait(50)
+        connectToServer(resourceInstanceIdentifier)
-        .then(connectToServer(resourceInstanceIdentifier))
        .then<void, QSharedPointer<QLocalSocket> >([this, &future](const QSharedPointer<QLocalSocket> &s) {
            Q_ASSERT(s);
            socket = s;
            future.setFinished();
-        },
+        }, [&future, counter, this](int errorCode, const QString &errorString) {
-        [&future, counter](int errorCode, const QString &errorString) {
+            static int waitTime = 10;
-            const int maxRetries = 10;
+            static int timeout = 500;
+            static int maxRetries = timeout / waitTime;
            if (*counter > maxRetries) {
-                Trace() << "Giving up";
+                TracePrivate() << "Giving up";
                future.setError(-1, "Failed to connect to socket");
            } else {
-                future.setFinished();
+                KAsync::wait(waitTime).then<void>([&future]() {
+                    future.setFinished();
+                }).exec();
            }
            *counter = *counter + 1;
-        }).exec();
+        })
+        .exec();
    });
 }
 KAsync::Job<void> ResourceAccess::Private::initializeSocket()
 {
    return KAsync::start<void>([this](KAsync::Future<void> &future) {
-        Trace() << "Trying to connect";
+        TracePrivate() << "Trying to connect";
        connectToServer(resourceInstanceIdentifier).then<void, QSharedPointer<QLocalSocket> >([this, &future](const QSharedPointer<QLocalSocket> &s) {
-            Trace() << "Connected to resource, without having to start it.";
+            TracePrivate() << "Connected to resource, without having to start it.";
            Q_ASSERT(s);
            socket = s;
            future.setFinished();
        },
        [this, &future](int errorCode, const QString &errorString) {
-            Trace() << "Failed to connect, starting resource";
+            TracePrivate() << "Failed to connect, starting resource";
            //We failed to connect, so let's start the resource
            QStringList args;
            args << resourceInstanceIdentifier;
            qint64 pid = 0;
            if (QProcess::startDetached("sink_synchronizer", args, QDir::homePath(), &pid)) {
-                Trace() << "Started resource " << pid;
+                TracePrivate() << "Started resource " << pid;
                tryToConnect()
                .then<void>([&future]() {
                    future.setFinished();
@@ -232,12 +237,12 @@ ResourceAccess::ResourceAccess(const QByteArray &resourceInstanceIdentifier)
    : ResourceAccessInterface(),
      d(new Private(getResourceName(resourceInstanceIdentifier), resourceInstanceIdentifier, this))
 {
-    log("Starting access");
+    Log() << "Starting access";
 }
 ResourceAccess::~ResourceAccess()
 {
-    log("Closing access");
+    Log() << "Closing access";
    if (!d->resultHandler.isEmpty()) {
        Warning() << "Left jobs running while shutting down ResourceAccess: " << d->resultHandler.keys();
    }
@@ -380,9 +385,11 @@ void ResourceAccess::open()
    if (d->openingSocket) {
        return;
    }
+    auto time = QSharedPointer<QTime>::create();
+    time->start();
    d->openingSocket = true;
-    d->initializeSocket().then<void>([this]() {
+    d->initializeSocket().then<void>([this, time]() {
-        Trace() << "Socket is initialized";
+        Trace() << "Socket is initialized." << Log::TraceTime(time->elapsed());
        d->openingSocket = false;
        QObject::connect(d->socket.data(), &QLocalSocket::disconnected,
                this, &ResourceAccess::disconnected);
@@ -400,7 +407,7 @@ void ResourceAccess::open()
 void ResourceAccess::close()
 {
-    log(QString("Closing %1").arg(d->socket->fullServerName()));
+    Log() << QString("Closing %1").arg(d->socket->fullServerName());
    Trace() << "Pending commands: " << d->pendingCommands.size();
    Trace() << "Queued commands: " << d->commandQueue.size();
    d->socket->close();
@@ -412,7 +419,7 @@ void ResourceAccess::sendCommand(const QSharedPointer<QueuedCommand> &command)
    //TODO: we should have a timeout for commands
    d->messageId++;
    const auto messageId = d->messageId;
-    log(QString("Sending command \"%1\" with messageId %2").arg(QString(Sink::Commands::name(command->commandId))).arg(d->messageId));
+    Log() << QString("Sending command \"%1\" with messageId %2").arg(QString(Sink::Commands::name(command->commandId))).arg(d->messageId);
    Q_ASSERT(command->callback);
    registerCallback(d->messageId, [this, messageId, command](int errorCode, QString errorMessage) {
        Trace() << "Command complete " << messageId;
@@ -452,7 +459,7 @@ void ResourceAccess::connected()
        return;
    }
-    log(QString("Connected: %1").arg(d->socket->fullServerName()));
+    Log() << QString("Connected: %1").arg(d->socket->fullServerName());
    {
        flatbuffers::FlatBufferBuilder fbb;
@@ -472,7 +479,7 @@ void ResourceAccess::connected()
 void ResourceAccess::disconnected()
 {
-    log(QString("Disconnected from %1").arg(d->socket->fullServerName()));
+    Log() << QString("Disconnected from %1").arg(d->socket->fullServerName());
    d->socket->close();
    emit ready(false);
 }
@@ -480,7 +487,7 @@ void ResourceAccess::disconnected()
 void ResourceAccess::connectionError(QLocalSocket::LocalSocketError error)
 {
    if (error == QLocalSocket::PeerClosedError) {
-        Log(d->resourceInstanceIdentifier) << "The resource closed the connection.";
+        Log() << "The resource closed the connection.";
        d->abortPendingOperations();
    } else {
        Warning() << QString("Connection error: %1 : %2").arg(error).arg(d->socket->errorString());
@@ -524,14 +531,17 @@ bool ResourceAccess::processMessageBuffer()
    switch (commandId) {
        case Commands::RevisionUpdateCommand: {
            auto buffer = Commands::GetRevisionUpdate(d->partialMessageBuffer.constData() + headerSize);
-            log(QString("Revision updated to: %1").arg(buffer->revision()));
+            Log() << QString("Revision updated to: %1").arg(buffer->revision());
+            Notification n;
+            n.type = Sink::Commands::NotificationType::NotificationType_RevisionUpdate;
+            emit notification(n);
            emit revisionChanged(buffer->revision());
            break;
        }
        case Commands::CommandCompletionCommand: {
            auto buffer = Commands::GetCommandCompletion(d->partialMessageBuffer.constData() + headerSize);
-            log(QString("Command with messageId %1 completed %2").arg(buffer->id()).arg(buffer->success() ? "sucessfully" : "unsuccessfully"));
+            Log() << QString("Command with messageId %1 completed %2").arg(buffer->id()).arg(buffer->success() ? "sucessfully" : "unsuccessfully");
            d->completeCommands << buffer->id();
            //The callbacks can result in this object getting destroyed directly, so we need to ensure we finish our work first
@@ -544,11 +554,11 @@ bool ResourceAccess::processMessageBuffer()
            auto buffer = Commands::GetNotification(d->partialMessageBuffer.constData() + headerSize);
            switch (buffer->type()) {
                case Sink::Commands::NotificationType::NotificationType_Shutdown:
-                    Log(d->resourceInstanceIdentifier) << "Received shutdown notification.";
+                    Log() << "Received shutdown notification.";
                    close();
                    break;
                case Sink::Commands::NotificationType::NotificationType_Inspection: {
-                    Log(d->resourceInstanceIdentifier) << "Received inspection notification.";
+                    Log() << "Received inspection notification.";
                    Notification n;
                    if (buffer->identifier()) {
                        //Don't use fromRawData, the buffer is gone once we invoke emit notification
@@ -566,6 +576,10 @@ bool ResourceAccess::processMessageBuffer()
                    }, this);
                }
                    break;
+                case Sink::Commands::NotificationType::NotificationType_Status:
+                case Sink::Commands::NotificationType::NotificationType_Warning:
+                case Sink::Commands::NotificationType::NotificationType_Progress:
+                case Sink::Commands::NotificationType::NotificationType_RevisionUpdate:
                default:
                    Warning() << "Received unknown notification: " << buffer->type();
                    break;
@@ -580,9 +594,9 @@ bool ResourceAccess::processMessageBuffer()
    return d->partialMessageBuffer.size() >= headerSize;
 }
-void ResourceAccess::log(const QString &message)
-{
-    Log(d->resourceInstanceIdentifier) << this << message;
 }
-}
+#pragma clang diagnostic push
+#pragma clang diagnostic ignored "-Wundefined-reinterpret-cast"
+#include "moc_resourceaccess.cpp"
+#pragma clang diagnostic pop
diff --git a/common/resourceaccess.h b/common/resourceaccess.h
index 73b676b..4c10adb 100644
--- a/common/resourceaccess.h
+++ b/common/resourceaccess.h
@@ -20,6 +20,7 @@
 #pragma once
+#include "sink_export.h"
 #include <QLocalSocket>
 #include <QObject>
 #include <QTimer>
@@ -34,7 +35,7 @@ namespace Sink
 struct QueuedCommand;
-class ResourceAccessInterface : public QObject
+class SINK_EXPORT ResourceAccessInterface : public QObject
 {
    Q_OBJECT
 public:
@@ -62,7 +63,7 @@ public Q_SLOTS:
    virtual void close() = 0;
 };
-class ResourceAccess : public ResourceAccessInterface
+class SINK_EXPORT ResourceAccess : public ResourceAccessInterface
 {
    Q_OBJECT
 public:
@@ -100,7 +101,6 @@ private Q_SLOTS:
 private:
    void connected();
-    void log(const QString &message);
    void registerCallback(uint messageId, const std::function<void(int error, const QString &)> &callback);
    void sendCommand(const QSharedPointer<QueuedCommand> &command);
diff --git a/common/resourceconfig.h b/common/resourceconfig.h
index cc9cb94..2108caa 100644
--- a/common/resourceconfig.h
+++ b/common/resourceconfig.h
@@ -19,12 +19,13 @@
 #pragma once
+#include "sink_export.h"
 #include <QList>
 #include <QByteArray>
 #include <QVariant>
 #include <QMap>
-class ResourceConfig
+class SINK_EXPORT ResourceConfig
 {
 public:
    static QMap<QByteArray, QByteArray> getResources();
diff --git a/common/resourcecontrol.cpp b/common/resourcecontrol.cpp
new file mode 100644
index 0000000..83558bd
--- /dev/null
+++ b/common/resourcecontrol.cpp
@@ -0,0 +1,125 @@
+/*
+ * Copyright (C) 2015 Christian Mollekopf <chrigi_1@fastmail.fm>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) version 3, or any
+ * later version accepted by the membership of KDE e.V. (or its
+ * successor approved by the membership of KDE e.V.), which shall
+ * act as a proxy defined in Section 6 of version 3 of the license.
+ *
+ * This library 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
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library.  If not, see <http://www.gnu.org/licenses/>.
+ */
+#include "resourcecontrol.h"
+#include <QTime>
+#include <QUuid>
+#include <functional>
+#include "resourceaccess.h"
+#include "commands.h"
+#include "log.h"
+#include "notifier.h"
+#undef DEBUG_AREA
+#define DEBUG_AREA "client.resourcecontrol"
+namespace Sink
+{
+KAsync::Job<void> ResourceControl::shutdown(const QByteArray &identifier)
+{
+    Trace() << "shutdown " << identifier;
+    auto time = QSharedPointer<QTime>::create();
+    time->start();
+    return ResourceAccess::connectToServer(identifier).then<void, QSharedPointer<QLocalSocket>>([identifier, time](QSharedPointer<QLocalSocket> socket, KAsync::Future<void> &future) {
+        //We can't currently reuse the socket
+        socket->close();
+        auto resourceAccess = QSharedPointer<Sink::ResourceAccess>::create(identifier);
+        resourceAccess->open();
+        resourceAccess->sendCommand(Sink::Commands::ShutdownCommand).then<void>([&future, resourceAccess, time]() {
+            Trace() << "Shutdown complete." << Log::TraceTime(time->elapsed());
+            future.setFinished();
+        }).exec();
+    },
+    [](int, const QString &) {
+        Trace() << "Resource is already closed.";
+        //Resource isn't started, nothing to shutdown
+    });
+}
+KAsync::Job<void> ResourceControl::start(const QByteArray &identifier)
+{
+    Trace() << "start " << identifier;
+    auto time = QSharedPointer<QTime>::create();
+    time->start();
+    auto resourceAccess = QSharedPointer<Sink::ResourceAccess>::create(identifier);
+    resourceAccess->open();
+    return resourceAccess->sendCommand(Sink::Commands::PingCommand).then<void>([resourceAccess, time]() {
+        Trace() << "Start complete." << Log::TraceTime(time->elapsed());
+    });
+}
+KAsync::Job<void> ResourceControl::flushMessageQueue(const QByteArrayList &resourceIdentifier)
+{
+    Trace() << "flushMessageQueue" << resourceIdentifier;
+    return KAsync::iterate(resourceIdentifier)
+    .template each<void, QByteArray>([](const QByteArray &resource, KAsync::Future<void> &future) {
+        Trace() << "Flushing message queue " << resource;
+        auto resourceAccess = QSharedPointer<Sink::ResourceAccess>::create(resource);
+        resourceAccess->open();
+        resourceAccess->synchronizeResource(false, true).then<void>([&future, resourceAccess]() {
+            future.setFinished();
+        }).exec();
+    });
+}
+KAsync::Job<void> ResourceControl::flushReplayQueue(const QByteArrayList &resourceIdentifier)
+{
+    return flushMessageQueue(resourceIdentifier);
+}
+template <class DomainType>
+KAsync::Job<void> ResourceControl::inspect(const Inspection &inspectionCommand)
+{
+    auto resource = inspectionCommand.resourceIdentifier;
+    auto time = QSharedPointer<QTime>::create();
+    time->start();
+    Trace() << "Sending inspection " << resource;
+    auto resourceAccess = QSharedPointer<Sink::ResourceAccess>::create(resource);
+    resourceAccess->open();
+    auto notifier = QSharedPointer<Sink::Notifier>::create(resourceAccess);
+    auto id = QUuid::createUuid().toByteArray();
+    return resourceAccess->sendInspectionCommand(id, ApplicationDomain::getTypeName<DomainType>(), inspectionCommand.entityIdentifier, inspectionCommand.property, inspectionCommand.expectedValue)
+        .template then<void>([resourceAccess, notifier, id, time](KAsync::Future<void> &future) {
+            notifier->registerHandler([&future, id, time](const Notification &notification) {
+                if (notification.id == id) {
+                    Trace() << "Inspection complete." << Log::TraceTime(time->elapsed());
+                    if (notification.code) {
+                        future.setError(-1, "Inspection returned an error: " + notification.message);
+                    } else {
+                        future.setFinished();
+                    }
+                }
+            });
+        });
+}
+#define REGISTER_TYPE(T) template KAsync::Job<void> ResourceControl::inspect<T>(const Inspection &); \
+REGISTER_TYPE(ApplicationDomain::Event);
+REGISTER_TYPE(ApplicationDomain::Mail);
+REGISTER_TYPE(ApplicationDomain::Folder);
+REGISTER_TYPE(ApplicationDomain::SinkResource);
+} // namespace Sink
diff --git a/common/resourcecontrol.h b/common/resourcecontrol.h
new file mode 100644
index 0000000..5bfa67f
--- /dev/null
+++ b/common/resourcecontrol.h
@@ -0,0 +1,62 @@
+/*
+ * Copyright (C) 2015 Christian Mollekopf <chrigi_1@fastmail.fm>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) version 3, or any
+ * later version accepted by the membership of KDE e.V. (or its
+ * successor approved by the membership of KDE e.V.), which shall
+ * act as a proxy defined in Section 6 of version 3 of the license.
+ *
+ * This library 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
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library.  If not, see <http://www.gnu.org/licenses/>.
+ */
+#pragma once
+#include "sink_export.h"
+#include <QByteArray>
+#include <Async/Async>
+#include "inspection.h"
+namespace Sink {
+namespace ResourceControl {
+template <class DomainType>
+KAsync::Job<void> SINK_EXPORT inspect(const Inspection &inspectionCommand);
+/**
+ * Shutdown resource.
+ */
+KAsync::Job<void> SINK_EXPORT shutdown(const QByteArray &resourceIdentifier);
+/**
+ * Start resource.
+ * 
+ * The resource is ready for operation once this command completes.
+ * This command is only necessary if a resource was shutdown previously,
+ * otherwise the resource process will automatically start as necessary.
+ */
+KAsync::Job<void> SINK_EXPORT start(const QByteArray &resourceIdentifier);
+/**
+ * Flushes any pending messages to disk
+ */
+KAsync::Job<void> SINK_EXPORT flushMessageQueue(const QByteArrayList &resourceIdentifier);
+/**
+ * Flushes any pending messages that haven't been replayed to the source.
+ */
+KAsync::Job<void> SINK_EXPORT flushReplayQueue(const QByteArrayList &resourceIdentifier);
+    }
+}
diff --git a/common/resultprovider.h b/common/resultprovider.h
index 6958dbf..2d6efbe 100644
--- a/common/resultprovider.h
+++ b/common/resultprovider.h
@@ -27,8 +27,6 @@
 #include "resultset.h"
 #include "log.h"
-using namespace async;
 namespace Sink {
 /**
@@ -47,6 +45,11 @@ public:
    }
+    virtual ~ResultProviderInterface()
+    {
+    }
    virtual void add(const T &value) = 0;
    virtual void modify(const T &value) = 0;
    virtual void remove(const T &value) = 0;
@@ -183,7 +186,7 @@ public:
    void onDone(const std::function<void()> &callback)
    {
-        mThreadBoundary = QSharedPointer<ThreadBoundary>::create();
+        mThreadBoundary = QSharedPointer<async::ThreadBoundary>::create();
        mOnDoneCallback = callback;
    }
@@ -212,7 +215,7 @@ private:
    QWeakPointer<ResultEmitter<T> > mResultEmitter;
    std::function<void()> mOnDoneCallback;
-    QSharedPointer<ThreadBoundary> mThreadBoundary;
+    QSharedPointer<async::ThreadBoundary> mThreadBoundary;
    std::function<void(const T &parent)> mFetcher;
 };
@@ -327,7 +330,7 @@ private:
    std::function<void(void)> clearHandler;
    std::function<void(const DomainType &parent)> mFetcher;
-    ThreadBoundary mThreadBoundary;
+    async::ThreadBoundary mThreadBoundary;
 };
 template<class DomainType>
diff --git a/common/storage.h b/common/storage.h
index 2d34f1f..ac03947 100644
--- a/common/storage.h
+++ b/common/storage.h
@@ -21,7 +21,7 @@
 #pragma once
-#include <sinkcommon_export.h>
+#include "sink_export.h"
 #include <string>
 #include <functional>
 #include <QString>
@@ -29,7 +29,7 @@
 namespace Sink
 {
-class SINKCOMMON_EXPORT Storage {
+class SINK_EXPORT Storage {
 public:
    enum AccessMode { ReadOnly, ReadWrite };
@@ -180,6 +180,13 @@ public:
    qint64 diskUsage() const;
    void removeFromDisk() const;
+    /**
+     * Clears all cached environments.
+     * 
+     * This only ever has to be called if a database was removed from another process.
+     */
+    static void clearEnv();
    static qint64 maxRevision(const Sink::Storage::Transaction &);
    static void setMaxRevision(Sink::Storage::Transaction &, qint64 revision);
@@ -200,6 +207,8 @@ public:
    static QByteArray assembleKey(const QByteArray &key, qint64 revision);
    static QByteArray uidFromKey(const QByteArray &key);
+    static NamedDatabase mainDatabase(const Sink::Storage::Transaction &, const QByteArray &type);
 private:
    std::function<void(const Storage::Error &error)> mErrorHandler;
diff --git a/common/storage_common.cpp b/common/storage_common.cpp
index ea97ac2..0b842d1 100644
--- a/common/storage_common.cpp
+++ b/common/storage_common.cpp
@@ -156,6 +156,11 @@ QByteArray Storage::uidFromKey(const QByteArray &key)
    return key.mid(0, 38);
 }
+Storage::NamedDatabase Storage::mainDatabase(const Sink::Storage::Transaction &t, const QByteArray &type)
+{
+    return t.openDatabase(type + ".main");
+}
 bool Storage::NamedDatabase::contains(const QByteArray &uid)
 {
    bool found = false;
diff --git a/common/storage_lmdb.cpp b/common/storage_lmdb.cpp
index 6539eb0..1efffc4 100644
--- a/common/storage_lmdb.cpp
+++ b/common/storage_lmdb.cpp
@@ -156,8 +156,8 @@ void Storage::NamedDatabase::remove(const QByteArray &k, const QByteArray &value
                     const std::function<void(const Storage::Error &error)> &errorHandler)
 {
    if (!d || !d->transaction) {
-        Error error(d->name.toLatin1() + d->db, ErrorCodes::GenericError, "Not open");
        if (d) {
+            Error error(d->name.toLatin1() + d->db, ErrorCodes::GenericError, "Not open");
            errorHandler ? errorHandler(error) : d->defaultErrorHandler(error);
        }
        return;
@@ -443,8 +443,8 @@ Storage::NamedDatabase Storage::Transaction::openDatabase(const QByteArray &db,
    d->implicitCommit = true;
    auto p = new Storage::NamedDatabase::Private(db, allowDuplicates, d->defaultErrorHandler, d->name, d->transaction);
    if (!p->openDatabase(d->requestedRead, errorHandler)) {
-        return Storage::NamedDatabase();
        delete p;
+        return Storage::NamedDatabase();
    }
    return Storage::NamedDatabase(p);
 }
@@ -463,7 +463,7 @@ QList<QByteArray> Storage::Transaction::getDatabaseNames() const
        MDB_val data;
        MDB_cursor *cursor;
-        rc = mdb_cursor_open(d->transaction, d->dbi, &cursor);
+        mdb_cursor_open(d->transaction, d->dbi, &cursor);
        if ((rc = mdb_cursor_get(cursor, &key, &data, MDB_FIRST)) == 0) {
            list << QByteArray::fromRawData((char*)key.mv_data, key.mv_size);
            while ((rc = mdb_cursor_get(cursor, &key, &data, MDB_NEXT)) == 0) {
@@ -615,4 +615,12 @@ void Storage::removeFromDisk() const
    mdb_env_close(env);
 }
+void Storage::clearEnv()
+{
+    for (auto env : Storage::Private::sEnvironments) {
+        mdb_env_close(env);
+    }
+    Storage::Private::sEnvironments.clear();
+}
 } // namespace Sink
diff --git a/common/clientapi.cpp b/common/store.cpp
index be9f3fd..6f080b5 100644
--- a/common/clientapi.cpp
+++ b/common/store.cpp
@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2014 Christian Mollekopf <chrigi_1@fastmail.fm>
+ * Copyright (C) 2015 Christian Mollekopf <chrigi_1@fastmail.fm>
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
@@ -18,14 +18,10 @@
 * License along with this library.  If not, see <http://www.gnu.org/licenses/>.
 */
-#include "clientapi.h"
+#include "store.h"
-#include <QtConcurrent/QtConcurrentRun>
+#include <QTime>
-#include <QTimer>
-#include <QEventLoop>
 #include <QAbstractItemModel>
-#include <QDir>
-#include <QUuid>
 #include <functional>
 #include <memory>
@@ -39,6 +35,9 @@
 #include "storage.h"
 #include "log.h"
+#undef DEBUG_AREA
+#define DEBUG_AREA "client.store"
 namespace Sink
 {
@@ -47,11 +46,6 @@ QString Store::storageLocation()
    return Sink::storageLocation();
 }
-QByteArray Store::resourceName(const QByteArray &instanceIdentifier)
-{
-    return Sink::resourceName(instanceIdentifier);
-}
 static QList<QByteArray> getResources(const QList<QByteArray> &resourceFilter, const QByteArray &type)
 {
    //Return the global resource (signified by an empty name) for types that don't eblong to a specific resource
@@ -156,47 +150,21 @@ KAsync::Job<void> Store::remove(const DomainType &domainObject)
    });
 }
-KAsync::Job<void> Store::shutdown(const QByteArray &identifier)
+KAsync::Job<void> Store::removeDataFromDisk(const QByteArray &identifier)
-{
-    Trace() << "shutdown " << identifier;
-    return ResourceAccess::connectToServer(identifier).then<void, QSharedPointer<QLocalSocket>>([identifier](QSharedPointer<QLocalSocket> socket, KAsync::Future<void> &future) {
-        //We can't currently reuse the socket
-        socket->close();
-        auto resourceAccess = QSharedPointer<Sink::ResourceAccess>::create(identifier);
-        resourceAccess->open();
-        resourceAccess->sendCommand(Sink::Commands::ShutdownCommand).then<void>([&future, resourceAccess]() {
-            Trace() << "Shutdown complete";
-            future.setFinished();
-        }).exec();
-    },
-    [](int, const QString &) {
-        Trace() << "Resource is already closed.";
-        //Resource isn't started, nothing to shutdown
-    })
-    //FIXME JOBAPI this is only required because we don't care about the return value of connectToServer
-    .template then<void>([](){});
-}
-KAsync::Job<void> Store::start(const QByteArray &identifier)
 {
-    Trace() << "start " << identifier;
+    //All databases are going to become invalid, nuke the environments
+    //TODO: all clients should react to a notification the resource
+    Sink::Storage::clearEnv();
+    Trace() << "Remove data from disk " << identifier;
+    auto time = QSharedPointer<QTime>::create();
+    time->start();
    auto resourceAccess = QSharedPointer<Sink::ResourceAccess>::create(identifier);
    resourceAccess->open();
-    return resourceAccess->sendCommand(Sink::Commands::PingCommand).then<void>([resourceAccess]() {
+    return resourceAccess->sendCommand(Sink::Commands::RemoveFromDiskCommand).then<void>([resourceAccess, time]() {
-        Trace() << "Start complete";
+        Trace() << "Remove from disk complete." << Log::TraceTime(time->elapsed());
    });
 }
-void Store::removeFromDisk(const QByteArray &identifier)
-{
-    //TODO By calling the resource executable with a --remove option instead
-    //we can ensure that no other resource process is running at the same time
-    QDir dir(Sink::storageLocation());
-    for (const auto &folder : dir.entryList(QStringList() << identifier + "*")) {
-        Sink::Storage(Sink::storageLocation(), folder, Sink::Storage::ReadWrite).removeFromDisk();
-    }
-}
 KAsync::Job<void> Store::synchronize(const Sink::Query &query)
 {
    Trace() << "synchronize" << query.resources;
@@ -208,30 +176,7 @@ KAsync::Job<void> Store::synchronize(const Sink::Query &query)
        resourceAccess->synchronizeResource(true, false).then<void>([&future, resourceAccess]() {
            future.setFinished();
        }).exec();
-    })
+    });
-    //FIXME JOBAPI this is only required because we don't care about the return value of each (and each shouldn't even have a return value)
-    .template then<void>([](){});
-}
-KAsync::Job<void> Store::flushMessageQueue(const QByteArrayList &resourceIdentifier)
-{
-    Trace() << "flushMessageQueue" << resourceIdentifier;
-    return KAsync::iterate(resourceIdentifier)
-    .template each<void, QByteArray>([](const QByteArray &resource, KAsync::Future<void> &future) {
-        Trace() << "Flushing message queue " << resource;
-        auto resourceAccess = QSharedPointer<Sink::ResourceAccess>::create(resource);
-        resourceAccess->open();
-        resourceAccess->synchronizeResource(false, true).then<void>([&future, resourceAccess]() {
-            future.setFinished();
-        }).exec();
-    })
-    //FIXME JOBAPI this is only required because we don't care about the return value of each (and each shouldn't even have a return value)
-    .template then<void>([](){});
-}
-KAsync::Job<void> Store::flushReplayQueue(const QByteArrayList &resourceIdentifier)
-{
-    return flushMessageQueue(resourceIdentifier);
 }
 template <class DomainType>
@@ -295,63 +240,10 @@ KAsync::Job<QList<typename DomainType::Ptr> > Store::fetch(const Sink::Query &qu
    });
 }
-template <class DomainType>
-KAsync::Job<void> Resources::inspect(const Inspection &inspectionCommand)
-{
-    auto resource = inspectionCommand.resourceIdentifier;
-    Trace() << "Sending inspection " << resource;
-    auto resourceAccess = QSharedPointer<Sink::ResourceAccess>::create(resource);
-    resourceAccess->open();
-    auto notifier = QSharedPointer<Sink::Notifier>::create(resourceAccess);
-    auto id = QUuid::createUuid().toByteArray();
-    return resourceAccess->sendInspectionCommand(id, ApplicationDomain::getTypeName<DomainType>(), inspectionCommand.entityIdentifier, inspectionCommand.property, inspectionCommand.expectedValue)
-        .template then<void>([resourceAccess, notifier, id](KAsync::Future<void> &future) {
-            notifier->registerHandler([&future, id](const Notification &notification) {
-                if (notification.id == id) {
-                    if (notification.code) {
-                        future.setError(-1, "Inspection returned an error: " + notification.message);
-                    } else {
-                        future.setFinished();
-                    }
-                }
-            });
-        });
-}
-class Sink::Notifier::Private {
-public:
-    Private()
-        : context(new QObject)
-    {
-    }
-    QList<QSharedPointer<ResourceAccess> > resourceAccess;
-    QList<std::function<void(const Notification &)> > handler;
-    QSharedPointer<QObject> context;
-};
-Notifier::Notifier(const QSharedPointer<ResourceAccess> &resourceAccess)
-    : d(new Sink::Notifier::Private)
-{
-    QObject::connect(resourceAccess.data(), &ResourceAccess::notification, d->context.data(), [this](const Notification &notification) {
-        for (const auto &handler : d->handler) {
-            handler(notification);
-        }
-    });
-    d->resourceAccess << resourceAccess;
-}
-void Notifier::registerHandler(std::function<void(const Notification &)> handler)
-{
-    d->handler << handler;
-}
 #define REGISTER_TYPE(T) template KAsync::Job<void> Store::remove<T>(const T &domainObject); \
    template KAsync::Job<void> Store::create<T>(const T &domainObject); \
    template KAsync::Job<void> Store::modify<T>(const T &domainObject); \
    template QSharedPointer<QAbstractItemModel> Store::loadModel<T>(Query query); \
-    template KAsync::Job<void> Resources::inspect<T>(const Inspection &); \
    template KAsync::Job<T> Store::fetchOne<T>(const Query &); \
    template KAsync::Job<QList<T::Ptr> > Store::fetchAll<T>(const Query &); \
    template KAsync::Job<QList<T::Ptr> > Store::fetch<T>(const Query &, int); \
diff --git a/common/store.h b/common/store.h
new file mode 100644
index 0000000..6696833
--- /dev/null
+++ b/common/store.h
@@ -0,0 +1,101 @@
+/*
+ * Copyright (C) 2015 Christian Mollekopf <chrigi_1@fastmail.fm>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) version 3, or any
+ * later version accepted by the membership of KDE e.V. (or its
+ * successor approved by the membership of KDE e.V.), which shall
+ * act as a proxy defined in Section 6 of version 3 of the license.
+ *
+ * This library 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
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library.  If not, see <http://www.gnu.org/licenses/>.
+ */
+#pragma once
+#include "sink_export.h"
+#include <QString>
+#include <QSharedPointer>
+#include <Async/Async>
+#include "query.h"
+#include "applicationdomaintype.h"
+class QAbstractItemModel;
+namespace Sink {
+/**
+ * The unified Sink Store.
+ * 
+ * This is the primary interface for clients to interact with Sink.
+ * It provides a unified store where all data provided by various resources can be accessed and modified.
+ */
+namespace Store {
+QString SINK_EXPORT storageLocation();
+enum Roles {
+    DomainObjectRole = Qt::UserRole + 1, //Must be the same as in ModelResult
+    ChildrenFetchedRole,
+    DomainObjectBaseRole
+};
+/**
+    * Asynchronusly load a dataset with tree structure information
+    */
+template <class DomainType>
+QSharedPointer<QAbstractItemModel> SINK_EXPORT loadModel(Query query);
+/**
+ * Create a new entity.
+ */
+template <class DomainType>
+KAsync::Job<void> SINK_EXPORT create(const DomainType &domainObject);
+/**
+ * Modify an entity.
+ * 
+ * This includes moving etc. since these are also simple settings on a property.
+ */
+template <class DomainType>
+KAsync::Job<void> SINK_EXPORT modify(const DomainType &domainObject);
+/**
+ * Remove an entity.
+ */
+template <class DomainType>
+KAsync::Job<void> SINK_EXPORT remove(const DomainType &domainObject);
+/**
+ * Synchronize data to local cache.
+ */
+KAsync::Job<void> SINK_EXPORT synchronize(const Sink::Query &query);
+/**
+ * Removes all resource data from disk.
+ * 
+ * This will not touch the configuration. All commands that that arrived at the resource before this command will be dropped. All commands that arrived later will be executed.
+ */
+KAsync::Job<void> SINK_EXPORT removeDataFromDisk(const QByteArray &resourceIdentifier);
+template <class DomainType>
+KAsync::Job<DomainType> SINK_EXPORT fetchOne(const Sink::Query &query);
+template <class DomainType>
+KAsync::Job<QList<typename DomainType::Ptr> > SINK_EXPORT fetchAll(const Sink::Query &query);
+template <class DomainType>
+KAsync::Job<QList<typename DomainType::Ptr> > SINK_EXPORT fetch(const Sink::Query &query, int minimumAmount = 0);
+    }
+}
diff --git a/common/threadboundary.h b/common/threadboundary.h
index 0d8ed3b..7bea4ea 100644
--- a/common/threadboundary.h
+++ b/common/threadboundary.h
@@ -20,6 +20,8 @@
 #pragma once
+#include "sink_export.h"
 #include <QObject>
 #include <functional>
@@ -29,7 +31,7 @@ namespace async {
 * A helper class to invoke a method in a different thread using the event loop.
 * The ThreadBoundary object must live in the thread where the function should be called.
 */
-class ThreadBoundary : public QObject {
+class SINK_EXPORT ThreadBoundary : public QObject {
    Q_OBJECT
 public:
    ThreadBoundary();
diff --git a/common/typeindex.cpp b/common/typeindex.cpp
index 03ad8f7..b1bcf6a 100644
--- a/common/typeindex.cpp
+++ b/common/typeindex.cpp
@@ -22,6 +22,9 @@
 #include "index.h"
 #include <QDateTime>
+#undef DEBUG_AREA
+#define DEBUG_AREA "common.typeindex"
 TypeIndex::TypeIndex(const QByteArray &type)
    : mType(type)
 {
@@ -111,9 +114,9 @@ ResultSet TypeIndex::query(const Sink::Query &query, QSet<QByteArray> &appliedFi
                Warning() << "Error in index: " <<  error.message << property;
            });
            appliedFilters << property;
+            Trace() << "Index lookup on " << property << " found " << keys.size() << " keys.";
+            return ResultSet(keys);
        }
-        Trace() << "Index lookup on " << property << " found " << keys.size() << " keys.";
-        return ResultSet(keys);
    }
    Trace() << "No matching index";
    return ResultSet(keys);