23 files changed, 102 insertions, 31 deletions
diff --git a/common/commands/synchronize.fbs b/common/commands/synchronize.fbs
index 5528166..62f4b2b 100644
--- a/common/commands/synchronize.fbs
+++ b/common/commands/synchronize.fbs
@@ -3,6 +3,7 @@ namespace Sink.Commands;
 table Synchronize {
    sourceSync: bool; //Synchronize with source
    localSync: bool; //Ensure all queues are processed so the local state is up-to date.
+    query: string;
 }
 root_type Synchronize;
diff --git a/common/genericresource.cpp b/common/genericresource.cpp
index 39bd39e..1fc7744 100644
--- a/common/genericresource.cpp
+++ b/common/genericresource.cpp
@@ -399,9 +399,9 @@ void GenericResource::processCommand(int commandId, const QByteArray &data)
    }
 }
-KAsync::Job<void> GenericResource::synchronizeWithSource()
+KAsync::Job<void> GenericResource::synchronizeWithSource(const Sink::QueryBase &query)
 {
-    return KAsync::start<void>([this]() {
+    return KAsync::start<void>([this, query] {
        Sink::Notification n;
        n.id = "sync";
@@ -413,7 +413,7 @@ KAsync::Job<void> GenericResource::synchronizeWithSource()
        SinkLog() << " Synchronizing";
        // Changereplay would deadlock otherwise when trying to open the synchronization store
        enableChangeReplay(false);
-        return mSynchronizer->synchronize()
+        return mSynchronizer->synchronize(query)
            .then<void>([this](const KAsync::Error &error) {
                enableChangeReplay(true);
                if (!error) {
diff --git a/common/genericresource.h b/common/genericresource.h
index 687e307..3736c8f 100644
--- a/common/genericresource.h
+++ b/common/genericresource.h
@@ -47,7 +47,7 @@ public:
    virtual ~GenericResource();
    virtual void processCommand(int commandId, const QByteArray &data) Q_DECL_OVERRIDE;
-    virtual KAsync::Job<void> synchronizeWithSource() Q_DECL_OVERRIDE;
+    virtual KAsync::Job<void> synchronizeWithSource(const Sink::QueryBase &query) Q_DECL_OVERRIDE;
    virtual KAsync::Job<void> processAllMessages() Q_DECL_OVERRIDE;
    virtual void setLowerBoundRevision(qint64 revision) Q_DECL_OVERRIDE;
    virtual KAsync::Job<void>
diff --git a/common/listener.cpp b/common/listener.cpp
index 0742017..c3c6bc2 100644
--- a/common/listener.cpp
+++ b/common/listener.cpp
@@ -245,7 +245,13 @@ void Listener::processCommand(int commandId, uint messageId, const QByteArray &c
                timer->start();
                auto job = KAsync::null<void>();
                if (buffer->sourceSync()) {
-                    job = loadResource().synchronizeWithSource();
+                    Sink::QueryBase query;
+                    if (buffer->query()) {
+                        auto data = QByteArray::fromStdString(buffer->query()->str());
+                        QDataStream stream(&data, QIODevice::ReadOnly);
+                        stream >> query;
+                    }
+                    job = loadResource().synchronizeWithSource(query);
                }
                if (buffer->localSync()) {
                    job = job.then<void>(loadResource().processAllMessages());
diff --git a/common/query.cpp b/common/query.cpp
index 3b717aa..caca775 100644
--- a/common/query.cpp
+++ b/common/query.cpp
@@ -26,7 +26,7 @@ using namespace Sink;
 static const int registerQuery = qRegisterMetaTypeStreamOperators<Sink::QueryBase>();
-QDebug operator<<(QDebug dbg, const Sink::Query::Comparator &c)
+QDebug operator<<(QDebug dbg, const Sink::QueryBase::Comparator &c)
 {
    if (c.comparator == Sink::Query::Comparator::Equals) {
        dbg.nospace() << "== " << c.value;
@@ -39,6 +39,24 @@ QDebug operator<<(QDebug dbg, const Sink::Query::Comparator &c)
    return dbg.space();
 }
+QDebug operator<<(QDebug dbg, const Sink::QueryBase &query)
+{
+    dbg.nospace() << "Query [" << query.type() << "]\n";
+    dbg.nospace() << "  Filter: " << query.getBaseFilters() << "\n";
+    dbg.nospace() << "  Ids: " << query.ids() << "\n";
+    dbg.nospace() << "  Sorting: " << query.sortProperty() << "\n";
+    return dbg.maybeSpace();
+}
+QDebug operator<<(QDebug dbg, const Sink::Query &query)
+{
+    dbg << static_cast<Sink::QueryBase>(query);
+    dbg.nospace() << "  Requested: " << query.requestedProperties << "\n";
+    dbg.nospace() << "  Parent: " << query.parentProperty << "\n";
+    dbg.nospace() << "  IsLive: " << query.liveQuery() << "\n";
+    return dbg.maybeSpace();
+}
 QDataStream & operator<< (QDataStream &stream, const Sink::QueryBase::Comparator &comparator)
 {
    stream << comparator.comparator;
diff --git a/common/query.h b/common/query.h
index 4a73357..925b014 100644
--- a/common/query.h
+++ b/common/query.h
@@ -467,7 +467,9 @@ private:
 }
-QDebug SINK_EXPORT operator<<(QDebug dbg, const Sink::Query::Comparator &c);
+QDebug SINK_EXPORT operator<<(QDebug dbg, const Sink::QueryBase::Comparator &);
+QDebug SINK_EXPORT operator<<(QDebug dbg, const Sink::QueryBase &);
+QDebug SINK_EXPORT operator<<(QDebug dbg, const Sink::Query &);
 QDataStream & SINK_EXPORT operator<< (QDataStream &stream, const Sink::QueryBase &query);
 QDataStream & SINK_EXPORT operator>> (QDataStream &stream, Sink::QueryBase &query);
diff --git a/common/queryrunner.cpp b/common/queryrunner.cpp
index cf6fcf8..99f9c2d 100644
--- a/common/queryrunner.cpp
+++ b/common/queryrunner.cpp
@@ -62,7 +62,7 @@ template <class DomainType>
 QueryRunner<DomainType>::QueryRunner(const Sink::Query &query, const Sink::ResourceContext &context, const QByteArray &bufferType)
    : QueryRunnerBase(), mResourceContext(context), mResourceAccess(mResourceContext.resourceAccess()), mResultProvider(new ResultProvider<typename DomainType::Ptr>), mBatchSize(query.limit)
 {
-    SinkTrace() << "Starting query";
+    SinkTrace() << "Starting query. Is live:" << query.liveQuery() << " Limit: " << query.limit;
    if (query.limit && query.sortProperty().isEmpty()) {
        SinkWarning() << "A limited query without sorting is typically a bad idea.";
    }
diff --git a/common/resource.cpp b/common/resource.cpp
index db64d33..f81f094 100644
--- a/common/resource.cpp
+++ b/common/resource.cpp
@@ -46,7 +46,7 @@ void Resource::processCommand(int commandId, const QByteArray &data)
    Q_UNUSED(data)
 }
-KAsync::Job<void> Resource::synchronizeWithSource()
+KAsync::Job<void> Resource::synchronizeWithSource(const Sink::QueryBase &query)
 {
    return KAsync::null<void>();
 }
diff --git a/common/resource.h b/common/resource.h
index 1dbc365..3cc326c 100644
--- a/common/resource.h
+++ b/common/resource.h
@@ -28,6 +28,7 @@ namespace Sink {
 class FacadeFactory;
 class AdaptorFactoryRegistry;
 class ResourceContext;
+class QueryBase;
 /**
 * Resource interface
@@ -44,7 +45,7 @@ public:
    /**
     * Execute synchronization with the source.
     */
-    virtual KAsync::Job<void> synchronizeWithSource();
+    virtual KAsync::Job<void> synchronizeWithSource(const Sink::QueryBase &);
    /**
     * Process all internal messages, thus ensuring the store is up to date and no pending modifications are left.
diff --git a/common/resourceaccess.cpp b/common/resourceaccess.cpp
index e509292..1847949 100644
--- a/common/resourceaccess.cpp
+++ b/common/resourceaccess.cpp
@@ -301,6 +301,25 @@ KAsync::Job<void> ResourceAccess::synchronizeResource(bool sourceSync, bool loca
    return sendCommand(Commands::SynchronizeCommand, fbb);
 }
+KAsync::Job<void> ResourceAccess::synchronizeResource(const Sink::QueryBase &query)
+{
+    flatbuffers::FlatBufferBuilder fbb;
+    QByteArray queryString;
+    {
+        QDataStream stream(&queryString, QIODevice::WriteOnly);
+        stream << query;
+    }
+    auto q = fbb.CreateString(queryString.toStdString());
+    auto builder = Sink::Commands::SynchronizeBuilder(fbb);
+    builder.add_sourceSync(true);
+    builder.add_localSync(false);
+    builder.add_query(q);
+    Sink::Commands::FinishSynchronizeBuffer(fbb, builder.Finish());
+    open();
+    return sendCommand(Commands::SynchronizeCommand, fbb);
+}
 KAsync::Job<void> ResourceAccess::sendCreateCommand(const QByteArray &uid, const QByteArray &resourceBufferType, const QByteArray &buffer)
 {
    flatbuffers::FlatBufferBuilder fbb;
diff --git a/common/resourceaccess.h b/common/resourceaccess.h
index 5d66246..95cb667 100644
--- a/common/resourceaccess.h
+++ b/common/resourceaccess.h
@@ -50,6 +50,7 @@ public:
    virtual KAsync::Job<void> sendCommand(int commandId) = 0;
    virtual KAsync::Job<void> sendCommand(int commandId, flatbuffers::FlatBufferBuilder &fbb) = 0;
    virtual KAsync::Job<void> synchronizeResource(bool remoteSync, bool localSync) = 0;
+    virtual KAsync::Job<void> synchronizeResource(const Sink::QueryBase &filter) = 0;
    virtual KAsync::Job<void> sendCreateCommand(const QByteArray &uid, const QByteArray &resourceBufferType, const QByteArray &buffer)
    {
@@ -107,6 +108,7 @@ public:
    KAsync::Job<void> sendCommand(int commandId) Q_DECL_OVERRIDE;
    KAsync::Job<void> sendCommand(int commandId, flatbuffers::FlatBufferBuilder &fbb) Q_DECL_OVERRIDE;
    KAsync::Job<void> synchronizeResource(bool remoteSync, bool localSync) Q_DECL_OVERRIDE;
+    KAsync::Job<void> synchronizeResource(const Sink::QueryBase &filter) Q_DECL_OVERRIDE;
    KAsync::Job<void> sendCreateCommand(const QByteArray &uid, const QByteArray &resourceBufferType, const QByteArray &buffer) Q_DECL_OVERRIDE;
    KAsync::Job<void>
    sendModifyCommand(const QByteArray &uid, qint64 revision, const QByteArray &resourceBufferType, const QByteArrayList &deletedProperties, const QByteArray &buffer, const QByteArrayList &changedProperties) Q_DECL_OVERRIDE;
diff --git a/common/store.cpp b/common/store.cpp
index 2ea5e22..41b4867 100644
--- a/common/store.cpp
+++ b/common/store.cpp
@@ -129,13 +129,8 @@ KAsync::Job<void> queryResource(const QByteArray resourceType, const QByteArray
 template <class DomainType>
 QSharedPointer<QAbstractItemModel> Store::loadModel(Query query)
 {
-    SinkTrace() << "Query: " << ApplicationDomain::getTypeName<DomainType>();
+    query.setType(ApplicationDomain::getTypeName<DomainType>());
-    SinkTrace() << "  Requested: " << query.requestedProperties;
+    SinkTrace() << "Loading model: " << query;
-    SinkTrace() << "  Filter: " << query.getBaseFilters();
-    SinkTrace() << "  Parent: " << query.parentProperty;
-    SinkTrace() << "  Ids: " << query.ids();
-    SinkTrace() << "  IsLive: " << query.liveQuery();
-    SinkTrace() << "  Sorting: " << query.sortProperty();
    auto model = QSharedPointer<ModelResult<DomainType, typename DomainType::Ptr>>::create(query, query.requestedProperties);
    //* Client defines lifetime of model
@@ -276,6 +271,27 @@ KAsync::Job<void> Store::synchronize(const Sink::Query &query)
        });
 }
+KAsync::Job<void> Store::synchronize(const Sink::SyncScope &scope)
+{
+    auto resources = getResources(scope.getResourceFilter()).keys();
+    SinkTrace() << "synchronize" << resources;
+    return KAsync::value(resources)
+        .template each([scope](const QByteArray &resource) {
+            SinkTrace() << "Synchronizing " << resource;
+            auto resourceAccess = ResourceAccessFactory::instance().getAccess(resource, ResourceConfig::getResourceType(resource));
+            return resourceAccess->synchronizeResource(scope)
+                .addToContext(resourceAccess)
+                .then<void>([](const KAsync::Error &error) {
+                        if (error) {
+                            SinkWarning() << "Error during sync.";
+                            return KAsync::error<void>(error);
+                        }
+                        SinkTrace() << "synced.";
+                        return KAsync::null<void>();
+                    });
+        });
+}
 template <class DomainType>
 KAsync::Job<DomainType> Store::fetchOne(const Sink::Query &query)
 {
diff --git a/common/store.h b/common/store.h
index 571ffff..931e473 100644
--- a/common/store.h
+++ b/common/store.h
@@ -86,6 +86,8 @@ KAsync::Job<void> SINK_EXPORT remove(const DomainType &domainObject);
 */
 KAsync::Job<void> SINK_EXPORT synchronize(const Sink::Query &query);
+KAsync::Job<void> SINK_EXPORT synchronize(const Sink::SyncScope &query);
 /**
 * Removes all resource data from disk.
 *
diff --git a/common/synchronizer.cpp b/common/synchronizer.cpp
index 206cf5e..85c68e4 100644
--- a/common/synchronizer.cpp
+++ b/common/synchronizer.cpp
@@ -228,12 +228,12 @@ void Synchronizer::modify(const DomainType &entity)
    modifyEntity(entity.identifier(), entity.revision(), ApplicationDomain::getTypeName<DomainType>(), entity);
 }
-KAsync::Job<void> Synchronizer::synchronize()
+KAsync::Job<void> Synchronizer::synchronize(const Sink::QueryBase &query)
 {
    SinkTrace() << "Synchronizing";
    mSyncInProgress = true;
    mMessageQueue->startTransaction();
-    return synchronizeWithSource().syncThen<void>([this]() {
+    return synchronizeWithSource(query).syncThen<void>([this]() {
        mSyncStore.clear();
        mMessageQueue->commit();
        mSyncInProgress = false;
diff --git a/common/synchronizer.h b/common/synchronizer.h
index 12bb587..c03c425 100644
--- a/common/synchronizer.h
+++ b/common/synchronizer.h
@@ -41,7 +41,7 @@ public:
    virtual ~Synchronizer();
    void setup(const std::function<void(int commandId, const QByteArray &data)> &enqueueCommandCallback, MessageQueue &messageQueue);
-    KAsync::Job<void> synchronize();
+    KAsync::Job<void> synchronize(const Sink::QueryBase &query);
    //Read only access to main storage
    Storage::EntityStore &store();
@@ -91,7 +91,7 @@ protected:
    // template <typename DomainType>
    // void remove(const DomainType &entity);
-    virtual KAsync::Job<void> synchronizeWithSource() = 0;
+    virtual KAsync::Job<void> synchronizeWithSource(const Sink::QueryBase &query) = 0;
 private:
    void modifyIfChanged(Storage::EntityStore &store, const QByteArray &bufferType, const QByteArray &sinkId, const Sink::ApplicationDomain::ApplicationDomainType &entity);
diff --git a/examples/dummyresource/resourcefactory.cpp b/examples/dummyresource/resourcefactory.cpp
index 242a772..46e67f3 100644
--- a/examples/dummyresource/resourcefactory.cpp
+++ b/examples/dummyresource/resourcefactory.cpp
@@ -109,7 +109,7 @@ class DummySynchronizer : public Sink::Synchronizer {
        SinkTrace() << "Sync of " << count << " entities of type " << bufferType << " done." << Sink::Log::TraceTime(time->elapsed());
    }
-    KAsync::Job<void> synchronizeWithSource() Q_DECL_OVERRIDE
+    KAsync::Job<void> synchronizeWithSource(const Sink::QueryBase &) Q_DECL_OVERRIDE
    {
        SinkLog() << " Synchronizing with the source";
        return KAsync::syncStart<void>([this]() {
@@ -146,7 +146,7 @@ DummyResource::~DummyResource()
 }
-KAsync::Job<void> DummyResource::synchronizeWithSource()
+KAsync::Job<void> DummyResource::synchronizeWithSource(const Sink::QueryBase &query)
 {
    SinkTrace() << "Synchronize with source and sending a notification about it";
    Sink::Notification n;
@@ -155,7 +155,7 @@ KAsync::Job<void> DummyResource::synchronizeWithSource()
    n.message = "We're connected";
    n.code = Sink::ApplicationDomain::ConnectedStatus;
    emit notify(n);
-    return GenericResource::synchronizeWithSource();
+    return GenericResource::synchronizeWithSource(query);
 }
 KAsync::Job<void> DummyResource::inspect(int inspectionType, const QByteArray &inspectionId, const QByteArray &domainType, const QByteArray &entityId, const QByteArray &property, const QVariant &expectedValue)
diff --git a/examples/dummyresource/resourcefactory.h b/examples/dummyresource/resourcefactory.h
index 46a557e..8ef27a6 100644
--- a/examples/dummyresource/resourcefactory.h
+++ b/examples/dummyresource/resourcefactory.h
@@ -32,7 +32,7 @@ public:
    DummyResource(const Sink::ResourceContext &resourceContext, const QSharedPointer<Sink::Pipeline> &pipeline = QSharedPointer<Sink::Pipeline>());
    virtual ~DummyResource();
-    KAsync::Job<void> synchronizeWithSource() Q_DECL_OVERRIDE;
+    KAsync::Job<void> synchronizeWithSource(const Sink::QueryBase &) Q_DECL_OVERRIDE;
    KAsync::Job<void> inspect(int inspectionType, const QByteArray &inspectionId, const QByteArray &domainType, const QByteArray &entityId, const QByteArray &property, const QVariant &expectedValue) Q_DECL_OVERRIDE;
 };
diff --git a/examples/imapresource/imapresource.cpp b/examples/imapresource/imapresource.cpp
index 34a4376..d5a59b9 100644
--- a/examples/imapresource/imapresource.cpp
+++ b/examples/imapresource/imapresource.cpp
@@ -272,7 +272,7 @@ public:
    }
-    KAsync::Job<void> synchronizeWithSource() Q_DECL_OVERRIDE
+    KAsync::Job<void> synchronizeWithSource(const Sink::QueryBase &query) Q_DECL_OVERRIDE
    {
        SinkLog() << " Synchronizing";
        return KAsync::start<void>([this]() {
diff --git a/examples/maildirresource/maildirresource.cpp b/examples/maildirresource/maildirresource.cpp
index 23920d4..820ec2f 100644
--- a/examples/maildirresource/maildirresource.cpp
+++ b/examples/maildirresource/maildirresource.cpp
@@ -330,7 +330,7 @@ public:
        SinkLog() << "Synchronized " << count << " mails in " << listingPath << Sink::Log::TraceTime(elapsed) << " " << elapsed/qMax(count, 1) << " [ms/mail]";
    }
-    KAsync::Job<void> synchronizeWithSource() Q_DECL_OVERRIDE
+    KAsync::Job<void> synchronizeWithSource(const Sink::QueryBase &query) Q_DECL_OVERRIDE
    {
        SinkLog() << " Synchronizing";
        return KAsync::start<void>([this]() {
diff --git a/examples/mailtransportresource/mailtransportresource.cpp b/examples/mailtransportresource/mailtransportresource.cpp
index 5f8ba89..d1a2e77 100644
--- a/examples/mailtransportresource/mailtransportresource.cpp
+++ b/examples/mailtransportresource/mailtransportresource.cpp
@@ -108,7 +108,7 @@ public:
        return KAsync::null<void>();
    }
-    KAsync::Job<void> synchronizeWithSource() Q_DECL_OVERRIDE
+    KAsync::Job<void> synchronizeWithSource(const Sink::QueryBase &query) Q_DECL_OVERRIDE
    {
        SinkLog() << " Synchronizing";
        return KAsync::start<void>([this](KAsync::Future<void> future) {
diff --git a/tests/dummyresourcetest.cpp b/tests/dummyresourcetest.cpp
index f8d8543..eea63c0 100644
--- a/tests/dummyresourcetest.cpp
+++ b/tests/dummyresourcetest.cpp
@@ -136,7 +136,7 @@ private slots:
    void testResourceSync()
    {
        ::DummyResource resource(getContext());
-        auto job = resource.synchronizeWithSource();
+        auto job = resource.synchronizeWithSource(Sink::QueryBase());
        // TODO pass in optional timeout?
        auto future = job.exec();
        future.waitForFinished();
diff --git a/tests/maildirsyncbenchmark.cpp b/tests/maildirsyncbenchmark.cpp
index d84c758..ab09395 100644
--- a/tests/maildirsyncbenchmark.cpp
+++ b/tests/maildirsyncbenchmark.cpp
@@ -87,7 +87,7 @@ private slots:
        MaildirResource resource("sink.maildir.test1", pipeline);
        QTime time;
        time.start();
-        resource.Sink::GenericResource::synchronizeWithSource().exec().waitForFinished();
+        resource.Sink::GenericResource::synchronizeWithSource(Sink::QueryBase()).exec().waitForFinished();
        std::cout << "Sync took " << time.elapsed() << std::endl;
        resource.processAllMessages().exec().waitForFinished();
        const auto allProcessedTime = time.elapsed();
diff --git a/tests/testimplementations.h b/tests/testimplementations.h
index cf7a3da..111c884 100644
--- a/tests/testimplementations.h
+++ b/tests/testimplementations.h
@@ -70,6 +70,10 @@ public:
    {
        return KAsync::null<void>();
    }
+    KAsync::Job<void> synchronizeResource(const Sink::QueryBase &) Q_DECL_OVERRIDE
+    {
+        return KAsync::null<void>();
+    }
 public slots:
    void open() Q_DECL_OVERRIDE
@@ -111,7 +115,7 @@ public:
    {
    }
-    KAsync::Job<void> synchronizeWithSource() Q_DECL_OVERRIDE
+    KAsync::Job<void> synchronizeWithSource(const Sink::QueryBase &query) Q_DECL_OVERRIDE
    {
        return KAsync::null<void>();
    }