pipelines.hpp

#ifndef SRC_COMMON_PIPELINE_PIPELINES_HPP_
#define SRC_COMMON_PIPELINE_PIPELINES_HPP_
 
#include <optional>
#include <stdexcept>
#include <cassert>
 
#include "common/pipeline/stages.hpp"
 
#define DEFAULT_NUM_STAGES 10
 
namespace found {
 
template<typename Input,
         typename Output,
         size_t N = DEFAULT_NUM_STAGES>
class Pipeline : public FunctionStage<Input, Output> {
 public:
    void DoAction() override {
        this->Run(this->resource);
    }
 
    virtual Output Run(const Input &input) = 0;
 
 protected:
    Action *stages[N];
    size_t size = 0;
    bool ready = false;
    std::optional<Output> finalProduct;
 
    inline void AddStageHelper(Action &stage) {
        if (this->ready) throw std::invalid_argument("Pipeline is already ready");
        this->stages[size++] = &stage;
    }
 
    inline void CompleteHelper(Action &stage) {
        assert(this->size < N);
        if (this->ready) throw std::invalid_argument("Pipeline is already ready");
        this->stages[size++] = &stage;
        this->ready = true;
    }
 
    inline void DoActionHelper() {
        for (size_t i = 0; i < this->size; i++) {
            this->stages[i]->DoAction();
        }
    }
};
 
template<typename Input, typename Output, size_t N = DEFAULT_NUM_STAGES>
class SequentialPipeline : public Pipeline<Input, Output, N> {
 public:
    SequentialPipeline() = default;
 
    template<typename I, typename O> SequentialPipeline &AddStage(FunctionStage<I, O> &stage) {
        if (this->size == 0) {
            if (!std::is_same<Input, I>::value) {
                throw std::invalid_argument("The initial input type is not correct");
            }
           this->firstResource = reinterpret_cast<Input *>(&stage.GetResource());
        } else {
            // Chain here, and blindly trust the user
            *this->lastProduct = static_cast<void *>(&stage.GetResource());
        }
        // Add to our list
        Pipeline<Input, Output, N>::AddStageHelper(stage);
        // Now, reset the lastProduct to be of this stage
        this->lastProduct = reinterpret_cast<void **>(&stage.GetProduct());
        // Return the pipeline for chaining
        return *this;
    }
 
    template<typename I> SequentialPipeline &Complete(FunctionStage<I, Output> &stage) {
        assert(this->size < N);
        if (this->ready) throw std::invalid_argument("Pipeline is already ready");
        this->AddStage(stage);
        this->ready = true;
        return *this;
    }
 
    Output Run(const Input &input) override {
        assert(!this->finalProduct);
        // MANUAL VERIFICATION: The below branch is fully tested via pipeline-test.cpp
        if (!this->ready)
            throw std::runtime_error("This is an illegal action: the pipeline is not ready yet");
        // Don't use "Input resource" unless necessary
        *this->firstResource = input;
        // If this pipeline is not composed, construct output
        // into ourself, otherwise construct into the outer pipeline
        if (this->product == nullptr) {
            // engage our finalProduct
            this->finalProduct.emplace();
            *this->lastProduct = &this->finalProduct.value();
            this->product = &this->finalProduct.value();
        } else {
            *this->lastProduct = this->product;
        }
        Pipeline<Input, Output, N>::DoActionHelper();
        return *this->product;
    }
 
 private:
    Input *firstResource = nullptr;
    void **lastProduct = nullptr;
};
 
template<typename T, size_t N = DEFAULT_NUM_STAGES>
class ModifyingPipeline : public Pipeline<T, T, N> {
 public:
    ModifyingPipeline() = default;
 
    ModifyingPipeline &AddStage(ModifyingStage<T> &stage) {
        assert(this->size < N - 1);
        Pipeline<T, T, N>::AddStageHelper(stage);
        return *this;
    }
 
    ModifyingPipeline &Complete(ModifyingStage<T> &stage) {
        assert(this->size < N);
        Pipeline<T, T, N>::CompleteHelper(stage);
        return *this;
    }
 
    T Run(const T &input) override {
        assert(!this->finalProduct);
        if (!this->ready)
            throw std::runtime_error("This is an illegal action: the pipeline is not ready yet");
        // Construct here if there is no next product,
        // or construct there if there is
        if (this->product == nullptr) {
            // engage our finalProduct
            this->finalProduct.emplace();
            this->product = &this->finalProduct.value();
        }
        *this->product = input;
        for (size_t i = 0; i < this->size; i++) {
            dynamic_cast<ModifyingStage<T> *>(this->stages[i])->SetResource(*this->product);
        }
        Pipeline<T, T, N>::DoActionHelper();
        return *this->product;
    }
};
 
}  // namespace found
 
#endif  // SRC_COMMON_PIPELINE_PIPELINES_HPP_