feat: introduce the special Pipe collect_to_vec to use for test

cdviz-collector/src/errors.rs

@@ -43,6 +43,8 @@ pub(crate) enum Error {
     ConfigReader(figment::Error),
     CloudEventBuilder(cloudevents::event::EventBuilderError),
     CloudEventMessage(cloudevents::message::Error),
+    // MutexPoisoned(std::sync::PoisonError<std::sync::MutexGuard<'static, Message>>),
+    // MutexPoisoned<T>(std::sync::PoisonError<T>),
     // ConfigTomlError(toml::de::Error),
     MultiHash(multihash::Error),
     #[display("{txt}")]

cdviz-collector/src/pipes/collect_to_vec.rs

@@ -0,0 +1,53 @@
+#[allow(dead_code)]
+use super::Pipe;
+use crate::errors::{Error, Result};
+use serde::{Deserialize, Serialize};
+use std::{
+    fmt::Debug,
+    sync::{Arc, Mutex},
+};
+
+#[derive(Debug, Deserialize, Serialize)]
+pub(crate) struct Config {}
+
+type Buffer<I> = Arc<Mutex<Vec<I>>>;
+pub struct Processor<I> {
+    // TODO do we need to use Mutex?
+    buffer: Buffer<I>,
+}
+
+impl<I> Pipe for Processor<I> {
+    type Input = I;
+    fn send(&mut self, input: Self::Input) -> Result<()> {
+        //.lock().unwrap() if mutex
+        self.buffer.lock().map_err(|err| Error::from(err.to_string()))?.push(input);
+        Ok(())
+    }
+}
+
+pub struct Collector<I> {
+    buffer: Buffer<I>,
+}
+
+impl<I> Collector<I>
+where
+    I: Clone,
+{
+    #[allow(dead_code)] // mainly use in tests
+    pub fn new() -> Self {
+        Self { buffer: Arc::new(Mutex::new(vec![])) }
+    }
+
+    #[allow(dead_code)] // mainly use in tests
+    pub fn create_pipe(&self) -> Processor<I> {
+        Processor { buffer: Arc::clone(&self.buffer) }
+    }
+
+    #[allow(dead_code)] // mainly use in tests
+    pub fn try_into_iter(&self) -> Result<std::vec::IntoIter<I>> {
+        self.buffer
+            .lock()
+            .map_err(|err| Error::from(err.to_string()))
+            .map(|buffer| buffer.clone().into_iter())
+    }
+}

cdviz-collector/src/pipes/mod.rs

@@ -1,4 +1,5 @@
 use crate::errors::Result;
+pub mod collect_to_vec;
 pub mod discard_all;
 pub mod log;
 pub mod passthrough;

cdviz-collector/src/sources/mod.rs

@@ -67,7 +67,7 @@ pub(crate) fn start(_name: &str, config: Config, tx: Sender<Message>) -> JoinHan
     })
 }
 
-#[derive(Debug, Deserialize, Serialize, Default)]
+#[derive(Debug, Clone, Deserialize, Serialize, Default, PartialEq, Eq)]
 pub struct EventSource {
     pub metadata: Value,
     pub header: HashMap<String, String>,