Some fixes and basic test

Author: Stringy

collector/lib/Pipeline.h

@@ -60,18 +60,18 @@ class Queue {
     return data;
   }
 
-  std::shared_lock<std::shared_mutex> read_lock() {
+  std::shared_lock<std::shared_mutex> read_lock() const {
     return std::shared_lock(mx_);
   }
 
-  std::unique_lock<std::shared_mutex> write_lock() {
+  std::unique_lock<std::shared_mutex> write_lock() const {
     return std::unique_lock(mx_);
   }
 
  private:
   std::queue<T> inner_;
 
-  std::shared_mutex mx_;
+  mutable std::shared_mutex mx_;
 };
 
 template <class T>
@@ -81,7 +81,7 @@ class Producer {
 
   ~Producer() { Stop(); }
 
-  virtual T next();
+  virtual T next() = 0;
 
   void Start() {
     thread_.Start([this] { Run(); });
@@ -110,7 +110,7 @@ class Consumer {
 
   ~Consumer() { Stop(); }
 
-  virtual void handle(const T& event);
+  virtual void handle(const T& event) = 0;
 
   void Start() {
     thread_.Start([this] { Run(); });
@@ -140,7 +140,7 @@ class Transformer {
 
   ~Transformer() { Stop(); }
 
-  virtual std::optional<Out> transform(const In& event);
+  virtual std::optional<Out> transform(const In& event) = 0;
 
   void Start() {
     thread_.Start([this] { Run(); });

collector/test/PipelineTests.cpp

@@ -0,0 +1,75 @@
+#include <chrono>
+#include <memory>
+#include <ratio>
+#include <thread>
+#include <vector>
+
+#include "Pipeline.h"
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+
+namespace collector {
+
+class IntProducer : public Producer<int> {
+ public:
+  IntProducer(std::shared_ptr<Queue<int>>& input, int limit) : Producer(input), limit_(limit) {}
+
+  int next() override {
+    n_++;
+    if (n_ > limit_) {
+      Stop();
+    }
+    return n_;
+  }
+
+ private:
+  int n_ = 0;
+  int limit_;
+};
+
+class IntConsumer : public Consumer<int> {
+ public:
+  IntConsumer(std::shared_ptr<Queue<int>>& input, std::vector<int> output) : Consumer(input), events_(output) {}
+
+  void handle(const int& event) override {
+    events_.push_back(event);
+  }
+
+  std::vector<int>& Events() {
+    return events_;
+  }
+
+ private:
+  std::vector<int> events_;
+};
+
+class EvenIntFilter : public Filter<int> {
+ public:
+  std::optional<int> transform(const int& event) {
+    if (event % 2 == 0) {
+      return {event};
+    }
+    return std::nullopt;
+  }
+};
+
+TEST(PipelineTests, TestBasic) {
+  std::shared_ptr<Queue<int>> queue;
+
+  std::vector<int> output;
+
+  std::unique_ptr<Producer<int>> producer = std::make_unique<IntProducer>(queue, 10);
+  std::unique_ptr<Consumer<int>> consumer = std::make_unique<IntConsumer>(queue, output);
+
+  producer->Start();
+  consumer->Start();
+
+  std::this_thread::sleep_for(std::chrono::milliseconds(200));
+
+  consumer->Stop();
+  producer->Stop();
+
+  EXPECT_TRUE(output.size() == 10);
+}
+
+}  // namespace collector