ValueStream.hpp

#pragma once

#include <folly/futures/Future.h>

#include <folly/Executor.h>


namespace Showcase
{

  class Showcase;

  template <typename T>
  class ValueStream : public std::enable_shared_from_this<ValueStream<T>>
  {
  public:
    typedef T ValueType;
    ValueStream() {}
    virtual ~ValueStream() {}

    int forEachSync(const std::function<void(const T&)>& f, int N,
                    const std::function<bool(const T&)>& endCond=nullptr);

    ValueStream(const ValueStream&) = delete;
    ValueStream& operator=(const ValueStream&) = delete;

    virtual folly::Future<T> next() = 0;
  };

  template <typename T>
  using ValueStreamPtr = std::shared_ptr<ValueStream<T>>;


  // ---------------------------------------------------------------------
  // ---------------------------------------------------------------------
  // ---------------------------------------------------------------------


  template <typename T>
  int ValueStream<T>::forEachSync(const std::function<void(const T&)>& f, int max,
                                  const std::function<bool(const T&)>& endCond)
  {
    int consumed = 0;

    while(consumed < max) {
      auto future = next().wait();

      ValueType ev = future.value();
      if(endCond && endCond(ev))
        break;

      f(ev);
      ++consumed;
    }
    return consumed;
  }

}