Make it not do those segfault thingies

Author: Stringy

collector/lib/Pipeline.h

@@ -1,6 +1,8 @@
 #ifndef _COLLECTOR_PIPELINE_H
 #define _COLLECTOR_PIPELINE_H
 
+#include <chrono>
+#include <condition_variable>
 #include <memory>
 #include <mutex>
 #include <optional>
@@ -17,16 +19,6 @@ class Queue {
   Queue() {}
   ~Queue() {}
 
-  T front() {
-    auto lock = read_lock();
-    return inner_.front();
-  }
-
-  T back() {
-    auto lock = read_lock();
-    return inner_.back();
-  }
-
   bool empty() {
     auto lock = read_lock();
     return inner_.empty();
@@ -38,34 +30,36 @@ class Queue {
   }
 
   void push(const T& elem) {
-    auto lock = write_lock();
-    auto e = elem;
-    inner_.push(std::move(e));
+    {
+      auto lock = write_lock();
+      auto e = elem;
+      inner_.push(std::move(e));
+    }
     state_changed_.notify_one();
   }
 
   void push(T&& elem) {
-    auto lock = write_lock();
-    inner_.push(elem);
+    {
+      auto lock = write_lock();
+      inner_.push(elem);
+    }
     state_changed_.notify_one();
   }
 
-  template <class... Args>
-  decltype(auto) emplace(Args&&... args) {
+  std::optional<T> pop(std::chrono::milliseconds wait_max = std::chrono::milliseconds(10)) {
     auto lock = write_lock();
-    decltype(auto) out = inner_.emplace(std::forward<Args>(args)...);
-    state_changed_.notify_one();
-    return out;
-  }
+    if (inner_.empty()) {
+      auto pred = [this]() {
+        return !inner_.empty();
+      };
 
-  T pop() {
-    auto lock = write_lock();
-    if (empty()) {
-      state_changed_.wait(lock, [this]() { return empty(); });
+      if (!state_changed_.wait_for(lock, wait_max, pred)) {
+        return std::nullopt;
+      }
     }
     T data = inner_.front();
     inner_.pop();
-    return data;
+    return {data};
   }
 
   std::shared_lock<std::shared_mutex> read_lock() const {
@@ -94,7 +88,7 @@ class Producer {
     }
   }
 
-  virtual T next() = 0;
+  virtual std::optional<T> next() = 0;
 
   void Start() {
     thread_.Start([this] { Run(); });
@@ -107,7 +101,10 @@ class Producer {
   void Run() {
     while (!thread_.should_stop()) {
       auto event = next();
-      output_->push(event);
+      if (!event.has_value()) {
+        break;
+      }
+      output_->push(event.value());
     }
   }
 
@@ -140,7 +137,10 @@ class Consumer {
   void Run() {
     while (!thread_.should_stop()) {
       auto event = input_->pop();
-      handle(event);
+      if (!event.has_value()) {
+        continue;
+      }
+      handle(event.value());
     }
   }
 
@@ -170,7 +170,11 @@ class Transformer {
   void Run() {
     while (!thread_.should_stop()) {
       auto event = input_->pop();
-      auto transformed = transform(event);
+      if (!event.has_value()) {
+        continue;
+      }
+
+      auto transformed = transform(event.value());
       if (transformed.has_value()) {
         output_.push(transformed.value());
       }

collector/test/PipelineTests.cpp

@@ -1,5 +1,6 @@
 #include <chrono>
 #include <memory>
+#include <optional>
 #include <ratio>
 #include <thread>
 #include <vector>
@@ -14,12 +15,12 @@ class IntProducer : public Producer<int> {
  public:
   IntProducer(std::shared_ptr<Queue<int>>& input, int limit) : Producer(input), limit_(limit) {}
 
-  int next() override {
+  std::optional<int> next() override {
     n_++;
     if (n_ > limit_) {
-      return limit_;
+      return std::nullopt;
     }
-    return n_;
+    return {n_};
   }
 
  private:
@@ -29,18 +30,14 @@ class IntProducer : public Producer<int> {
 
 class IntConsumer : public Consumer<int> {
  public:
-  IntConsumer(std::shared_ptr<Queue<int>>& input, std::vector<int> output) : Consumer(input), events_(output) {}
+  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_;
+  std::vector<int>& events_;
 };
 
 class EvenIntFilter : public Filter<int> {