feat(pipeline): add pipeline header

CMakeLists.txt

@@ -9,8 +9,16 @@ project(
 find_package(spdlog REQUIRED)
 find_package(avcpp REQUIRED)
 
+
 # ##############################################################################
 
+add_library(mosh-me-lib STATIC
+    ${PROJECT_SOURCE_DIR}/src/pipeline.cpp
+)
+target_link_libraries(mosh-me-lib PUBLIC avcpp::avcpp spdlog::spdlog)
+target_compile_features(mosh-me-lib PUBLIC cxx_std_23)
+target_include_directories(mosh-me-lib PUBLIC ${PROJECT_SOURCE_DIR}/include)
+
 add_executable(mosh-me
     ${PROJECT_SOURCE_DIR}/app/mosh-me.cpp
 )
@@ -30,7 +38,19 @@ install(TARGETS mosh-me mk-moshable)
 
 ################################################################################
 
+if (BUILD_TESTING)
+  find_package(Catch2 REQUIRED)
+  add_executable(mosh-me_test
+    ${PROJECT_SOURCE_DIR}/test/pipeline.cpp
+  )
+  target_link_libraries(mosh-me_test mosh-me-lib Catch2::Catch2WithMain)
+endif()
+
+################################################################################
+
 if(CMAKE_EXPORT_COMPILE_COMMANDS)
   set(CMAKE_CXX_STANDARD_INCLUDE_DIRECTORIES ${CMAKE_CXX_IMPLICIT_INCLUDE_DIRECTORIES})
   set(CMAKE_C_STANDARD_INCLUDE_DIRECTORIES ${CMAKE_C_IMPLICIT_INCLUDE_DIRECTORIES})
 endif()
+
+

flake.nix

@@ -41,6 +41,7 @@
             export CMAKE_EXPORT_COMPILE_COMMANDS=ON
             export CMAKE_BUILD_TYPE=Debug
             export CMAKE_GENERATOR=Ninja
+            export BUILD_TESTING=ON
           '';
           packages = [
             pkgs.clang-tools

include/mosh-me/pipeline.hpp

@@ -0,0 +1,86 @@
+#pragma once
+#include <memory>
+#include <optional>
+
+namespace mosh_me {
+
+template <typename A, typename B>
+concept AttachableTo = requires {
+  typename A::output_type;
+  typename B::input_type;
+} && std::convertible_to<typename A::output_type, typename B::input_type>;
+
+template <typename Output> class PipeOutput {
+public:
+  virtual std::optional<Output> pull() noexcept = 0;
+};
+
+template <typename Input> class PipeInput {
+private:
+  std::shared_ptr<PipeOutput<Input>> input_;
+
+protected:
+  std::optional<Input> fetchInput() { return input_->pull(); }
+
+public:
+  void linkInput(std::shared_ptr<PipeOutput<Input>> input) { input_ = input; };
+};
+
+template <typename Input, typename Output>
+class PipeWorker : public PipeOutput<Output>, public PipeInput<Input> {};
+
+template <typename Output> class PipeSource {
+public:
+  std::shared_ptr<PipeOutput<Output>> source_;
+
+  PipeSource(std::shared_ptr<PipeOutput<Output>> p) : source_(p) {}
+
+  std::optional<Output> pull() {
+    if (source_) {
+      return source_->pull();
+    } else {
+      return std::nullopt;
+    }
+  }
+};
+
+template <typename Input, typename Output> class PipeLink {
+public:
+  std::shared_ptr<PipeInput<Input>> input_;
+  std::shared_ptr<PipeOutput<Output>> output_;
+
+  PipeLink(std::shared_ptr<PipeInput<Input>> i,
+           std::shared_ptr<PipeOutput<Output>> o)
+      : input_(i), output_(o) {}
+  PipeLink(std::shared_ptr<PipeWorker<Input, Output>> p)
+      : input_(p), output_(p) {}
+
+  template <typename I, typename O>
+  PipeLink<I, Output> operator<<(PipeLink<I, O> &&other) {
+    input_->linkInput(std::move(other.output_));
+    return PipeLink<I, Output>(std::move(other.input_), std::move(output_));
+  }
+
+  template <typename O> PipeSource<Output> operator<<(PipeSource<O> &&other) {
+    input_->linkInput(std::move(other.source_));
+    return PipeSource<Output>(std::move(output_));
+  }
+};
+
+template <typename PipeType, typename O> class AsPipeSource {
+public:
+  template <typename... Args> static PipeSource<O> link(Args &&...args) {
+    return PipeSource<O>(
+        std::make_shared<PipeType>(std::forward<Args>(args)...));
+  }
+};
+
+template <typename PipeType, typename I, typename O> class AsPipeLink {
+public:
+  template <typename... Args> static PipeLink<I, O> link(Args &&...args) {
+    return PipeLink<I, O>(
+        std::make_shared<PipeType>(std::forward<Args>(args)...));
+  }
+};
+
+} // namespace mosh_me

nix/mosh-me.nix

@@ -6,12 +6,13 @@
   stdenv,
   pkg-config,
   avcpp,
+  catch2_3,
 }:
 stdenv.mkDerivation (finalAttrs: {
   name = "mosh-me";
   src = ../.;
   nativeBuildInputs = [cmake ninja pkg-config];
-  buildInputs = [spdlog avcpp];
+  buildInputs = [spdlog avcpp] ++ lib.optional finalAttrs.doCheck catch2_3;
   cmakeFlags = [
     (lib.cmakeBool "BUILD_TESTING" finalAttrs.doCheck)
   ];

test/pipeline.cpp

@@ -0,0 +1,75 @@
+#include "mosh-me/pipeline.hpp"
+#include <catch2/catch_all.hpp>
+#include <spdlog/fmt/fmt.h>
+#include <spdlog/fmt/ranges.h>
+
+using namespace mosh_me;
+
+class CountProducer : public PipeOutput<int>,
+                      public AsPipeSource<CountProducer, int> {
+private:
+  int counter_ = 0;
+
+public:
+  std::optional<int> pull() noexcept override {
+    if (auto i = counter_++; i < 10) {
+      return i;
+    } else {
+      return std::nullopt;
+    }
+  }
+};
+
+class Doubler : public PipeWorker<int, int>,
+                public AsPipeLink<Doubler, int, int> {
+public:
+  std::optional<int> pull() noexcept override {
+    return fetchInput().and_then(
+        [](int x) { return std::make_optional(x * 2); });
+  }
+};
+
+class Multiplier : public PipeWorker<int, int>,
+                   public AsPipeLink<Multiplier, int, int> {
+  int f_;
+
+public:
+  Multiplier(int f) : f_(f) {}
+  std::optional<int> pull() noexcept override {
+    return fetchInput().and_then(
+        [this](int x) { return std::make_optional(f_ * x); });
+  }
+};
+
+class Sum2 : public PipeWorker<int, int>, public AsPipeLink<Sum2, int, int> {
+private:
+  std::optional<int> last_;
+
+public:
+  std::optional<int> pull() noexcept override {
+    if (auto x = fetchInput(); x) {
+      if (last_) {
+        auto sum = *last_ + *x;
+        last_ = std::nullopt;
+        return sum;
+      } else {
+        last_ = x;
+        return pull();
+      }
+    }
+    return std::nullopt;
+  }
+};
+
+TEST_CASE("Pipeline") {
+  auto pipe = Sum2::link() << Doubler::link() << Multiplier::link(10)
+                           << CountProducer::link();
+
+  auto count = 0;
+  while (auto s = pipe.pull()) {
+    REQUIRE(*s == ((count * 2) * 20) + ((count * 2 + 1) * 20));
+    count++;
+  }
+
+  REQUIRE(count == 5);
+}