benlau/asyncfuture

Use QFuture like a Promise object

asyncfuture's Issues

Chaining Futures and Chained Cancellation

When setting up a chain of futures a common use case is to have a single cancellation token associated with the chain such that if it is ever canceled then the chain will short circuit and not execute any more of the chain by unsubscribing the next in sequence.
Here's a small example:

  auto testToken = deferred<void>();

  auto firstHandler = [] { std::cout << "firstHandler called\n"; };

  auto secondHandler = [testToken]() mutable {
    std::cout << "secondHandler called\n";
    testToken.cancel();
  };

  auto thirdHandler = [] { std::cout << "Oops, did not want thirdHandler to be called\n"; };

  auto starter = deferred<void>();

  QFuture<void> f1 = starter.future();
  QFuture<void> f2 = observe(f1).subscribe(firstHandler).future();
  QFuture<void> f3 = observe(f2).subscribe(secondHandler).future();
  QFuture<void> f4 = observe(f3).subscribe(thirdHandler).future();

  testToken.complete(f4);

  testToken.subscribe([] { std::cout << "chain completed\n"; },
                      [] { std::cout << "chain canceled\n"; });
  starter.complete();

Output:

firstHandler called
secondHandler called
chain canceled
Oops, did not want thirdHandler to be called

The code is behaving exactly as it should since f3 completes which causes f4 with the thirdHandler to be called - only the testToken was canceled, and f4 was not.

What I would like to do in a clean way is cancel the next future in the chain from within the handler, and there are many ways to do that but none of the methods I have come up with are very ergonomic.

One solution is to just pass a value indicating whether to continue or not, for example:

  auto testToken = deferred<bool>();

  auto firstHandler = [](bool cont) {
    if (!cont)
      return false;
    std::cout << "firstHandler called\n";
    return true;
  };
  auto secondHandler = [](bool cont) mutable {
    if (!cont)
      return false;
    std::cout << "secondHandler called\n";
    // cancel chain by returning false here
    return false;
  };
  auto thirdHandler = [](bool cont) {
    if (!cont)
      return false;
    std::cout << "Oops, did not want thirdHandler to be called\n";
    return true;
  };

  auto starter = deferred<bool>();

  QFuture<bool> f1 = starter.future();
  QFuture<bool> f2 = observe(f1).subscribe(firstHandler).future();
  QFuture<bool> f3 = observe(f2).subscribe(secondHandler).future();
  QFuture<bool> f4 = observe(f3).subscribe(thirdHandler).future();

  testToken.complete(f4);

  testToken.subscribe(
      [](bool cont) {
        if (cont) {
          std::cout << "chain completed\n";
        } else {
          std::cout << "chain canceled by passing value\n";
        }
      },
      [] { std::cout << "chain canceled\n"; });
  starter.complete(true);

Output:

firstHandler called
secondHandler called
chain canceled by passing value

But this solution leads to every handler always being called, and also pollutes the arguments with an unrelated bool.

There are other solutions such as creating a deferred for each place the chain may be canceled and then chaining the calls through those deferred's, or establishing the chain from within the handlers themselves (e.g. firstHandler invokes the call to secondHandler). But these solutions are very messy to work with and make the handlers less composable as functions.

Maybe I have overlooked something in the API and there is a simple way to achieve what I am looking for, any help would be appreciated.

In C# using Tasks it is done by passing a cancellation token along with the handler function when subscribing, e.g. it would look something like this:

CancellationToken token;
QFuture<void> f1 = GetSomeFuture();
QFuture<void> f2 = observe(f1).subscribe(firstHandler, token).future();

Then if anything calls token.cancel() it would actually unsubscribe any functions that were subscribed with the token. Also long running functions in the chain can decide to regularly check the token to see if token.isCanceled() to stop mid function call.

waitForFinished() waits forever for a nested AsyncFuture::observer

Hello,

I've got one specific use case that could not resolve. Here is the test:

#include <gmock/gmock.h>
#include <gtest/gtest.h>
#include <asyncfuture/asyncfuture.h>

namespace {
QFuture<void> createCompletedFuture() {
    return {};
}

QFuture<void> createNestedFuture() {
    auto deferred = AsyncFuture::Deferred<void>();
    AsyncFuture::observe(createCompletedFuture()).subscribe([=]() mutable {
        deferred.complete();
    });
    return deferred.future();
}
}

