Class CollectProcess

include/process/collect.hpp:119–198

template <typename T>
class CollectProcess : public Process<CollectProcess<T>>
{
public:
  CollectProcess(
      const std::vector<Future<T>>& _futures,
      Promise<std::vector<T>>* _promise)
    : ProcessBase(ID::generate("__collect__")),
      futures(_futures),
      promise(_promise),
      ready(0) {}

  ~CollectProcess() override
  {
    delete promise;
  }

protected:
  void initialize() override
  {
    // Stop this nonsense if nobody cares.
    promise->future().onDiscard(defer(this, &CollectProcess::discarded));

    foreach (const Future<T>& future, futures) {
      future.onAny(defer(this, &CollectProcess::waited, lambda::_1));
      future.onAbandoned(defer(this, &CollectProcess::abandoned));
    }
  }

private:
  void abandoned()
  {
    // There is no use waiting because this future will never complete
    // so terminate this process which will cause `promise` to get
    // deleted and our future to also be abandoned.
    terminate(this);
  }

  void discarded()
  {
    foreach (Future<T> future, futures) {
      future.discard();
    }

    // NOTE: we discard the promise after we set discard on each of
    // the futures so that there is a happens-before relationship that
    // can be assumed by callers.
    promise->discard();

    terminate(this);
  }

  void waited(const Future<T>& future)
  {
    if (future.isFailed()) {
      promise->fail("Collect failed: " + future.failure());
      terminate(this);
    } else if (future.isDiscarded()) {
      promise->fail("Collect failed: future discarded");