Keep thread alive until the end of the query, and cleanup the resultSet.

2 files changed, 97 insertions, 25 deletions

diff --git a/common/clientapi.h b/common/clientapi.h
index c6c43ee..4786398 100644
--- a/common/clientapi.h
+++ b/common/clientapi.h
@@ -107,7 +107,7 @@ namespace async {
         {
             if (!mResultEmitter) {
                 //We have to go over a separate var and return that, otherwise we'd delete the emitter immediately again
-                auto sharedPtr = QSharedPointer<ResultEmitter<T> >::create();
+                auto sharedPtr = QSharedPointer<ResultEmitter<T> >(new ResultEmitter<T>, [this](ResultEmitter<T> *emitter){ done(); delete emitter; });
                 mResultEmitter = sharedPtr;
                 return sharedPtr;
             }
@@ -124,9 +124,30 @@ namespace async {
             mQueryRunner = runner;
         }
 
+        void onDone(const std::function<void()> &callback)
+        {
+            mOnDoneCallback = callback;
+        }
+
+        bool isDone() const
+        {
+            //The existance of the emitter currently defines wether we're done or not.
+            return mResultEmitter.toStrongRef().isNull();
+        }
+
     private:
+        void done()
+        {
+            qWarning() << "done";
+            if (mOnDoneCallback) {
+                mOnDoneCallback();
+            }
+            mOnDoneCallback = std::function<void()>();
+        }
+
         QWeakPointer<ResultEmitter<T> > mResultEmitter;
         QSharedPointer<QObject> mQueryRunner;
+        std::function<void()> mOnDoneCallback;
     };
 
     /*
@@ -482,31 +503,46 @@ public:
     static QSharedPointer<ResultEmitter<typename DomainType::Ptr> > load(Query query)
     {
         auto resultSet = QSharedPointer<ResultProvider<typename DomainType::Ptr> >::create();
+        // FIXME This is ridiculous but otherwise I can't release the shared pointer before the thread quits
+        auto resultSetPtr = QSharedPointer<QSharedPointer<ResultProvider<typename DomainType::Ptr> > >::create(resultSet);
 
         //Execute the search in a thread.
         //We must guarantee that the emitter is returned before the first result is emitted.
         //The result provider must be threadsafe.
-        async::run([resultSet, query](){
+        async::run([resultSetPtr, query](){
             // Query all resources and aggregate results
             const QList<QByteArray> resources = query.resources;
             Async::start<QList<QByteArray>>([resources](){return resources;})
-            .template each<void, QByteArray>([query, resultSet](const QByteArray &resource, Async::Future<void> &future) {
+            .template each<void, QByteArray>([query, resultSetPtr](const QByteArray &resource, Async::Future<void> &future) {
+                //TODO pass resource identifier to factory
                 auto facade = FacadeFactory::instance().getFacade<DomainType>(resource);
-                // TODO The following is a necessary hack to keep the facade alive.
-                // Otherwise this would reduce to:
-                // facade->load(query, addCallback).exec();
-                // We somehow have to guarantee that the facade remains valid for the duration of the job
-                // TODO: Use one result set per facade, and merge the results separately
-                // resultSet->addSubset(facade->query(query));
-                facade->load(query, resultSet).template then<void>([&future, facade]() {
+                if (auto resultSet = *resultSetPtr) {
+                    facade->load(query, resultSet).template then<void>([&future](){future.setFinished();}).exec();
+                } else {
+                    qWarning() << "result set is already gone";
                     future.setFinished();
-                }).exec();
-            }).template then<void>([resultSet]() {
+                }
+            }).template then<void>([resultSetPtr]() {
                 qDebug() << "Query complete";
-                resultSet->complete();
-            }).exec().waitForFinished(); //We use the eventloop provided by waitForFinished to keep the thread alive until all is done
-            //FIXME for live query the thread dies after the initial query?
-            //TODO associate the thread with the query runner
+                if (auto resultSet = *resultSetPtr) {
+                    resultSet->complete();
+                } else {
+                    qWarning() << "result set is already gone";
+                }
+            }).exec();
+
+            if (auto resultSet = *resultSetPtr) {
+                resultSetPtr->clear();
+                if (!resultSet->isDone()) {
+                    QEventLoop eventLoop;
+                    resultSet->onDone([&eventLoop](){
+                        eventLoop.quit();
+                    });
+                    eventLoop.exec();
+                }
+            } else {
+                qWarning() << "result set is already gone";
+            }
         });
         return resultSet->emitter();
     }
diff --git a/tests/clientapitest.cpp b/tests/clientapitest.cpp
index 7bcd7b0..057495e 100644
--- a/tests/clientapitest.cpp
+++ b/tests/clientapitest.cpp
@@ -24,10 +24,11 @@ class DummyResourceFacade : public Akonadi2::StoreFacade<Akonadi2::ApplicationDo
 {
 public:
     ~DummyResourceFacade(){};
-    virtual Async::Job<void> create(const Akonadi2::ApplicationDomain::Event &domainObject){ return Async::null<void>(); };
-    virtual Async::Job<void> modify(const Akonadi2::ApplicationDomain::Event &domainObject){ return Async::null<void>(); };
-    virtual Async::Job<void> remove(const Akonadi2::ApplicationDomain::Event &domainObject){ return Async::null<void>(); };
-    virtual Async::Job<qint64> load(const Akonadi2::Query &query, const std::function<void(const Akonadi2::ApplicationDomain::Event::Ptr &)> &resultCallback)
+    Async::Job<void> create(const Akonadi2::ApplicationDomain::Event &domainObject) Q_DECL_OVERRIDE { return Async::null<void>(); };
+    Async::Job<void> modify(const Akonadi2::ApplicationDomain::Event &domainObject) Q_DECL_OVERRIDE { return Async::null<void>(); };
+    Async::Job<void> remove(const Akonadi2::ApplicationDomain::Event &domainObject) Q_DECL_OVERRIDE { return Async::null<void>(); };
+
+    Async::Job<qint64> load(const Akonadi2::Query &query, const std::function<void(const Akonadi2::ApplicationDomain::Event::Ptr &)> &resultCallback)
     {
         return Async::start<qint64>([this, resultCallback](Async::Future<qint64> &future) {
             qDebug() << "load called";
@@ -39,14 +40,23 @@ public:
         });
     }
 
-    Async::Job<void> load(const Akonadi2::Query &query, const QSharedPointer<Akonadi2::ResultProvider<Akonadi2::ApplicationDomain::Event::Ptr> > &resultProvider)
+    Async::Job<void> load(const Akonadi2::Query &query, const QSharedPointer<Akonadi2::ResultProvider<Akonadi2::ApplicationDomain::Event::Ptr> > &resultProvider) Q_DECL_OVERRIDE 
     {
         auto runner = QSharedPointer<QueryRunner>::create(query);
         //The runner only lives as long as the resultProvider
         resultProvider->setQueryRunner(runner);
-        runner->setQuery([this, resultProvider, query](qint64 oldRevision, qint64 newRevision) -> Async::Job<qint64> {
+        QWeakPointer<Akonadi2::ResultProvider<Akonadi2::ApplicationDomain::Event::Ptr> > weakResultProvider = resultProvider;
+        capturedResultProvider = resultProvider;
+        runner->setQuery([this, weakResultProvider, query](qint64 oldRevision, qint64 newRevision) -> Async::Job<qint64> {
             qDebug() << "Creating query for revisions: " << oldRevision << newRevision;
-            return Async::start<qint64>([this, resultProvider, query](Async::Future<qint64> &future) {
+            return Async::start<qint64>([this, weakResultProvider, query](Async::Future<qint64> &future) {
+                auto resultProvider = weakResultProvider.toStrongRef();
+                if (!resultProvider) {
+                    Warning() << "Tried executing query after result provider is already gone";
+                    future.setError(0, QString());
+                    future.setFinished();
+                    return;
+                }
                 //TODO only emit changes and don't replace everything
                 resultProvider->clear();
                 //rerun query
@@ -84,6 +94,7 @@ public:
 
     QList<Akonadi2::ApplicationDomain::Event::Ptr> results;
     QSharedPointer<RevisionNotifier> notifier;
+    QWeakPointer<Akonadi2::ResultProvider<Akonadi2::ApplicationDomain::Event::Ptr> > capturedResultProvider;
 };
 
 class ClientAPITest : public QObject
@@ -123,8 +134,8 @@ private Q_SLOTS:
         facade.results << QSharedPointer<Akonadi2::ApplicationDomain::Event>::create("resource", "id", 0, QSharedPointer<Akonadi2::ApplicationDomain::BufferAdaptor>());
 
         Akonadi2::FacadeFactory::instance().registerFacade<Akonadi2::ApplicationDomain::Event, DummyResourceFacade>("dummyresource",
-            [&facade](bool &externallManage){
-                externallManage = true;
+            [&facade](bool &externallyManaged){
+                externallyManaged = true;
                 return &facade;
             }
         );
@@ -145,6 +156,31 @@ private Q_SLOTS:
         QTRY_COMPARE(result.size(), 2);
     }
 
+    void testQueryLifetime()
+    {
+        DummyResourceFacade facade;
+        facade.results << QSharedPointer<Akonadi2::ApplicationDomain::Event>::create("resource", "id", 0, QSharedPointer<Akonadi2::ApplicationDomain::BufferAdaptor>());
+
+        Akonadi2::FacadeFactory::instance().registerFacade<Akonadi2::ApplicationDomain::Event, DummyResourceFacade>("dummyresource",
+            [&facade](bool &externallyManaged){
+                externallyManaged = true;
+                return &facade;
+            }
+        );
+
+        Akonadi2::Query query;
+        query.resources << "dummyresource";
+        query.liveQuery = true;
+
+        {
+            async::SyncListResult<Akonadi2::ApplicationDomain::Event::Ptr> result(Akonadi2::Store::load<Akonadi2::ApplicationDomain::Event>(query));
+            result.exec();
+            QCOMPARE(result.size(), 1);
+        }
+        //It's running in a separate thread, so we have to wait for a moment.
+        QTRY_VERIFY(!facade.capturedResultProvider);
+    }
+
 };
 
 QTEST_MAIN(ClientAPITest)