StreamConsumer.hpp

#pragma once

#include <memory>

#include "ValueStream.hpp"

namespace Showcase
{

  template <typename T>
  class StreamConsumer : public std::enable_shared_from_this<StreamConsumer<T>>
  {
  public:
    StreamConsumer(ValueStreamPtr<T> stream);
    virtual ~StreamConsumer();

    void init();

    virtual void onValue(const T& val) = 0;

    void pauseConsumption();
    void continueConsumption();

  protected:
    void initListener();

  private:
    ValueStreamPtr<T> m_stream;
    bool m_consume;
  };
  template <typename T>
  using StreamConsumerPtr = std::shared_ptr<StreamConsumer<T>>;

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

  template <typename T>
  class FuncStreamConsumer : public StreamConsumer<T>
  {
  public:
    FuncStreamConsumer(ValueStreamPtr<T> stream,
                       std::function<void(const T&)> onValue);
    virtual void onValue(const T& val) override;

  private:
    std::function<void(const T&)> m_onValue;
  };

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

  template <typename T>
  StreamConsumer<T>::StreamConsumer(ValueStreamPtr<T> stream)
    : m_stream(stream),
      m_consume(true)
  {
    assert(stream);
  }

  template <typename T>
  StreamConsumer<T>::~StreamConsumer()
  {
  }

  template <typename T>
  void StreamConsumer<T>::init()
  {
    initListener();
  }

  template <typename T>
  void StreamConsumer<T>::pauseConsumption()
  {
    m_consume = false;
  }

  template <typename T>
  void StreamConsumer<T>::continueConsumption()
  {
    bool didConsume = m_consume;
    m_consume = true;
    if(!didConsume)
      initListener();
  }

  template <typename T>
  void StreamConsumer<T>::initListener()
  {
    std::weak_ptr<StreamConsumer<T>> weak = this->shared_from_this();
    auto func = [weak](T&& ev) {
      auto me = weak.lock();
      if(me) {
        auto tmp = std::move(ev);
        me->onValue(tmp);
      }

      auto f = [weak] {
        auto me = weak.lock();
        if(me && me->m_consume) {
          me->initListener();
        }
      };

      f();
    };

    //m_stream->next().via(m_executor).then(func);
    m_stream->next().thenValue(func);
  }

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

  template <typename T>
  FuncStreamConsumer<T>::FuncStreamConsumer(ValueStreamPtr<T> stream,
                                            std::function<void(const T&)> onValue)
    : StreamConsumer<T>(stream),
      m_onValue(onValue)
  {
  }

  template <typename T>
  inline void FuncStreamConsumer<T>::onValue(const T& val)
  {
    assert(m_onValue);
    m_onValue(val);
  }

}