TEST(NestedAsyncFutureTest, Testing)
{
    auto future = createNestedFuture();
    future.waitForFinished();
}

This test never finishes.
I cannot use complete(future) API because inside of subscribe() lambda I also need to emit some signals in the real code.
Is there a workaround for this? or is this a bug?

Thank you!

Reporting "started" state from DeferredFuture

In my use case, a function compute() that is called by UI classes returns a QFuture. The UI classes use a QFutureWatcher to update their state. They also depend on the started() signal, showing a progress bar when a computation is going on.
The calculation performed inside compute() consists basically of two parts, the first of which may or may not be cached. It looks about like this:

QFuture<double> compute(QVector<QPointF> coords) const
{
	QFuture<System> as = getSystem();
	auto f = [=](System&& system)
	{
		return QtConcurrent::mapped(std::move(coords), Worker(system));
	};

	auto rv = AsyncFuture::observe(as).subscribe(f);
	return rv.future();
}

So I needed the future being returned to trigger the started() signal at some point. The following patch to asyncfuture.h worked for me:

     void complete(QFuture<T> future) {
+     QFutureInterface<T>::reportStarted();
+
        incRefCount();
        auto onFinished = [=]() {
            this->completeByFinishedFuture<T>(future);
            this->decRefCount();
        };
        ...

This feels a little bit like a hack and I'm not sure when exactly the started() signal should be emitted for the most generally useful behaviour, but as the QtConcurrent implementations of QFuture always trigger started, I think DeferredFuture should also do this at some point for consistency.

QFuture (aka a DeferredFuture) return from asyncfuture isn't threadsafe and causes crashes

DeferredFuture inherits QObject, but uses a sharedpointer to do reference count when copying a QFuture around. If the QFuture is copied to another thread and the reference count goes to zero. The DeferredFuture is delete in a seperate thread that is not equal to the QObject's thread. This is bad because, QObject needs to be deleted in the thread it lives in. When the QObject is signal from it's thread, but is delete from a seperate thread, it crashes the program. This happens intermittently, but using Qt with address sanitation turned on, will easily show the issue.

When DeferredFuture's need to get delete, they just need to use a QMetaObject::invokeMethod to do the deletion on their thread.

I'll upload a Pull request and a testcase soon to demonstrate the issue.

QT Creator can't find TestRunner QtQuickTest Automator, whare are they from?

what dependencies that I have to install ?

Combiner should honor child progress

The following testcase should pass:

    QVector<int> ints(100);
    std::iota(ints.begin(), ints.end(), ints.size());
    std::function<int (int)> func = [](int x)->int {
        QThread::msleep(100);
        return x * x;
    };
    QFuture<int> mappedFuture = QtConcurrent::mapped(ints, func);
    QFuture<int> runFuture = QtConcurrent::run([]() {
        QThread::msleep(100);
        return 10;
    });

    AsyncFuture::Combinator combine;
    combine << mappedFuture << runFuture;

    auto combineFuture = combine.future();

    QCOMPARE(combineFuture.progressMinimum(), 0);
    QCOMPARE(combineFuture.progressMaximum(), ints.size() + 1);

    int progress = 0;
    AsyncFuture::observe(combineFuture).onProgress([&progress, combineFuture](){
        QVERIFY(progress <= combineFuture.progressValue());
        progress = combineFuture.progressValue();
    });

    await(combineFuture);

    QCOMPARE(combineFuture.progressMaximum(), combineFuture.progressValue());

Creating a completed Deferred is very heavy

Chaining asyncfuture is useful to chain QFuture's together. Sometimes at the begin of the chain, it's useful to have completed QFuture that come from a fixed value. To achieve this with asyncfuture, a programmer needs to use deferred() and then complete it with deferred::complete(). Although this works, deferred creates a Deferred object and a that holds a DeferredFuture. A DeferredFuture is a fairly heavy weight object since inherits QObject and hold a mutex. A DeferrerFuture is usually stored as a shared object and require reference counting.

To reduce this overhead for the specific job having a QFuture that's automatically complete, I propose having a QFuture<void> finished() and QFuture<T> finished<T>() functions.

They would be implement as the following:

inline QFuture<void> finished() {
   QFutureInterface<void> fi;
   fi.reportFinished();
   return QFuture<void>(&fi);
}

template <typename T> QFuture<T> finished(const T &val) {
   QFutureInterface<T> fi;
   fi.reportFinished(&val);
   return QFuture<T>(&fi);
}

Port asyncfuture to use Qt 6.3

Now that Qt 6.3 supports whenAll (https://doc-snapshots.qt.io/qt6-6.3/qtfuture.html#whenAll) we should be able to port asyncfuture's backend to use Qt 6.3 API. That's probably performance benefits since Qt QFuture api doesn't use QObject to combine futures.

Exception Handling

It may be a good idea to add exception handling in a way similar to QtConcurrent.

Something like the following patch may do the trick:

---------------------- 3rdparty/asyncfuture/asyncfuture.h ----------------------
index 9abf14e..a658653 100644
@@ -960,18 +960,23 @@ eval(Functor functor, QFuture<T> future) {
 template <typename DeferredType, typename RetType, typename T, typename Completed, typename Canceled>
 static DeferredFuture<DeferredType>* execute(QFuture<T> future, QObject* contextObject, Completed onCompleted, Canceled onCanceled) {
 
     DeferredFuture<DeferredType>* defer = new DeferredFuture<DeferredType>();
     defer->autoDelete = true;
     watch(future,
           contextObject,
           contextObject,[=]() {
-        Value<RetType> value = eval(onCompleted, future);
-        defer->complete(value);
+		try {
+			Value<RetType> value = eval(onCompleted, future);
+			defer->complete(value);
+		} catch(const QException& e) {
+			defer->reportException(e);
+		} catch(const std::exception&) {
+			defer->reportException(QUnhandledException());
     }, [=]() {
         onCanceled();
         defer->cancel();
     });
 
     if (contextObject) {
         defer->cancel(contextObject, &QObject::destroyed);
     }

feature request: fsanitize=thread results?

As newcomer I would like to see that testing with -fsanitize=thread have been done.
In perfect world of course this should be part of CI, but not sure how much efforts this requires,
because of you need compile Qt from source code with -fsanitize=thread flag.

So may be just mention in readme that such kind of testing was done for OS Z/compiler Y/Qt X and all tests are green?

Allow progress watching is work threads

Currently asyncfuture doesn't support watching progress on work thread because the future watcher is always move to the main thread. If the work thread has an eventloop, this should work. I propose being able to send a context object into Observable::onProgress(QObject* context) to support multi threaded progress watching.

Remove/Update deprecated Qt functions from testcases

With Qt 5.14.1, AsyncFuture testcases produce many warnings about using deprecated qt functions. These should be update so AsyncFuture continues to work in the future.

Combined() should forward cancel() to child futures

If a combine is created, the future() return from it should cancel all the child futures, if it's canceled.

The following testcase should pass:

void BugTests::test_combine_forward_cancel() {

    QAtomicInt runCount = 0;
    const int count = 100;

    QThreadPool::globalInstance()->setMaxThreadCount(2);

    auto createMappedFuture = [&runCount, count]() {
        QVector<int> ints(count);
        std::iota(ints.begin(), ints.end(), ints.size());
        std::function<int (int)> func = [&runCount](int x)->int {
            QThread::msleep(10);
            runCount++;
            return x * x;
        };
        QFuture<int> mappedFuture = QtConcurrent::mapped(ints, func);
        return mappedFuture;
    };

    QList<QFuture<int>> futures;

    for(int i = 0; i < 4; i++) {
        futures.append(createMappedFuture());
    }

    auto c = combine();
    c << futures;

    auto combineFuture = c.future();

    observe(combineFuture).onProgress([&combineFuture](){
        //Cancel after startup
        combineFuture.cancel();
    });

    await(combineFuture);
    QThreadPool::globalInstance()->waitForDone();

    //All the sub futures should be canceled
    for(auto future : futures) {
        QVERIFY(future.isCanceled());
    }

    //Make sure the sub futures didn't run all the way through
    double fullNumberOfRuns = futures.size() * count;
    double ratio = runCount / fullNumberOfRuns;
    QVERIFY(ratio < 0.1);

    QThreadPool::globalInstance()->setMaxThreadCount(QThread::idealThreadCount());
}

How does one error a AsyncFuture::Deferred?

And how does one communicate that error?
Is this what 'cancel' is for? - doesn't sound this way as cancel is the interface of the future not of the promise.

And a related question, having a failed QFuture - how does one "rescue" it? - i.e.: having this chain:
return observe(reading).subscribe(updateCache).future(); - suppose reading fails and suppose there is a way of correcting it:
return observe(reading).subscribe(updateCache, rescue).future();
the above doesn't work.

Allow querying Combinator whether or not any futures have been added

My motivation is the following use case:

I have a result that depends on an arbitrary number of inputs, where arbitrary means: possibly also 0. Each of the inputs can either be immediately available or require delegation to a thread. Now I use AsyncFuture::Combinator to detect when all the results are ready.

My first attempt looked like this:

[...]
auto combiner = AsyncFuture::combine();
QVector<QFuture<Result>> results;
for(auto item: items) {
    auto future = getCachedResultOrCompute(item);
    results << future;
    combiner << future; 
}
return combiner.subscribe([results]() {
   // do something with the results...
}).future();

This approach has one problem though: If items is empty, the callback given to subscribe will never get called. So, it'd be convenient if it were possible to write something like the following wrapper to subscribe on Combinator:

template<typename RetType, typename Func>
inline QFuture<RetType> subscribeToCombinator(Func f, AsyncFuture::Combinator combinator)
{
	auto cancelToken = AsyncFuture::deferred<void>();
	if(combinator.isEmpty()) {
             auto rv = AsyncFuture::deferred<RetType>();
             rv.complete(RetType());
             return rv.future();
	}

	return combinator.subscribe(f).future();
}

This requires a method isEmpty (or whatever you'd like to call it) on Combinator, allowing the client to query whether or not there is already a QFuture on Combinator. As far as I can see, it is easily implented by adding the following code to Combinator:
bool isEmpty() const { return combinedFuture->count == 0; }

how to update widget in main ui from future work?

I have a test , but got some errors.

TaskCenter.h TaskCenter.cpp

#include "asyncfuture.h"
#include "automator.h"

SnapshotWidget snapshotwidget;
QFuture<void> m_subscribeFuture;
QFuture<void> m_doWorkFuture;
int m_count = 0;

TaskCenter::TaskCenter(){}
void TaskCenter::realtime_snapshot_update(int interval) {

    if(m_doWorkFuture.isRunning() || m_doWorkFuture.isStarted()) {
        m_doWorkFuture.cancel();
    }

    QFuture<void> runFuture;
    for (int i = 0 ; i < 50000;i++) {

        runFuture = QtConcurrent::run(&this->snapshotwidget, static_cast<void(SnapshotWidget::*)()>(&SnapshotWidget::dynamic_update_snapshot));

        m_doWorkFuture = runFuture;

        m_subscribeFuture = AsyncFuture::observe(m_doWorkFuture).subscribe([this](){
            qDebug() << "m_count: " << m_count;
            m_count++;
        },
        [this](){
            //
        }).future();

        Automator::wait(interval * 1000);

    }
    return;
}

SnapshotWidget.h SnapshotWidget.cpp

QWidget* anotherwdt = new QWidget();

void SnapshotWidget::dynamic_update_snapshot()
{

    if(!ui->onewdt->layout()){

        FlowLayout *anotherwdtlay = new FlowLayout();
        anotherwdt->setLayout(anotherwdtlay);

        QScrollArea *scrollarea = new QScrollArea();
        scrollarea->setWidget(anotherwdt);

        QVBoxLayout* onewdtlay = new QVBoxLayout();
        ui->onewdt->setLayout(onewdtlay);
        ui->onewdt->layout()->addWidget(scrollarea);

    }

    QLayout* anotherwdtlay = anotherwdt->layout();

    qDeleteAll(anotherwdt->findChildren<QCustomPlot *>(QString()));

    QCustomPlot* plot = new QCustomPlot();

    ... 
    ...
    anotherwdtlay->addWidget(plot);

    return;
}

the error messages are in dynamic_update_snapshot:
QObject::setParent: Cannot set parent, new parent is in a different thread

and the child widgets din't change.

please help me how to modify the codes?

thanks!

Suggestion: Forwarding "cancel" from a DeferredFuture to underlying QtConcurrent future...

It's again about this type of usage pattern:

QFuture<double> compute(QVector<QPointF> coords) const
{
	QFuture<System> sys = getSystem();
	auto f = [=](System&& system)
	{
		return QtConcurrent::mapped(coords, Worker(system));
	};

	auto rv = AsyncFuture::observe(sys).subscribe(f);
	return rv.future();
}

In this case, if client code cancels the future returned by compute(), I need it to actually cancel the QtConcurrent::mapped operation. I may be wrong but I think that this behaviour is generally desirable, as there is nothing else I can see to which the cancel() could relate in this type of situation. The idea is that the QFuture returned by a Observable::future() basically "becomes" the QFuture returned from the lambda function. However, right now cancelling the returned QFuture has no effect at all.

To achieve it, I wrote the following:

template<typename RetType, typename InpType, typename Func>
class FutureFunctor
{
public:
	explicit FutureFunctor(Func wrapped) :
		_wrapped(wrapped),
		_s(std::make_shared<State>())
	{}

	FutureFunctor(const FutureFunctor&) = default;
	FutureFunctor(FutureFunctor&&) = default;

	void cancel() const
	{
		_s->isCancelled = true;
		_s->future.cancel();
	}

	QFuture<RetType> operator()(InpType&& input) const
	{
		Q_ASSERT(!_s->isCalled);
		_s->isCalled = true;
		if(!_s->isCancelled) {
			_s->future = _wrapped(std::forward<InpType>(input));
		}

		return _s->future;
	}

private:
	struct State {
		QFuture<RetType> future;
		bool isCalled = false;
		bool isCancelled = false;
	};

	const Func _wrapped;
	std::shared_ptr<State> _s;
};

template<typename RetType, typename InpType, typename Func>
QFuture<RetType> apply(Func f, QFuture<InpType> input)
{
	auto ff = FutureFunctor<RetType, InpType, Func>(f);
	auto rv = AsyncFuture::observe(input).subscribe(ff);
	rv.subscribe([]() {}, [ff]() { ff.cancel(); });
	return rv.future();
}

Applying it to the use case above, it looks like so:

QFuture<double> compute(QVector<QPointF> coords) const
{
    QFuture<System> sys = getSystem();
    auto f = [=](System&& system)
    {
        return QtConcurrent::mapped(coords, Worker(system));
    };

    return apply<double>(f, sys);
}

As I think that this feature can be generally useful, I thought you might want to include it in the lib, either as an additional layer like in my code, or maybe also by modifying the behaviour of `subscribe in some way.

Add `contextObject` to function `subscribe`

Considering this situation: Display a dialog, and then use AsyncFuture to send a network request, and the result will be displayed on the interface when the network request responds. I used the following logic (pseudo code):

Dialog::request()
{
    auto future = ... ; // request data from network
    observe(future).subscribe([=](QString data){  // onComplete
        ui->label->setText(data);
    },
    [=](){ // onCanceled
        ui->errorLabel->setText("error");
    });
}

But before the network reply, the user may close the dialog. In this case, the ui->label in onComplete has been destroyed, causing a crash.

Is there any plan to add contextObject to function subscribe? So I can modify code like this

Dialog::request()
{
    auto future = ... ; // request data from network
-    observe(future).subscribe([=](QString data){  // onComplete
+    observe(future).subscribe(this, [=](QString data){  // onComplete
        ui->label->setText(data);
    },
    [=](){ // onCanceled
        ui->errorLabel->setText("error");
    });
}

Any plans to upgrade asyncfuture to support Qt6?

As you probably know, this library doesn't work with Qt6. I'm currently working to upgrade one of my projects from Qt5 to Qt6, so I'm wondering if plans exist to upgrade this library to support it? Or if I should port my code to use something else?

Thanks in advance for taking a peek!

Combine should support observables

I was a bit surprised to find that this doesn't work:

Observable<void> o;
auto all = combine() << o;

sure you can just call o.future() but it seems like that should be overloaded in combine.
Could this just be added to the Combinator?

    template <typename T>
    Combinator& operator<<(Observable<T> observable)
    {
        combinedFuture->addFuture(observable.future());
        return *this;
    }

feature request: support move only types

It would be nice to support types with deleted copy constructor/assign operator,
for AsyncFuture::Observable and AsyncFuture::Deferred instantation, for example:

struct Foo {
	explicit Foo(std::string s): s_{s} {}
	Foo(const Foo &) = delete;
	Foo &operator=(const Foo &) = delete;
	Foo(Foo &&o): s_{std::move(o.s_)} { }
	Foo &operator=(Foo &&o) {
		s_ = std::move(o.s_);
		return *this;
	}
...
};

Observable().subscribe(...).future() aka DeferredFuture should forward progress on chained pipeline

For example this pipeline should work:

Run 100 concurrent jobs, then setup another 100 jobs, finally setup another 100 jobs.

The user should be able to get the correct progress by watching the last future in this pipeline. Below is a test-case.

    QVector<int> ints(100);
    std::iota(ints.begin(), ints.end(), ints.size());
    std::function<int (int)> func = [](int x)->int {
        QThread::msleep(100);
        return x * x;
    };
    QFuture<int> mappedFuture = QtConcurrent::mapped(ints, func);

    auto nextFuture = AsyncFuture::observe(mappedFuture).subscribe([ints, func](){
        QFuture<int> mappedFuture2 = QtConcurrent::mapped(ints, func);
        return mappedFuture2;
    }).future();

    bool nextExecuted2 = false;
    auto nextFuture2 = AsyncFuture::observe(nextFuture).subscribe([&nextExecuted2, ints, func](){
        QFuture<int> mappedFuture2 = QtConcurrent::mapped(ints, func);
        nextExecuted2 = true;
        return mappedFuture2;
    }).future();

    int progress = -1;
    AsyncFuture::observe(nextFuture2).onProgress([&progress, nextFuture2](){
        QVERIFY2(progress <= nextFuture2.progressValue(), QString("%1 <= %2").arg(progress).arg(nextFuture2.progressValue()).toLocal8Bit());
        progress = nextFuture2.progressValue();
    });

    await(nextFuture2);

    QCOMPARE(nextFuture2.progressMinimum(), 0);
    QCOMPARE(nextFuture2.progressMaximum(), ints.size() * 3);

    QCOMPARE(nextExecuted2, true);
    QCOMPARE(nextFuture2.progressValue(), ints.size() * 3);

Automator header is missing from repo

Hello,
Could please add Automator class in the repo
Thanks

Context QObject should take const-pointer

As far as I can see Observable<..>::context could and should take a const QObject*, as QObject::connect(..) does as well.

Missing documentation on building testcases

It would be nice if the README.md has documentation on how to build the unit testcases. They currently require downloading qpm and installing the dependancies which aren't documented.

Combined operator<<() isn't thread safe if used outside main thread

An example of this issue is if two futures being added to Combined(). If future1 finished before future2 has been added, this will complete() the Combined(). Even though in this use case, we want the code below to print out "Finished!" once both future1 and future2 are finished. If future1 is finished before future2 is added, Finished() will be prematurely, called.

auto combine = Combined() << future1 << future2;
combine.subscribe([]() { qDebug() << "Finished!"; }

I don't think this can be easily be fixed without api changes to how Combine() is handled. Perhaps it better to advocate using async on the main thread only. The deferred needs to have a proper context object to make this work correrctly. By default deferred is move to the main thread.

Add support for onResultReadyAt

I was trying to load data one by one, like supported in QFuture::resultReadyAt.

Here is an example:

struct ComplexData{
    QString fruit, color;
};
using ComplexDataPtr = QSharedPointer<ComplexData>;

// Simulate network reply content (QNetworkReply)
QFuture<QString> dataLoaderFuture(){
    return QtConcurrent::run([](){
            QList<QString> list{"Apple,red","Banana,yellow","Orange,orange"};
            auto replyContent = list.join(";");
            qDebug()<<"Sending reply:"<<replyContent;
            return replyContent;
    });
}

// The future holds multiple results, one for each fruit.
QFuture<ComplexDataPtr> getFruitsOneByOne_AsyncFuture(){
    qDebug()<<"Getting fruits one by one (using AsyncFuture).";
    auto defer = AsyncFuture::deferred<ComplexDataPtr>();
    auto reply = dataLoaderFuture(); // returns QString "Apple,red;Banana,yellow;Orange,orange"
    auto observe = AsyncFuture::observe(reply).subscribe([=]() {
        qDebug()<<"Creating fruits from reply";
        QList<ComplexDataPtr> results;
        QString replyContent = reply.result();
        QStringList records = replyContent.split(";");
        auto d = defer;
        int index = 0;
        for ( const auto &record : records ){
            results << createFruitFromRecord(record);
            d.reportResult(results.last(),index++); // DOES NOT WORK AS EXPECTED
        }
        //d.complete(results); // THIS IS NOT WHAT I WANT IN THIS TEST CASE
    });
    return defer.future();
}

The Deferred class lacks a suitable method. I naivly tried to add the method reportResult but it did not work.

void reportStarted() {
     deferredFuture->reportStarted();
 }

 // NOT PART OF ORIGINAL LIBRARY, ADDED BECAUSE
 // I COULD NOT GET "resultReadyAt" SIGNAL TO WORK
 // AND THIS WAS MY ATTEMPT TO IMPLEMENT IT (NO SUCCESS)
 void reportResult(const T &result, int index){
     deferredFuture->reportResult(result, index);
 }
protected:
 QSharedPointer<Private::DeferredFuture<T> > deferredFuture;
};

Canceled future will still call finished() even through they're canceled

In my use case, a class maybe called to do work multiple time. It's generally useful to start an processing the data, but if new data is submitted, the class restarts the processing and cancels previous runs. For example:

class TestClass {
    public:
        void doWork() {

            if(m_doWorkFuture.isRunning() || m_doWorkFuture.isStarted()) {
                m_doWorkFuture.cancel();
            }

            int currentCount = m_count;
            auto runFuture = QtConcurrent::run([currentCount](){
                return currentCount + 1;
            });

            m_doWorkFuture = runFuture;

            m_subscribeFuture = observe(runFuture).subscribe([this](){
                m_finishCount++;
            },
            [this](){
                m_cancelCount++;
            }).future();

        }

        void waitToFinish() {
            await(m_subscribeFuture);
        }

        QFuture<void> m_subscribeFuture;
        QFuture<int> m_doWorkFuture;
        int m_count = 0;
        int m_finishCount = 0;
        int m_cancelCount = 0;
    };

    TestClass myTest;
    myTest.doWork();
    myTest.doWork();
    myTest.doWork();
    myTest.waitToFinish();

    QCOMPARE(myTest.m_finishCount, 1);
    QCOMPARE(myTest.m_cancelCount, 2);`

This test currently fails on the master on windows.

Unable to create a AsyncFuture::Deferred<QList<T>>

This is a very handy library - thank you. But:
Consider this function:

QFuture<QList<double>> foo() {
    AsyncFuture::Deferred<QList<double>> _terrain_fetched;
    _terrain_fetched.complete(QList<double>());
    return _terrain_fetched.future();
}

and the compiler error:

In file included from ../foo.h:7,
                 from ../foo.cpp:4:
../AirBoss-MissionPlanner/src/SmartMissionPlanner/AsyncFuture.h: In instantiation of ‘void AsyncFuture::Private::DeferredFuture<T>::reportResult(QList<R>&) [with R = double; T = QList<double>]’:
../AirBoss-MissionPlanner/src/SmartMissionPlanner/AsyncFuture.h:505:9:   required from ‘void AsyncFuture::Private::DeferredFuture<T>::complete(QList<R>&) [with R = double; T = QList<double>]’
../AirBoss-MissionPlanner/src/SmartMissionPlanner/AsyncFuture.h:1280:9:   required from ‘void AsyncFuture::Deferred<T>::complete(T) [with T = QList<double>]’
../foo.cpp:78:46:   required from here
../AirBoss-MissionPlanner/src/SmartMissionPlanner/AsyncFuture.h:661:46: error: no matching function for call to ‘AsyncFuture::Private::DeferredFuture<QList<double> >::reportResult(double&, int&)’
             QFutureInterface<T>::reportResult(value[i], i);
             ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^~~~~~~~~~~~~
In file included from ../../../Qt5.12.5/5.12.5/gcc_64/include/QtCore/qfuture.h:45,
                 from ../../../Qt5.12.5/5.12.5/gcc_64/include/QtCore/QtCore:85,
                 from ../../../Qt5.12.5/5.12.5/gcc_64/include/QtConcurrent/QtConcurrentDepends:3,
                 from ../../../Qt5.12.5/5.12.5/gcc_64/include/QtConcurrent/QtConcurrent:3,
                 from ../foo.cpp:1:
../../../Qt5.12.5/5.12.5/gcc_64/include/QtCore/qfutureinterface.h:189:13: note: candidate: ‘void QFutureInterface<T>::reportResult(const T*, int) [with T = QList<double>]’
 inline void QFutureInterface<T>::reportResult(const T *result, int index)
             ^~~~~~~~~~~~~~~~~~~
../../../Qt5.12.5/5.12.5/gcc_64/include/QtCore/qfutureinterface.h:189:13: note:   no known conversion for argument 1 from ‘double’ to ‘const QList<double>*’
../../../Qt5.12.5/5.12.5/gcc_64/include/QtCore/qfutureinterface.h:209:13: note: candidate: ‘void QFutureInterface<T>::reportResult(const T&, int) [with T = QList<double>]’
 inline void QFutureInterface<T>::reportResult(const T &result, int index)
             ^~~~~~~~~~~~~~~~~~~
../../../Qt5.12.5/5.12.5/gcc_64/include/QtCore/qfutureinterface.h:209:13: note:   no known conversion for argument 1 from ‘double’ to ‘const QList<double>&’
../foo.cpp:43:40: warning: ‘airmap::geography::bounding_box fromQt(const QGeoRectangle&)’ defined but not used [-Wunused-function]
 static airmap::geography::bounding_box fromQt(const QGeoRectangle& bbox) {

for some reason the compiler prefers to match this instead of this.

Cancel not propagated to child future when previous future already finished

If a chain of futures is cancelled with the first future finished by a signal the inner subscribe onCancelled callback on a future of signal is not called, but the outer future is cancelled correctly.

Expected:
The onCancelled of the current running inner future is called

Note:
This test is based on fork of @vpicaver asyncfuture on commit @0da13489746f0412e6bad66bfe0acef54915adb2

class TestClass : public QObject
{
    Q_OBJECT
public:
signals:
    void someTestSignal();
    void someOtherTestSignal();
};

TEST_F(AsyncFutureTest, futureFromTwoSignalsWithFirstSignalCalledShouldBeCancelled)
{
    TestClass testclass;
    auto deferWasCancelled = AsyncFuture::deferred<bool>();
    auto innerDeferWasCancelled = AsyncFuture::deferred<bool>();
    auto future = AsyncFuture::observe(&testclass, &TestClass::someTestSignal)
                      .subscribe(
                          [&innerDeferWasCancelled, &testclass]() {
                              return AsyncFuture::observe(&testclass,
                                                          &TestClass::someOtherTestSignal)
                                  .subscribe(
                                      [&innerDeferWasCancelled]() {
                                          innerDeferWasCancelled.complete(false);
                                      },
                                      [&innerDeferWasCancelled]() {
//following deferred gets never called
                                          innerDeferWasCancelled.complete(true);
                                      });
                          },
                          [&deferWasCancelled]() { deferWasCancelled.complete(true); })
                      .future();
    emit testclass.someTestSignal();
    QTimer::singleShot(40, [&future]() { future.cancel(); });
    waitUntil(future);
    ASSERT_TRUE(future.isCanceled());
    ASSERT_TRUE(future.isFinished());
//following waitUntil gets never finished
    waitUntil(innerDeferWasCancelled.future());
    ASSERT_TRUE(innerDeferWasCancelled.future().result());
    waitUntil(deferWasCancelled.future());
    ASSERT_TRUE(deferWasCancelled.future().result());
}

Crash when using .context() where the contexObject thread is different

If the contextObject is in a different thread than the current execution thread, asyncfuture can cause a crash. This is because QFutureWatcher lives in the current execution thread() but communicating to the contextObject's thread(). In this case, it's possible for the QFutureWatcher to emit bath canceled() and finished() and since both the signals in a queued eventloop, they are handled. By the time they're handled, QFutureWatcher has been destroyed, causing a crash in the other single handler. This isn't a problem if QFutureWatcher and contextObject live in the same thread because QFutureWatcher gets disconnected, when it's destroyed.

'Any' combiner for futures

Currently, this library offers a combiner which resolves once all the futures added to it have resolved.

I don't think there is a 'combiner' which resolves once ANY ONE of the futures added to it have resolved. It would be nice to have this as a feature.

Perhaps 2 versions could be implemented. One would behave like Promise.any() and the other like Promise.race().