PipelineStage.h

Go to the documentation of this file.

#ifndef _SEFRAMEWORK_PIPELINE_PIPELINESTAGE_H
#define _SEFRAMEWORK_PIPELINE_PIPELINESTAGE_H

#include "SEFramework/Source/SourceInterface.h"
#include "SEUtils/Observable.h"

namespace SourceXtractor {

class SelectionCriteria;

struct ProcessSourcesEvent {

  const std::shared_ptr<SelectionCriteria> m_selection_criteria;  // Used to identify the Sources to process

  explicit ProcessSourcesEvent(std::shared_ptr<SelectionCriteria> selection_criteria)
      : m_selection_criteria(std::move(selection_criteria)) {}
};

template <typename T>
class PipelineReceiver {
public:
  virtual ~PipelineReceiver() = default;

  virtual void receiveSource(std::unique_ptr<T> source)        = 0;

  virtual void receiveProcessSignal(const ProcessSourcesEvent& event) = 0;
};

template <typename T>
class PipelineEmitter : public Observable<T> {
public:
  ~PipelineEmitter() override = default;

  void setNextStage(std::shared_ptr<PipelineReceiver<T>> next) {
    if (m_next_stage) {
      throw Elements::Exception() << "Next stage already set";
    }
    m_next_stage = std::move(next);
  }

protected:
  void sendSource(std::unique_ptr<T> source) const {
    Observable<T>::notifyObservers(*source);
    if (m_next_stage) {
      m_next_stage->receiveSource(std::move(source));
    }
  }

  void sendProcessSignal(const ProcessSourcesEvent& event) const {
    if (m_next_stage) {
      m_next_stage->receiveProcessSignal(event);
    }
  }

private:
  std::shared_ptr<PipelineReceiver<T>> m_next_stage;
};

}  // namespace SourceXtractor

#endif  // _SEFRAMEWORK_PIPELINE_PIPELINESTAGE_H