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