13 files changed, 233 insertions, 262 deletions

diff --git a/common/changereplay.cpp b/common/changereplay.cpp
index fbd556f..e3b7158 100644
--- a/common/changereplay.cpp
+++ b/common/changereplay.cpp
@@ -72,7 +72,7 @@ KAsync::Job<void> ChangeReplay::replayNextRevision()
 {
     auto lastReplayedRevision = QSharedPointer<qint64>::create(0);
     auto topRevision = QSharedPointer<qint64>::create(0);
-    return KAsync::start<void>([this, lastReplayedRevision, topRevision]() {
+    return KAsync::syncStart<void>([this, lastReplayedRevision, topRevision]() {
             mReplayInProgress = true;
             mMainStoreTransaction = mStorage.createTransaction(Storage::ReadOnly, [](const Sink::Storage::Error &error) {
                 SinkWarning() << error.message;
@@ -90,11 +90,9 @@ KAsync::Job<void> ChangeReplay::replayNextRevision()
             SinkTrace() << "Changereplay from " << *lastReplayedRevision << " to " << *topRevision;
         })
         .then(KAsync::dowhile(
-            [this, lastReplayedRevision, topRevision](KAsync::Future<bool> &future) {
+            [this, lastReplayedRevision, topRevision]() -> KAsync::Job<KAsync::ControlFlowFlag> {
                     if (*lastReplayedRevision >= *topRevision) {
-                        future.setValue(false);
-                        future.setFinished();
-                        return;
+                        return KAsync::value(KAsync::Break);
                     }
 
                     qint64 revision = *lastReplayedRevision + 1;
@@ -109,12 +107,15 @@ KAsync::Job<void> ChangeReplay::replayNextRevision()
                                 [&lastReplayedRevision, type, this, &replayJob, &exitLoop, revision](const QByteArray &key, const QByteArray &value) -> bool {
                                     SinkTrace() << "Replaying " << key;
                                     if (canReplay(type, key, value)) {
-                                        replayJob = replay(type, key, value).then<void>([this, revision, lastReplayedRevision]() {
-                                            recordReplayedRevision(revision);
-                                            *lastReplayedRevision = revision;
-                                        },
-                                        [revision](int, QString) {
-                                            SinkTrace() << "Change replay failed" << revision;
+                                        replayJob = replay(type, key, value).then<void>([this, revision, lastReplayedRevision](const KAsync::Error &error) {
+                                            if (error) {
+                                                SinkTrace() << "Change replay failed" << revision;
+                                                return KAsync::error(error);
+                                            } else {
+                                                recordReplayedRevision(revision);
+                                                *lastReplayedRevision = revision;
+                                            }
+                                            return KAsync::null();
                                         });
                                         exitLoop = true;
                                     } else {
@@ -128,23 +129,26 @@ KAsync::Job<void> ChangeReplay::replayNextRevision()
                         }
                         revision++;
                     }
-                    replayJob.then<void>([this, revision, lastReplayedRevision, topRevision, &future]() {
-                        SinkTrace() << "Replayed until " << revision;
-                        recordReplayedRevision(*lastReplayedRevision);
-                        QTimer::singleShot(0, [&future, lastReplayedRevision, topRevision]() {
-                            future.setValue((*lastReplayedRevision < *topRevision));
-                            future.setFinished();
-                        });
-                    },
-                    [this, revision, &future](int, QString) {
-                        SinkTrace() << "Change replay failed" << revision;
-                        //We're probably not online or so, so postpone retrying
-                        future.setValue(false);
-                        future.setFinished();
-                    }).exec();
-
+                    return replayJob.then<KAsync::ControlFlowFlag>([this, revision, lastReplayedRevision, topRevision](const KAsync::Error &error) ->KAsync::Job<KAsync::ControlFlowFlag> {
+                        if (error) {
+                            SinkTrace() << "Change replay failed" << revision;
+                            //We're probably not online or so, so postpone retrying
+                            return KAsync::value(KAsync::Break);
+                        } else {
+                            SinkTrace() << "Replayed until " << revision;
+                            recordReplayedRevision(*lastReplayedRevision);
+                            if (*lastReplayedRevision < *topRevision) {
+                                return KAsync::wait(0).then(KAsync::value(KAsync::Continue));
+                            } else {
+                                return KAsync::value(KAsync::Break);
+                            }
+                        }
+                        //We shouldn't ever get here
+                        Q_ASSERT(false);
+                        return KAsync::value(KAsync::Break);
+                    });
             }))
-        .then<void>([this, lastReplayedRevision]() {
+        .syncThen<void>([this, lastReplayedRevision]() {
             recordReplayedRevision(*lastReplayedRevision);
             mMainStoreTransaction.abort();
             if (allChangesReplayed()) {
diff --git a/common/genericresource.cpp b/common/genericresource.cpp
index 7136882..f5b1775 100644
--- a/common/genericresource.cpp
+++ b/common/genericresource.cpp
@@ -100,7 +100,7 @@ private slots:
         }
         mProcessingLock = true;
         auto job = processPipeline()
-                       .then<void>([this]() {
+                       .syncThen<void>([this]() {
                            mProcessingLock = false;
                            if (messagesToProcessAvailable()) {
                                process();
@@ -122,7 +122,8 @@ private slots:
                 return mPipeline->newEntity(queuedCommand->command()->Data(), queuedCommand->command()->size());
             case Sink::Commands::InspectionCommand:
                 if (mInspect) {
-                    return mInspect(queuedCommand->command()->Data(), queuedCommand->command()->size()).then<qint64>([]() { return -1; });
+                    return mInspect(queuedCommand->command()->Data(), queuedCommand->command()->size())
+                        .syncThen<qint64>([]() { return -1; });
                 } else {
                     return KAsync::error<qint64>(-1, "Missing inspection command.");
                 }
@@ -131,7 +132,7 @@ private slots:
         }
     }
 
-    KAsync::Job<qint64, qint64> processQueuedCommand(const QByteArray &data)
+    KAsync::Job<qint64> processQueuedCommand(const QByteArray &data)
     {
         flatbuffers::Verifier verifyer(reinterpret_cast<const uint8_t *>(data.constData()), data.size());
         if (!Sink::VerifyQueuedCommandBuffer(verifyer)) {
@@ -143,13 +144,13 @@ private slots:
         SinkTrace() << "Dequeued Command: " << Sink::Commands::name(commandId);
         return processQueuedCommand(queuedCommand)
             .then<qint64, qint64>(
-                [this, commandId](qint64 createdRevision) -> qint64 {
+                [this, commandId](const KAsync::Error &error, qint64 createdRevision) -> KAsync::Job<qint64> {
+                    if (error) {
+                        SinkWarning() << "Error while processing queue command: " << error.errorMessage;
+                        return KAsync::error<qint64>(error);
+                    }
                     SinkTrace() << "Command pipeline processed: " << Sink::Commands::name(commandId);
-                    return createdRevision;
-                },
-                [](int errorCode, QString errorMessage) {
-                    // FIXME propagate error, we didn't handle it
-                    SinkWarning() << "Error while processing queue command: " << errorMessage;
+                    return KAsync::value<qint64>(createdRevision);
                 });
     }
 
@@ -157,31 +158,31 @@ private slots:
     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, time](KAsync::Future<void> &future) {
-                    queue->dequeueBatch(sBatchSize,
-                             [this, time](const QByteArray &data) {
-                                 time->start();
-                                 return KAsync::start<void>([this, data, time](KAsync::Future<void> &future) {
-                                     processQueuedCommand(data)
-                                         .then<void, qint64>([&future, this, time](qint64 createdRevision) {
-                                             SinkTrace() << "Created revision " << createdRevision << ". Processing took: " << Log::TraceTime(time->elapsed());
-                                             future.setFinished();
-                                         })
-                                         .exec();
-                                 });
-                             })
-                        .then<void>([&future, queue]() { future.setFinished(); },
-                            [&future](int i, QString error) {
-                                if (i != MessageQueue::ErrorCodes::NoMessageFound) {
-                                    SinkWarning() << "Error while getting message from messagequeue: " << error;
+        return KAsync::syncStart<void>([this]() { mPipeline->startTransaction(); })
+            .then(KAsync::dowhile(
+                [this, queue, time]() -> KAsync::Job<KAsync::ControlFlowFlag> {
+                    return queue->dequeueBatch(sBatchSize,
+                        [this, time](const QByteArray &data) -> KAsync::Job<void> {
+                            time->start();
+                            return processQueuedCommand(data)
+                            .syncThen<void, qint64>([this, time](qint64 createdRevision) {
+                                SinkTrace() << "Created revision " << createdRevision << ". Processing took: " << Log::TraceTime(time->elapsed());
+                            });
+                        })
+                        .then<KAsync::ControlFlowFlag>([queue](const KAsync::Error &error) {
+                                if (error) {
+                                    if (error.errorCode != MessageQueue::ErrorCodes::NoMessageFound) {
+                                        SinkWarning() << "Error while getting message from messagequeue: " << error.errorMessage;
+                                    }
                                 }
-                                future.setFinished();
-                            })
-                        .exec();
+                                if (queue->isEmpty()) {
+                                    return KAsync::value<KAsync::ControlFlowFlag>(KAsync::Break);
+                                } else {
+                                    return KAsync::value<KAsync::ControlFlowFlag>(KAsync::Continue);
+                                }
+                            });
                 }))
-            .then<void>([this]() { mPipeline->commit(); });
+            .syncThen<void>([this](const KAsync::Error &) { mPipeline->commit(); });
     }
 
     KAsync::Job<void> processPipeline()
@@ -198,18 +199,20 @@ private slots:
 
         // 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) {
+        return KAsync::dowhile(
+            [it, this]() {
                 auto time = QSharedPointer<QTime>::create();
                 time->start();
 
                 auto queue = it->next();
-                processQueue(queue)
-                    .then<void>([this, &future, time]() {
+                return processQueue(queue)
+                    .syncThen<KAsync::ControlFlowFlag>([this, time, it]() {
                         SinkTrace() << "Queue processed." << Log::TraceTime(time->elapsed());
-                        future.setFinished();
-                    })
-                    .exec();
+                        if (it->hasNext()) {
+                            return KAsync::Continue;
+                        }
+                        return KAsync::Break;
+                    });
             });
     }
 
@@ -251,22 +254,19 @@ GenericResource::GenericResource(const QByteArray &resourceType, const QByteArra
             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::Notification::Inspection;
-                        n.id = inspectionId;
-                        n.code = Sink::Notification::Success;
-                        emit notify(n);
-                    },
-                    [=](int code, const QString &message) {
-                        Warning_area("resource.inspection") << "Inspection failed: " << inspectionType << inspectionId << entityId << message;
+                    [=](const KAsync::Error &error) {
                         Sink::Notification n;
                         n.type = Sink::Notification::Inspection;
-                        n.message = message;
                         n.id = inspectionId;
-                        n.code = Sink::Notification::Failure;
+                        if (error) {
+                            Warning_area("resource.inspection") << "Inspection failed: " << inspectionType << inspectionId << entityId << error.errorMessage;
+                            n.code = Sink::Notification::Failure;
+                        } else {
+                            Log_area("resource.inspection") << "Inspection was successful: " << inspectionType << inspectionId << entityId;
+                            n.code = Sink::Notification::Success;
+                        }
                         emit notify(n);
+                        return KAsync::null();
                     })
                 .exec();
             return KAsync::null<void>();
@@ -420,7 +420,7 @@ void GenericResource::processCommand(int commandId, const QByteArray &data)
 
 KAsync::Job<void> GenericResource::synchronizeWithSource()
 {
-    return KAsync::start<void>([this](KAsync::Future<void> &future) {
+    return KAsync::start<void>([this]() {
 
         Sink::Notification n;
         n.id = "sync";
@@ -432,23 +432,21 @@ KAsync::Job<void> GenericResource::synchronizeWithSource()
         SinkLog() << " Synchronizing";
         // Changereplay would deadlock otherwise when trying to open the synchronization store
         enableChangeReplay(false);
-        mSynchronizer->synchronize()
-            .then<void>([this, &future]() {
-                SinkLog() << "Done Synchronizing";
-                Sink::Notification n;
-                n.id = "sync";
-                n.type = Sink::Notification::Status;
-                n.message = "Synchronization has ended.";
-                n.code = Sink::ApplicationDomain::ConnectedStatus;
-                emit notify(n);
-
-                enableChangeReplay(true);
-                future.setFinished();
-            }, [this, &future](int errorCode, const QString &error) {
+        return mSynchronizer->synchronize()
+            .then<void>([this](const KAsync::Error &error) {
                 enableChangeReplay(true);
-                future.setError(errorCode, error);
-            })
-            .exec();
+                if (!error) {
+                    SinkLog() << "Done Synchronizing";
+                    Sink::Notification n;
+                    n.id = "sync";
+                    n.type = Sink::Notification::Status;
+                    n.message = "Synchronization has ended.";
+                    n.code = Sink::ApplicationDomain::ConnectedStatus;
+                    emit notify(n);
+                    return KAsync::null();
+                }
+                return KAsync::error(error);
+            });
     });
 }
 
diff --git a/common/listener.cpp b/common/listener.cpp
index a051293..027d9ae 100644
--- a/common/listener.cpp
+++ b/common/listener.cpp
@@ -248,13 +248,17 @@ void Listener::processCommand(int commandId, uint messageId, const QByteArray &c
                 if (buffer->localSync()) {
                     job = job.then<void>(loadResource().processAllMessages());
                 }
-                job.then<void>([callback, timer]() {
-                       SinkTrace() << "Sync took " << Sink::Log::TraceTime(timer->elapsed());
-                       callback(true);
-                   }, [callback](int errorCode, const QString &msg) {
-                       SinkWarning() << "Sync failed: " << msg;
-                       callback(false);
-                   })
+                job.then<void>([callback, timer](const KAsync::Error &error) {
+                        if (error) {
+                            SinkWarning() << "Sync failed: " << error.errorMessage;
+                            callback(false);
+                            return KAsync::error(error);
+                        } else {
+                            SinkTrace() << "Sync took " << Sink::Log::TraceTime(timer->elapsed());
+                            callback(true);
+                            return KAsync::null();
+                        }
+                    })
                     .exec();
                 return;
             } else {
diff --git a/common/messagequeue.cpp b/common/messagequeue.cpp
index 3567a10..28eacb7 100644
--- a/common/messagequeue.cpp
+++ b/common/messagequeue.cpp
@@ -5,37 +5,6 @@
 
 SINK_DEBUG_AREA("messagequeue")
 
-static KAsync::Job<void> waitForCompletion(QList<KAsync::Future<void>> &futures)
-{
-    auto context = new QObject;
-    return KAsync::start<void>([futures, context](KAsync::Future<void> &future) {
-               const auto total = futures.size();
-               auto count = QSharedPointer<int>::create();
-               int i = 0;
-               for (KAsync::Future<void> subFuture : futures) {
-                   i++;
-                   if (subFuture.isFinished()) {
-                       *count += 1;
-                       continue;
-                   }
-                   // FIXME bind lifetime all watcher to future (repectively the main job
-                   auto watcher = QSharedPointer<KAsync::FutureWatcher<void>>::create();
-                   QObject::connect(watcher.data(), &KAsync::FutureWatcher<void>::futureReady, [count, total, &future]() {
-                       *count += 1;
-                       if (*count == total) {
-                           future.setFinished();
-                       }
-                   });
-                   watcher->setFuture(subFuture);
-                   context->setProperty(QString("future%1").arg(i).toLatin1().data(), QVariant::fromValue(watcher));
-               }
-               if (*count == total) {
-                   future.setFinished();
-               }
-           })
-        .then<void>([context]() { delete context; });
-}
-
 MessageQueue::MessageQueue(const QString &storageRoot, const QString &name) : mStorage(storageRoot, name, Sink::Storage::ReadWrite)
 {
 }
@@ -101,7 +70,7 @@ void MessageQueue::dequeue(const std::function<void(void *ptr, int size, std::fu
         return KAsync::start<void>([&value, resultHandler](KAsync::Future<void> &future) {
             resultHandler(const_cast<void *>(static_cast<const void *>(value.data())), value.size(), [&future](bool success) { future.setFinished(); });
         });
-    }).then<void>([]() {}, [errorHandler](int error, const QString &errorString) { errorHandler(Error("messagequeue", error, errorString.toLatin1())); }).exec();
+    }).onError([errorHandler](const KAsync::Error &error) { errorHandler(Error("messagequeue", error.errorCode, error.errorMessage.toLatin1())); }).exec();
 }
 
 KAsync::Job<void> MessageQueue::dequeueBatch(int maxBatchSize, const std::function<KAsync::Job<void>(const QByteArray &)> &resultHandler)
@@ -135,8 +104,8 @@ KAsync::Job<void> MessageQueue::dequeueBatch(int maxBatchSize, const std::functi
                 });
 
         // Trace() << "Waiting on " << waitCondition.size() << " results";
-        ::waitForCompletion(waitCondition)
-            .then<void>([this, resultCount, &future]() {
+        KAsync::waitForCompletion(waitCondition)
+            .syncThen<void>([this, resultCount, &future]() {
                 processRemovals();
                 if (*resultCount == 0) {
                     future.setError(static_cast<int>(ErrorCodes::NoMessageFound), "No message found");
diff --git a/common/pipeline.cpp b/common/pipeline.cpp
index 1d45340..ce864f7 100644
--- a/common/pipeline.cpp
+++ b/common/pipeline.cpp
@@ -233,7 +233,7 @@ KAsync::Job<qint64> Pipeline::newEntity(void const *command, size_t size)
 
     d->storeNewRevision(newRevision, fbb, bufferType, key);
 
-    return KAsync::start<qint64>([newRevision]() { return newRevision; });
+    return KAsync::value(newRevision);
 }
 
 KAsync::Job<qint64> Pipeline::modifiedEntity(void const *command, size_t size)
@@ -346,7 +346,7 @@ KAsync::Job<qint64> Pipeline::modifiedEntity(void const *command, size_t size)
     adaptorFactory->createBuffer(newAdaptor, fbb, metadataFbb.GetBufferPointer(), metadataFbb.GetSize());
 
     d->storeNewRevision(newRevision, fbb, bufferType, key);
-    return KAsync::start<qint64>([newRevision]() { return newRevision; });
+    return KAsync::value(newRevision);
 }
 
 KAsync::Job<qint64> Pipeline::deletedEntity(void const *command, size_t size)
@@ -433,7 +433,7 @@ KAsync::Job<qint64> Pipeline::deletedEntity(void const *command, size_t size)
         processor->deletedEntity(key, newRevision, *current, d->transaction);
     }
 
-    return KAsync::start<qint64>([newRevision]() { return newRevision; });
+    return KAsync::value(newRevision);
 }
 
 void Pipeline::cleanupRevision(qint64 revision)
diff --git a/common/queryrunner.cpp b/common/queryrunner.cpp
index 2e2e96d..052db39 100644
--- a/common/queryrunner.cpp
+++ b/common/queryrunner.cpp
@@ -86,7 +86,7 @@ QueryRunner<DomainType>::QueryRunner(const Sink::Query &query, const Sink::Resou
                 const auto  newRevisionAndReplayedEntities = worker.executeInitialQuery(query, parent, *resultProvider, mOffset[parentId], mBatchSize);
                 return newRevisionAndReplayedEntities;
             })
-                .template then<void, QPair<qint64, qint64>>([=](const QPair<qint64, qint64> &newRevisionAndReplayedEntities) {
+                .template syncThen<void, QPair<qint64, qint64>>([=](const QPair<qint64, qint64> &newRevisionAndReplayedEntities) {
                     mOffset[parentId] += newRevisionAndReplayedEntities.second;
                     // Only send the revision replayed information if we're connected to the resource, there's no need to start the resource otherwise.
                     if (query.liveQuery) {
@@ -110,7 +110,7 @@ QueryRunner<DomainType>::QueryRunner(const Sink::Query &query, const Sink::Resou
                        const auto newRevisionAndReplayedEntities = worker.executeIncrementalQuery(query, *resultProvider);
                        return newRevisionAndReplayedEntities;
                    })
-                .template then<void, QPair<qint64, qint64> >([query, this, resultProvider](const QPair<qint64, qint64> &newRevisionAndReplayedEntities) {
+                .template syncThen<void, QPair<qint64, qint64> >([query, this, resultProvider](const QPair<qint64, qint64> &newRevisionAndReplayedEntities) {
                     // Only send the revision replayed information if we're connected to the resource, there's no need to start the resource otherwise.
                     mResourceAccess->sendRevisionReplayedCommand(newRevisionAndReplayedEntities.first);
                     resultProvider->setRevision(newRevisionAndReplayedEntities.first);
diff --git a/common/resourceaccess.cpp b/common/resourceaccess.cpp
index 7b4d839..364616c 100644
--- a/common/resourceaccess.cpp
+++ b/common/resourceaccess.cpp
@@ -159,67 +159,65 @@ KAsync::Job<void> ResourceAccess::Private::tryToConnect()
     // We may have a socket from the last connection leftover
     socket.reset();
     auto counter = QSharedPointer<int>::create(0);
-    return KAsync::dowhile([this]() -> bool { return !socket; },
-        [this, counter](KAsync::Future<void> &future) {
+    return KAsync::dowhile(
+        [this, counter]() {
             SinkTrace() << "Loop";
-            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) {
-                        static int waitTime = 10;
-                        static int timeout = 500;
-                        static int maxRetries = timeout / waitTime;
-                        if (*counter > maxRetries) {
-                            SinkTrace() << "Giving up";
-                            future.setError(-1, "Failed to connect to socket");
+            return connectToServer(resourceInstanceIdentifier)
+                .then<KAsync::ControlFlowFlag, QSharedPointer<QLocalSocket>>(
+                    [this, counter](const KAsync::Error &error, const QSharedPointer<QLocalSocket> &s) {
+                        if (error) {
+                            static int waitTime = 10;
+                            static int timeout = 500;
+                            static int maxRetries = timeout / waitTime;
+                            if (*counter > maxRetries) {
+                                SinkTrace() << "Giving up";
+                                return KAsync::error<KAsync::ControlFlowFlag>("Failed to connect to socket");
+                            } else {
+                                *counter = *counter + 1;
+                                return KAsync::wait(waitTime).then(KAsync::value(KAsync::Continue));
+                            }
                         } else {
-                            KAsync::wait(waitTime).then<void>([&future]() { future.setFinished(); }).exec();
+                            Q_ASSERT(s);
+                            socket = s;
+                            return KAsync::value(KAsync::Break);
                         }
-                        *counter = *counter + 1;
-                    })
-                .exec();
+                    });
         });
 }
 
 KAsync::Job<void> ResourceAccess::Private::initializeSocket()
 {
-    return KAsync::start<void>([this](KAsync::Future<void> &future) {
+    return KAsync::start<void>([this] {
         SinkTrace() << "Trying to connect";
-        connectToServer(resourceInstanceIdentifier)
+        return connectToServer(resourceInstanceIdentifier)
             .then<void, QSharedPointer<QLocalSocket>>(
-                [this, &future](const QSharedPointer<QLocalSocket> &s) {
-                    SinkTrace() << "Connected to resource, without having to start it.";
-                    Q_ASSERT(s);
-                    socket = s;
-                    future.setFinished();
-                },
-                [this, &future](int errorCode, const QString &errorString) {
-                    SinkTrace() << "Failed to connect, starting resource";
-                    // We failed to connect, so let's start the resource
-                    QStringList args;
-                    if (Sink::Test::testModeEnabled()) {
-                        args << "--test";
-                    }
-                    args << resourceInstanceIdentifier << resourceName;
-                    qint64 pid = 0;
-                    if (QProcess::startDetached("sink_synchronizer", args, QDir::homePath(), &pid)) {
-                        SinkTrace() << "Started resource " << pid;
-                        tryToConnect()
-                            .then<void>([&future]() { future.setFinished(); },
-                                [this, &future](int errorCode, const QString &errorString) {
-                                    SinkWarning() << "Failed to connect to started resource";
-                                    future.setError(errorCode, errorString);
-                                })
-                            .exec();
+                [this](const KAsync::Error &error, const QSharedPointer<QLocalSocket> &s) {
+                    if (error) {
+                        SinkTrace() << "Failed to connect, starting resource";
+                        // We failed to connect, so let's start the resource
+                        QStringList args;
+                        if (Sink::Test::testModeEnabled()) {
+                            args << "--test";
+                        }
+                        args << resourceInstanceIdentifier << resourceName;
+                        qint64 pid = 0;
+                        if (QProcess::startDetached("sink_synchronizer", args, QDir::homePath(), &pid)) {
+                            SinkTrace() << "Started resource " << pid;
+                            return tryToConnect()
+                                .onError([this](const KAsync::Error &error) {
+                                        SinkWarning() << "Failed to connect to started resource";
+                                });
+                        } else {
+                            SinkWarning() << "Failed to start resource";
+                        }
+                        return KAsync::null();
                     } else {
-                        SinkWarning() << "Failed to start resource";
+                        SinkTrace() << "Connected to resource, without having to start it.";
+                        Q_ASSERT(s);
+                        socket = s;
+                        return KAsync::null();
                     }
-                })
-            .exec();
+                });
     });
 }
 
@@ -383,17 +381,18 @@ void ResourceAccess::open()
     d->openingSocket = true;
     d->initializeSocket()
         .then<void>(
-            [this, time]() {
-                SinkTrace() << "Socket is initialized." << Log::TraceTime(time->elapsed());
+            [this, time](const KAsync::Error &error) {
                 d->openingSocket = false;
-                QObject::connect(d->socket.data(), &QLocalSocket::disconnected, this, &ResourceAccess::disconnected);
-                QObject::connect(d->socket.data(), SIGNAL(error(QLocalSocket::LocalSocketError)), this, SLOT(connectionError(QLocalSocket::LocalSocketError)));
-                QObject::connect(d->socket.data(), &QIODevice::readyRead, this, &ResourceAccess::readResourceMessage);
-                connected();
-            },
-            [this](int error, const QString &errorString) {
-                d->openingSocket = false;
-                SinkWarning() << "Failed to initialize socket " << errorString;
+                if (error) {
+                    SinkWarning() << "Failed to initialize socket " << error.errorMessage;
+                } else {
+                    SinkTrace() << "Socket is initialized." << Log::TraceTime(time->elapsed());
+                    QObject::connect(d->socket.data(), &QLocalSocket::disconnected, this, &ResourceAccess::disconnected);
+                    QObject::connect(d->socket.data(), SIGNAL(error(QLocalSocket::LocalSocketError)), this, SLOT(connectionError(QLocalSocket::LocalSocketError)));
+                    QObject::connect(d->socket.data(), &QIODevice::readyRead, this, &ResourceAccess::readResourceMessage);
+                    connected();
+                }
+                return KAsync::null();
             })
         .exec();
 }
diff --git a/common/resourcecontrol.cpp b/common/resourcecontrol.cpp
index 7d092a4..f509318 100644
--- a/common/resourcecontrol.cpp
+++ b/common/resourcecontrol.cpp
@@ -41,22 +41,22 @@ KAsync::Job<void> ResourceControl::shutdown(const QByteArray &identifier)
     time->start();
     return ResourceAccess::connectToServer(identifier)
         .then<void, QSharedPointer<QLocalSocket>>(
-            [identifier, time](QSharedPointer<QLocalSocket> socket, KAsync::Future<void> &future) {
+            [identifier, time](const KAsync::Error &error, QSharedPointer<QLocalSocket> socket) {
+                if (error) {
+                    SinkTrace() << "Resource is already closed.";
+                    // Resource isn't started, nothing to shutdown
+                    return KAsync::null();
+                }
                 // We can't currently reuse the socket
                 socket->close();
                 auto resourceAccess = ResourceAccessFactory::instance().getAccess(identifier, ResourceConfig::getResourceType(identifier));
                 resourceAccess->open();
-                resourceAccess->sendCommand(Sink::Commands::ShutdownCommand)
-                    .then<void>([&future, resourceAccess, time]() {
+                return resourceAccess->sendCommand(Sink::Commands::ShutdownCommand)
+                    .addToContext(resourceAccess)
+                    .syncThen<void>([resourceAccess, time]() {
                         resourceAccess->close();
                         SinkTrace() << "Shutdown complete." << Log::TraceTime(time->elapsed());
-                        future.setFinished();
-                    })
-                    .exec();
-            },
-            [](int, const QString &) {
-                SinkTrace() << "Resource is already closed.";
-                // Resource isn't started, nothing to shutdown
+                    });
             });
 }
 
@@ -67,18 +67,19 @@ KAsync::Job<void> ResourceControl::start(const QByteArray &identifier)
     time->start();
     auto resourceAccess = ResourceAccessFactory::instance().getAccess(identifier, ResourceConfig::getResourceType(identifier));
     resourceAccess->open();
-    return resourceAccess->sendCommand(Sink::Commands::PingCommand).then<void>([resourceAccess, time]() { SinkTrace() << "Start complete." << Log::TraceTime(time->elapsed()); });
+    return resourceAccess->sendCommand(Sink::Commands::PingCommand).addToContext(resourceAccess).syncThen<void>([time]() { SinkTrace() << "Start complete." << Log::TraceTime(time->elapsed()); });
 }
 
 KAsync::Job<void> ResourceControl::flushMessageQueue(const QByteArrayList &resourceIdentifier)
 {
     SinkTrace() << "flushMessageQueue" << resourceIdentifier;
-    return KAsync::iterate(resourceIdentifier)
-        .template each<void, QByteArray>([](const QByteArray &resource, KAsync::Future<void> &future) {
+    return KAsync::value(resourceIdentifier)
+        .template each([](const QByteArray &resource) {
             SinkTrace() << "Flushing message queue " << resource;
             auto resourceAccess = ResourceAccessFactory::instance().getAccess(resource, ResourceConfig::getResourceType(resource));
             resourceAccess->open();
-            resourceAccess->synchronizeResource(false, true).then<void>([&future, resourceAccess]() { future.setFinished(); }).exec();
+            return resourceAccess->synchronizeResource(false, true)
+                .addToContext(resourceAccess);
         });
 }
 
diff --git a/common/resourcefacade.cpp b/common/resourcefacade.cpp
index bf4239d..1c56fe5 100644
--- a/common/resourcefacade.cpp
+++ b/common/resourcefacade.cpp
@@ -167,7 +167,7 @@ template <typename DomainType>
 KAsync::Job<void> LocalStorageFacade<DomainType>::create(const DomainType &domainObject)
 {
     auto configStoreIdentifier = mIdentifier;
-    return KAsync::start<void>([domainObject, configStoreIdentifier]() {
+    return KAsync::syncStart<void>([domainObject, configStoreIdentifier]() {
         const QByteArray type = domainObject.getProperty("type").toByteArray();
         const QByteArray providedIdentifier = domainObject.identifier().isEmpty() ? domainObject.getProperty("identifier").toByteArray() : domainObject.identifier();
         const QByteArray identifier = providedIdentifier.isEmpty() ? ResourceConfig::newIdentifier(type) : providedIdentifier;
@@ -192,7 +192,7 @@ template <typename DomainType>
 KAsync::Job<void> LocalStorageFacade<DomainType>::modify(const DomainType &domainObject)
 {
     auto configStoreIdentifier = mIdentifier;
-    return KAsync::start<void>([domainObject, configStoreIdentifier]() {
+    return KAsync::syncStart<void>([domainObject, configStoreIdentifier]() {
         const QByteArray identifier = domainObject.identifier();
         if (identifier.isEmpty()) {
             SinkWarning() << "We need an \"identifier\" property to identify the entity to configure.";
@@ -220,7 +220,7 @@ template <typename DomainType>
 KAsync::Job<void> LocalStorageFacade<DomainType>::remove(const DomainType &domainObject)
 {
     auto configStoreIdentifier = mIdentifier;
-    return KAsync::start<void>([domainObject, configStoreIdentifier]() {
+    return KAsync::syncStart<void>([domainObject, configStoreIdentifier]() {
         const QByteArray identifier = domainObject.identifier();
         if (identifier.isEmpty()) {
             SinkWarning() << "We need an \"identifier\" property to identify the entity to configure";
diff --git a/common/sourcewriteback.cpp b/common/sourcewriteback.cpp
index fe996cb..702d8e3 100644
--- a/common/sourcewriteback.cpp
+++ b/common/sourcewriteback.cpp
@@ -106,7 +106,7 @@ KAsync::Job<void> SourceWriteBack::replay(const QByteArray &type, const QByteArr
         job = replay(mail, operation, oldRemoteId, modifiedProperties);
     }
 
-    return job.then<void, QByteArray>([this, operation, type, uid, oldRemoteId](const QByteArray &remoteId) {
+    return job.syncThen<void, QByteArray>([this, operation, type, uid, oldRemoteId](const QByteArray &remoteId) {
         if (operation == Sink::Operation_Creation) {
             SinkTrace() << "Replayed creation with remote id: " << remoteId;
             if (remoteId.isEmpty()) {
@@ -127,13 +127,11 @@ KAsync::Job<void> SourceWriteBack::replay(const QByteArray &type, const QByteArr
         } else {
             SinkError() << "Unkown operation" << operation;
         }
-
-        mSyncStore.clear();
-        mEntityStore.clear();
-        mTransaction.abort();
-        mSyncTransaction.commit();
-    }, [this](int errorCode, const QString &errorMessage) {
-        SinkWarning() << "Failed to replay change: " << errorMessage;
+    })
+    .syncThen<void>([this](const KAsync::Error &error) {
+        if (error) {
+            SinkWarning() << "Failed to replay change: " << error.errorMessage;
+        }
         mSyncStore.clear();
         mEntityStore.clear();
         mTransaction.abort();
diff --git a/common/store.cpp b/common/store.cpp
index 07f41f8..c01d220 100644
--- a/common/store.cpp
+++ b/common/store.cpp
@@ -39,6 +39,8 @@
 SINK_DEBUG_AREA("store")
 
 Q_DECLARE_METATYPE(QSharedPointer<Sink::ResultEmitter<Sink::ApplicationDomain::SinkResource::Ptr>>)
+Q_DECLARE_METATYPE(QSharedPointer<Sink::ResourceAccessInterface>);
+Q_DECLARE_METATYPE(std::shared_ptr<void>);
 
 namespace Sink {
 
@@ -169,12 +171,10 @@ QSharedPointer<QAbstractItemModel> Store::loadModel(Query query)
         result.first.exec();
     }
 
-    KAsync::iterate(resources.keys())
-        .template each<void, QByteArray>([query, aggregatingEmitter, resources](const QByteArray &resourceInstanceIdentifier, KAsync::Future<void> &future) {
+    KAsync::value(resources.keys())
+        .template each([query, aggregatingEmitter, resources](const QByteArray &resourceInstanceIdentifier) {
             const auto resourceType = resources.value(resourceInstanceIdentifier);
-            queryResource<DomainType>(resourceType, resourceInstanceIdentifier, query, aggregatingEmitter).template then<void>([&future]() {
-                future.setFinished();
-            }).exec();
+            return queryResource<DomainType>(resourceType, resourceInstanceIdentifier, query, aggregatingEmitter);
         })
         .exec();
     model->fetchMore(QModelIndex());
@@ -201,7 +201,7 @@ KAsync::Job<void> Store::create(const DomainType &domainObject)
 {
     // Potentially move to separate thread as well
     auto facade = getFacade<DomainType>(domainObject.resourceInstanceIdentifier());
-    return facade->create(domainObject).template then<void>([facade]() {}, [](int errorCode, const QString &error) { SinkWarning() << "Failed to create"; });
+    return facade->create(domainObject).addToContext(std::shared_ptr<void>(facade)).onError([](const KAsync::Error &error) { SinkWarning() << "Failed to create"; });
 }
 
 template <class DomainType>
@@ -209,7 +209,7 @@ KAsync::Job<void> Store::modify(const DomainType &domainObject)
 {
     // Potentially move to separate thread as well
     auto facade = getFacade<DomainType>(domainObject.resourceInstanceIdentifier());
-    return facade->modify(domainObject).template then<void>([facade]() {}, [](int errorCode, const QString &error) { SinkWarning() << "Failed to modify"; });
+    return facade->modify(domainObject).addToContext(std::shared_ptr<void>(facade)).onError([](const KAsync::Error &error) { SinkWarning() << "Failed to modify"; });
 }
 
 template <class DomainType>
@@ -217,7 +217,7 @@ KAsync::Job<void> Store::remove(const DomainType &domainObject)
 {
     // Potentially move to separate thread as well
     auto facade = getFacade<DomainType>(domainObject.resourceInstanceIdentifier());
-    return facade->remove(domainObject).template then<void>([facade]() {}, [](int errorCode, const QString &error) { SinkWarning() << "Failed to remove"; });
+    return facade->remove(domainObject).addToContext(std::shared_ptr<void>(facade)).onError([](const KAsync::Error &error) { SinkWarning() << "Failed to remove"; });
 }
 
 KAsync::Job<void> Store::removeDataFromDisk(const QByteArray &identifier)
@@ -231,6 +231,7 @@ KAsync::Job<void> Store::removeDataFromDisk(const QByteArray &identifier)
     auto resourceAccess = ResourceAccessFactory::instance().getAccess(identifier, ResourceConfig::getResourceType(identifier));
     resourceAccess->open();
     return resourceAccess->sendCommand(Sink::Commands::RemoveFromDiskCommand)
+        .addToContext(resourceAccess)
         .then<void>([resourceAccess](KAsync::Future<void> &future) {
             if (resourceAccess->isReady()) {
                 //Wait for the resource shutdown
@@ -243,8 +244,8 @@ KAsync::Job<void> Store::removeDataFromDisk(const QByteArray &identifier)
                 future.setFinished();
             }
         })
-        .then<void>([resourceAccess, time]() {
-            SinkTrace() << "Remove from disk complete." << Log::TraceTime(time->elapsed()); 
+        .syncThen<void>([time]() {
+            SinkTrace() << "Remove from disk complete." << Log::TraceTime(time->elapsed());
         });
 }
 
@@ -253,41 +254,38 @@ KAsync::Job<void> Store::synchronize(const Sink::Query &query)
     SinkTrace() << "synchronize" << query.resources;
     auto resources = getResources(query.resources, query.accounts).keys();
     //FIXME only necessary because each doesn't propagate errors
-    auto error = new bool;
-    return KAsync::iterate(resources)
-        .template each<void, QByteArray>([query, error](const QByteArray &resource, KAsync::Future<void> &future) {
+    auto errorFlag = new bool;
+    return KAsync::value(resources)
+        .template each([query, errorFlag](const QByteArray &resource) {
             SinkTrace() << "Synchronizing " << resource;
             auto resourceAccess = ResourceAccessFactory::instance().getAccess(resource, ResourceConfig::getResourceType(resource));
             resourceAccess->open();
-            resourceAccess->synchronizeResource(true, false).then<void>([resourceAccess, &future]() {SinkTrace() << "synced."; future.setFinished(); },
-                    [&future, error](int errorCode, QString msg) { *error = true; SinkWarning() << "Error during sync."; future.setError(errorCode, msg); }).exec();
-        }).then<void>([error](KAsync::Future<void> &future) {
-            if (*error) {
-                future.setError(1, "Error during sync.");
-            } else {
-                future.setFinished();
+            return resourceAccess->synchronizeResource(true, false)
+                .addToContext(resourceAccess)
+                .then<void>([errorFlag](const KAsync::Error &error) {
+                        if (error) {
+                            *errorFlag = true;
+                            SinkWarning() << "Error during sync.";
+                            return KAsync::error<void>(error);
+                        }
+                        SinkTrace() << "synced.";
+                        return KAsync::null<void>();
+                    });
+        })
+        .then<void>([errorFlag]() {
+            if (*errorFlag) {
+                return KAsync::error<void>("Error during sync.");
             }
-            delete error;
+            delete errorFlag;
+            return KAsync::null<void>();
         });
 }
 
 template <class DomainType>
 KAsync::Job<DomainType> Store::fetchOne(const Sink::Query &query)
 {
-    return KAsync::start<DomainType>([query](KAsync::Future<DomainType> &future) {
-        // FIXME We could do this more elegantly if composed jobs would have the correct type (In that case we'd simply return the value from then continuation, and could avoid the
-        // outer job entirely)
-        fetch<DomainType>(query, 1)
-            .template then<void, QList<typename DomainType::Ptr>>(
-                [&future](const QList<typename DomainType::Ptr> &list) {
-                    future.setValue(*list.first());
-                    future.setFinished();
-                },
-                [&future](int errorCode, const QString &errorMessage) {
-                    future.setError(errorCode, errorMessage);
-                    future.setFinished();
-                })
-            .exec();
+    return fetch<DomainType>(query, 1).template then<DomainType, QList<typename DomainType::Ptr>>([](const QList<typename DomainType::Ptr> &list) {
+        return KAsync::value(*list.first());
     });
 }
 
diff --git a/common/synchronizer.cpp b/common/synchronizer.cpp
index 2d4fb8d..15a06e7 100644
--- a/common/synchronizer.cpp
+++ b/common/synchronizer.cpp
@@ -244,7 +244,7 @@ KAsync::Job<void> Synchronizer::synchronize()
     SinkTrace() << "Synchronizing";
     mSyncInProgress = true;
     mMessageQueue->startTransaction();
-    return synchronizeWithSource().then<void>([this]() {
+    return synchronizeWithSource().syncThen<void>([this]() {
         mSyncStore.clear();
         mEntityStore.clear();
         mMessageQueue->commit();
diff --git a/common/test.cpp b/common/test.cpp
index 5b4c899..1a8e11d 100644
--- a/common/test.cpp
+++ b/common/test.cpp
@@ -156,7 +156,7 @@ public:
             }
             resultProvider->initialResultSetComplete(parent);
         });
-        auto job = KAsync::start<void>([query, resultProvider]() {});
+        auto job = KAsync::syncStart<void>([query, resultProvider]() {});
         return qMakePair(job, emitter);
     }