6 files changed, 167 insertions, 53 deletions

diff --git a/common/datastorequery.cpp b/common/datastorequery.cpp
index f352b74..0fc9234 100644
--- a/common/datastorequery.cpp
+++ b/common/datastorequery.cpp
@@ -148,14 +148,9 @@ public:
     QHash<QByteArray, QVariant> mAggregateValues;
     QByteArray mReductionProperty;
     QByteArray mSelectionProperty;
-    enum SelectionComparator {
-        Max
-        /* Min, */
-        /* First */
-    };
-    SelectionComparator mSelectionComparator;
+    Query::Reduce::Selector::Comparator mSelectionComparator;
 
-    Reduce(const QByteArray &reductionProperty, const QByteArray &selectionProperty, SelectionComparator comparator, FilterBase::Ptr source, DataStoreQuery *store)
+    Reduce(const QByteArray &reductionProperty, const QByteArray &selectionProperty, Query::Reduce::Selector::Comparator comparator, FilterBase::Ptr source, DataStoreQuery *store)
         : FilterBase(source, store),
         mReductionProperty(reductionProperty),
         mSelectionProperty(selectionProperty),
@@ -177,9 +172,9 @@ public:
         return QByteArray();
     }
 
-    static bool compare(const QVariant &left, const QVariant &right, SelectionComparator comparator)
+    static bool compare(const QVariant &left, const QVariant &right, Query::Reduce::Selector::Comparator comparator)
     {
-        if (comparator == Max) {
+        if (comparator == Query::Reduce::Selector::Max) {
             return left > right;
         }
         return false;
@@ -412,13 +407,16 @@ void DataStoreQuery::setupQuery()
     /*     baseSet = Sort::Ptr::create(baseSet, mQuery.sortProperty); */
     /* } */
 
-    if (mQuery.threadLeaderOnly) {
-        auto reduce = Reduce::Ptr::create("threadId", "date", Reduce::Max, baseSet, this);
-        baseSet = reduce;
-    }
-    if (mQuery.bloomThread) {
-        auto reduce = Bloom::Ptr::create("threadId", baseSet, this);
-        baseSet = reduce;
+    for (const auto &stage : mQuery.filterStages) {
+        if (auto filter = stage.dynamicCast<Query::Filter>()) {
+
+        } else if (auto filter = stage.dynamicCast<Query::Reduce>()) {
+            auto reduce = Reduce::Ptr::create(filter->property, filter->selector.property, filter->selector.comparator, baseSet, this);
+            baseSet = reduce;
+        } else if (auto filter = stage.dynamicCast<Query::Bloom>()) {
+            auto reduce = Bloom::Ptr::create(filter->property, baseSet, this);
+            baseSet = reduce;
+        }
     }
 
     mCollector = Collector::Ptr::create(baseSet, this);
diff --git a/common/domain/applicationdomaintype.cpp b/common/domain/applicationdomaintype.cpp
index e6117aa..2f7c32b 100644
--- a/common/domain/applicationdomaintype.cpp
+++ b/common/domain/applicationdomaintype.cpp
@@ -29,6 +29,8 @@ SINK_DEBUG_AREA("applicationdomaintype");
 namespace Sink {
 namespace ApplicationDomain {
 
+constexpr const char *Mail::ThreadId::name;
+
 ApplicationDomainType::ApplicationDomainType()
     :mAdaptor(new MemoryBufferAdaptor())
 {
diff --git a/common/query.h b/common/query.h
index 1b909e5..3021fe2 100644
--- a/common/query.h
+++ b/common/query.h
@@ -174,34 +174,13 @@ public:
         return *this;
     }
 
-    template <typename T>
-    Query &filter(const QVariant &value)
-    {
-        propertyFilter.insert(T::name, value);
-        return *this;
-    }
-
-    template <typename T>
-    Query &filter(const Comparator &comparator)
-    {
-        propertyFilter.insert(T::name, comparator);
-        return *this;
-    }
-
-    template <typename T>
-    Query &filter(const ApplicationDomain::Entity &value)
-    {
-        propertyFilter.insert(T::name, QVariant::fromValue(value.identifier()));
-        return *this;
-    }
-
-    Query(const ApplicationDomain::Entity &value) : limit(0), liveQuery(false), synchronousQuery(false), threadLeaderOnly(false), bloomThread(false)
+    Query(const ApplicationDomain::Entity &value) : limit(0), liveQuery(false), synchronousQuery(false)
     {
         ids << value.identifier();
         resources << value.resourceInstanceIdentifier();
     }
 
-    Query(Flags flags = Flags()) : limit(0), liveQuery(false), synchronousQuery(false), threadLeaderOnly(false), bloomThread(false)
+    Query(Flags flags = Flags()) : limit(0), liveQuery(false), synchronousQuery(false)
     {
     }
 
@@ -236,9 +215,137 @@ public:
     int limit;
     bool liveQuery;
     bool synchronousQuery;
-    bool threadLeaderOnly;
-    bool bloomThread;
+
+    class FilterStage {
+    public:
+        virtual ~FilterStage(){};
+    };
+
+    QList<QSharedPointer<FilterStage>> filterStages;
+
+    /*
+    * Filters
+    */
+    class Filter : public FilterStage {
+        QByteArrayList ids;
+        QHash<QByteArray, Comparator> propertyFilter;
+        QByteArray sortProperty;
+    };
+
+    template <typename T>
+    Query &filter(const QVariant &value)
+    {
+        propertyFilter.insert(T::name, value);
+        return *this;
+    }
+
+    template <typename T>
+    Query &filter(const Comparator &comparator)
+    {
+        propertyFilter.insert(T::name, comparator);
+        return *this;
+    }
+
+    template <typename T>
+    Query &filter(const ApplicationDomain::Entity &value)
+    {
+        propertyFilter.insert(T::name, QVariant::fromValue(value.identifier()));
+        return *this;
+    }
+
+    Query &filter(const ApplicationDomain::SinkResource &resource)
+    {
+        resources << resource.identifier();
+        return *this;
+    }
+
+    Query &filter(const ApplicationDomain::SinkAccount &account)
+    {
+        accounts << account.identifier();
+        return *this;
+    }
+
+    class Reduce : public FilterStage {
+    public:
+
+        class Selector {
+        public:
+            enum Comparator {
+                Min, //get the minimum value
+                Max, //get the maximum value
+                First //Get the first result we get
+            };
+
+            template <typename SelectionProperty>
+            static Selector max()
+            {
+                return Selector(SelectionProperty::name, Max);
+            }
+
+            Selector(const QByteArray &p, Comparator c)
+                : property(p),
+                comparator(c)
+            {
+            }
+
+            QByteArray property;
+            Comparator comparator;
+        };
+
+        Reduce(const QByteArray &p, const Selector &s)
+            : property(p),
+            selector(s)
+        {
+        }
+
+        //Reduce on property
+        QByteArray property;
+        Selector selector;
+
+        //TODO add aggregate functions like:
+        //.count()
+        //.collect<Mail::sender>();
+        //...
+        //
+        //Potentially pass-in an identifier under which the result will be available in the result set.
+    };
+
+    template <typename T>
+    Reduce &reduce(const Reduce::Selector &s)
+    {
+        auto reduction = QSharedPointer<Reduce>::create(T::name, s);
+        filterStages << reduction;
+        return *reduction;
+    }
+
+    /**
+    * "Bloom" on a property.
+    *
+    * For every encountered value of a property,
+    * a result set is generated containing all entries with the same value.
+    *
+    * Example:
+    * For an input result set of one mail; return all emails with the same threadId.
+    */
+    class Bloom : public FilterStage {
+    public:
+        //Property to bloom on
+        QByteArray property;
+        Bloom(const QByteArray &p)
+            : property(p)
+        {
+        }
+    };
+
+    template <typename T>
+    void bloom()
+    {
+        auto bloom = QSharedPointer<Bloom>::create(T::name);
+        filterStages << bloom;
+    }
+
 };
+
 }
 
 QDebug operator<<(QDebug dbg, const Sink::Query::Comparator &c);
diff --git a/examples/maildirresource/tests/maildirmailsynctest.cpp b/examples/maildirresource/tests/maildirmailsynctest.cpp
index bafd5a8..2269362 100644
--- a/examples/maildirresource/tests/maildirmailsynctest.cpp
+++ b/examples/maildirresource/tests/maildirmailsynctest.cpp
@@ -22,8 +22,7 @@
 #include "../maildirresource.h"
 #include "../libmaildir/maildir.h"
 
-#include "common/test.h"
-#include "common/domain/applicationdomaintype.h"
+#include "test.h"
 
 #include "utils.h"
 
diff --git a/examples/maildirresource/tests/maildirthreadtest.cpp b/examples/maildirresource/tests/maildirthreadtest.cpp
index 69d51a6..5a23ccf 100644
--- a/examples/maildirresource/tests/maildirthreadtest.cpp
+++ b/examples/maildirresource/tests/maildirthreadtest.cpp
@@ -23,7 +23,7 @@
 #include "../libmaildir/maildir.h"
 
 #include "common/test.h"
-#include "common/domain/applicationdomaintype.h"
+#include "applicationdomaintype.h"
 
 #include "utils.h"
 
diff --git a/tests/mailthreadtest.cpp b/tests/mailthreadtest.cpp
index 89e5a85..e9fe499 100644
--- a/tests/mailthreadtest.cpp
+++ b/tests/mailthreadtest.cpp
@@ -66,10 +66,10 @@ void MailThreadTest::init()
 void MailThreadTest::testListThreadLeader()
 {
     Sink::Query query;
-    query.resources << mResourceInstanceIdentifier;
+    query.filter(SinkResource(mResourceInstanceIdentifier));
     query.request<Mail::Subject>().request<Mail::MimeMessage>().request<Mail::Folder>().request<Mail::Date>();
-    query.threadLeaderOnly = true;
     query.sort<Mail::Date>();
+    query.reduce<Mail::ThreadId>(Query::Reduce::Selector::max<Mail::Date>());
 
     // Ensure all local data is processed
     VERIFYEXEC(Store::synchronize(query));
@@ -127,11 +127,11 @@ void MailThreadTest::testIndexInMixedOrder()
     VERIFYEXEC(ResourceControl::flushMessageQueue(QByteArrayList() << mResourceInstanceIdentifier));
 
     Sink::Query query;
-    query.resources << mResourceInstanceIdentifier;
+    query.filter(SinkResource(mResourceInstanceIdentifier));
     query.request<Mail::Subject>().request<Mail::MimeMessage>().request<Mail::Folder>().request<Mail::Date>();
-    query.threadLeaderOnly = true;
-    query.sort<Mail::Date>();
     query.filter<Mail::Folder>(folder);
+    query.sort<Mail::Date>();
+    query.reduce<Mail::ThreadId>(Query::Reduce::Selector::max<Mail::Date>());
 
     Mail threadLeader;
 
@@ -147,6 +147,14 @@ void MailThreadTest::testIndexInMixedOrder()
         VERIFYEXEC(job);
     }
 
+    {
+        auto mail = Mail::create(mResourceInstanceIdentifier);
+        mail.setMimeMessage(message2->encodedContent());
+        mail.setFolder(folder);
+        VERIFYEXEC(Store::create(mail));
+    }
+    VERIFYEXEC(ResourceControl::flushMessageQueue(QByteArrayList() << mResourceInstanceIdentifier));
+
     //Ensure we find the thread leader still
     {
         auto job = Store::fetchAll<Mail>(query)
@@ -169,15 +177,15 @@ void MailThreadTest::testIndexInMixedOrder()
     //Ensure the thread is complete
     {
         Sink::Query query;
-        query.resources << mResourceInstanceIdentifier;
+        query.filter(SinkResource(mResourceInstanceIdentifier));
+        query.ids << threadLeader.identifier();
         query.request<Mail::Subject>().request<Mail::MimeMessage>().request<Mail::Folder>().request<Mail::Date>();
-        query.bloomThread = true;
         query.sort<Mail::Date>();
-        query.ids << threadLeader.identifier();
+        query.bloom<Mail::ThreadId>();
 
         auto job = Store::fetchAll<Mail>(query)
             .syncThen<void, QList<Mail::Ptr>>([=](const QList<Mail::Ptr> &mails) {
-                QCOMPARE(mails.size(), 2);
+                QCOMPARE(mails.size(), 3);
                 auto mail = *mails.first();
                 QCOMPARE(mail.getSubject(), QString::fromLatin1("Re: Re: 1"));
             });