GraphGenerators.hpp

#pragma once

#include "Export.hpp"

#include "BackendComponents.hpp"
#include "Events.hpp"

#include <folly/MoveWrapper.h>

#include <queue>

namespace Showcase
{

  class SHOWCASE_API GraphGenerator : public Generator<NodeEvent>
  {
  public:
    GraphGenerator();

    virtual QString protocol() const = 0;

    virtual bool init(const GraphNode& n);

    const GraphNode& node() const;

  private:
    GraphNode m_node;
  };
  typedef std::shared_ptr<GraphGenerator> GraphGeneratorPtr;


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


  class SHOWCASE_API PropertyGenerator : public Generator<PropertyEvent>
  {
  public:
    PropertyGenerator();

    virtual QString protocol() const = 0;

    virtual bool init(const GraphNode& n);

    virtual bool init(const GraphEdge& e);

    virtual bool init(const PropertyEvent&) { return false; }

    bool isEdgeGenerator() const;
    const NodeId& sourceNode() const;
    const NodeId& targetNode() const;
    const EdgeId& edgeId() const;

  private:
    EdgeId m_targetId; 
    bool m_isEdgeGenerator;
  };
  typedef std::shared_ptr<PropertyGenerator> PropertyGeneratorPtr;

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

  template <typename EventGenerator>
  class EmptyEventGenerator : public EventGenerator
  {
  public:
    typedef typename EventGenerator::ValueType Event;

    EmptyEventGenerator() : m_eventsSent(0) {}

    virtual QString protocol() const override { return QString(); }

    virtual folly::Future<typename EventGenerator::ValueType>
    next(folly::Executor* executor = nullptr) override
    {
      int sent = m_eventsSent++;
      if(sent == 0) {
        if(executor) {
          return folly::via(executor, [] { return Event(); });
        } else {
          return folly::makeFuture(Event());
        }
      } else {
        if(sent > 1) {
          m_promise.setValue(Event());
        }
        m_promise = folly::Promise<typename EventGenerator::ValueType>();
        return m_promise.getFuture();
      }
    }

  private:
    int m_eventsSent;
    folly::Promise<typename EventGenerator::ValueType> m_promise;
  };

  typedef EmptyEventGenerator<PropertyGenerator> EmptyPropertyGenerator;
  typedef EmptyEventGenerator<GraphGenerator> EmptyGraphGenerator;

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

  class Notifiable
  {
  public:
    virtual void changeObserved() = 0;
    virtual ~Notifiable() = default;
  };

  template <typename GeneratorType>
  class CachingGenerator : public GeneratorType, public Notifiable
  {
  public:
    static_assert(IsGenerator<GeneratorType>::value,
                  "Need to use Generator or its subclass as a type "
                  "parameter for SynchronousGenerator.");
    typedef typename GeneratorType::ValueType ValueType;

    virtual folly::Future<ValueType> next(folly::Executor* executor = nullptr) override;
    virtual void changeObserved() override;

  protected:
    virtual bool hasEventsInCache() const = 0;
    virtual ValueType nextEvent() = 0;
    virtual void updateCache() = 0;

  private:
    std::mutex m_mutex;

    class Promised
    {
    public:
      Promised(folly::Promise<ValueType>&& p, folly::Executor* exec)
        : promise(std::move(p)), executor(exec) {}
      Promised(Promised&& other) = default;
      folly::Promise<ValueType> promise;
      folly::Executor* executor = nullptr;
    };
    std::queue<Promised> m_promises;

    bool m_pendingCacheUpdate = true;
  };

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

  template <typename GeneratorType>
  folly::Future<typename GeneratorType::ValueType>
  CachingGenerator<GeneratorType>::next(folly::Executor* executor)
  {
    typedef CachingGenerator<GeneratorType> Me;
    auto ptr = this->shared_from_this();

    std::shared_ptr<Me> me = std::dynamic_pointer_cast<Me>(ptr);

    if(!me) {
      if(executor)
        return folly::via(executor, [] { return ValueType(); });
      else
        return folly::makeFuture<ValueType>(ValueType());
    }

    std::weak_ptr<Me> weak = me;

    auto f = [weak] {
      std::shared_ptr<Me> me = weak.lock();
      if(!me)
        return;

      std::unique_lock<std::mutex> lck(me->m_mutex);

      if(!me->hasEventsInCache() && me->m_pendingCacheUpdate) {
        me->m_pendingCacheUpdate = false;
        me->updateCache();
      }

      if(me->hasEventsInCache() && !me->m_promises.empty()) {
        auto promised = std::move(me->m_promises.front());
        me->m_promises.pop();


        auto val = me->nextEvent();
        lck.unlock();
        promised.promise.setValue(val);
      }
    };


    folly::Future<ValueType> fut(folly::exception_wrapper(std::runtime_error("empty")));
    {
      std::lock_guard<std::mutex> g(m_mutex);
      folly::Promise<ValueType> promise;
      fut = promise.getFuture();
      Promised p(std::move(promise), executor);
      m_promises.emplace(std::move(p));
    }

    if(executor) {
      executor->add(f);
    } else {
      f();
    }
    return fut;
  }

  template <typename GeneratorType>
  void CachingGenerator<GeneratorType>::changeObserved()
  {
    std::vector<Promised> promisesToFulfill;
    std::vector<ValueType> valuesForPromises;

    {
      std::unique_lock<std::mutex> lck(m_mutex);
      m_pendingCacheUpdate = true;

      if(!hasEventsInCache() && !m_promises.empty()) {
        updateCache();

        while(hasEventsInCache() && !m_promises.empty()) {
          auto promised = std::move(m_promises.front());
          m_promises.pop();
          promisesToFulfill.emplace_back(std::move(promised));
          valuesForPromises.emplace_back(std::move(nextEvent()));
        }
      }
    }

    for(size_t i = 0; i < promisesToFulfill.size(); ++i) {
      Promised p = std::move(promisesToFulfill[i]);
      ValueType val = std::move(valuesForPromises[i]);
      if(!p.executor) {
        p.promise.setValue(std::move(val));
      } else  {
        folly::MoveWrapper<folly::Promise<ValueType>> promise = folly::makeMoveWrapper(std::move(p.promise));
        folly::MoveWrapper<ValueType> v = folly::makeMoveWrapper(std::move(val));
        folly::via(p.executor, [promise, v] () mutable {
          auto& unwrapped = *promise;
          unwrapped.setValue(v.move());
        });
      }
    }
  }

}