Refactor

codetracer-python-recorder/src/runtime/io_capture/fd_mirror/controller.rs: 
codetracer-python-recorder/src/runtime/io_capture/fd_mirror/ledger.rs: 
codetracer-python-recorder/src/runtime/io_capture/fd_mirror/mod.rs: 
codetracer-python-recorder/src/runtime/io_capture/fd_mirror/stub.rs: 
codetracer-python-recorder/src/runtime/io_capture/fd_mirror/unix.rs: 
codetracer-python-recorder/src/runtime/io_capture/fd_mirror.rs: 
codetracer-python-recorder/src/runtime/io_capture/mod.rs: 
codetracer-python-recorder/src/runtime/io_capture/pipeline.rs: 
codetracer-python-recorder/src/runtime/io_capture/proxies/common.rs: 
codetracer-python-recorder/src/runtime/io_capture/proxies/input.rs: 
codetracer-python-recorder/src/runtime/io_capture/proxies/mod.rs: 
codetracer-python-recorder/src/runtime/io_capture/proxies/output.rs: 
codetracer-python-recorder/src/runtime/io_capture/proxies.rs: 
codetracer-python-recorder/src/runtime/io_capture/sink/batcher.rs: 
codetracer-python-recorder/src/runtime/io_capture/sink/enricher.rs: 
codetracer-python-recorder/src/runtime/io_capture/sink/mod.rs: 
codetracer-python-recorder/src/runtime/io_capture/sink/types.rs: 
codetracer-python-recorder/src/runtime/io_capture/sink.rs: 
codetracer-python-recorder/src/runtime/mod.rs: 

Signed-off-by: Tzanko Matev <[email protected]>

Author: tzanko-matev

codetracer-python-recorder/src/runtime/io_capture/fd_mirror.rs

@@ -0,0 +1,46 @@
+use super::ledger::MirrorLedgers;
+use crate::runtime::io_capture::sink::IoChunkConsumer;
+use std::sync::Arc;
+
+#[cfg(not(unix))]
+use super::stub as unix;
+#[cfg(unix)]
+use super::unix;
+
+#[cfg(not(unix))]
+pub use super::stub::FdMirrorError;
+#[cfg(unix)]
+pub use super::unix::FdMirrorError;
+
+pub struct FdMirrorController {
+    inner: Option<unix::FdMirrorController>,
+}
+
+impl FdMirrorController {
+    pub fn new(
+        ledgers: MirrorLedgers,
+        consumer: Arc<dyn IoChunkConsumer>,
+    ) -> Result<Self, FdMirrorError> {
+        let inner = if let Some(set) = ledgers.inner() {
+            Some(unix::FdMirrorController::new(set, consumer)?)
+        } else {
+            None
+        };
+        Ok(Self { inner })
+    }
+
+    pub fn shutdown(&mut self) {
+        if let Some(inner) = self.inner.as_mut() {
+            inner.shutdown();
+        }
+        self.inner = None;
+    }
+}
+
+impl Drop for FdMirrorController {
+    fn drop(&mut self) {
+        if let Some(inner) = self.inner.as_mut() {
+            inner.shutdown();
+        }
+    }
+}

codetracer-python-recorder/src/runtime/io_capture/fd_mirror/controller.rs

@@ -0,0 +1,42 @@
+use crate::runtime::io_capture::events::IoStream;
+use std::sync::Arc;
+
+#[cfg(not(unix))]
+use super::stub::MirrorLedgerSet;
+#[cfg(unix)]
+use super::unix::MirrorLedgerSet;
+
+#[derive(Clone, Default)]
+pub struct MirrorLedgers(Option<Arc<MirrorLedgerSet>>);
+
+impl MirrorLedgers {
+    pub fn new_enabled() -> Self {
+        #[cfg(unix)]
+        {
+            Self(Some(Arc::new(MirrorLedgerSet::new())))
+        }
+        #[cfg(not(unix))]
+        {
+            Self(None)
+        }
+    }
+
+    pub fn is_enabled(&self) -> bool {
+        self.0.is_some()
+    }
+
+    pub fn begin_proxy_write(&self, stream: IoStream, payload: &[u8]) -> Option<LedgerTicket> {
+        self.0
+            .as_ref()
+            .and_then(|inner| inner.begin_proxy_write(stream, payload))
+    }
+
+    pub(crate) fn inner(&self) -> Option<Arc<MirrorLedgerSet>> {
+        self.0.clone()
+    }
+}
+
+#[cfg(not(unix))]
+pub use super::stub::LedgerTicket;
+#[cfg(unix)]
+pub use super::unix::LedgerTicket;

codetracer-python-recorder/src/runtime/io_capture/fd_mirror/ledger.rs

@@ -0,0 +1,9 @@
+mod controller;
+mod ledger;
+#[cfg(not(unix))]
+mod stub;
+#[cfg(unix)]
+mod unix;
+
+pub use controller::FdMirrorController;
+pub use ledger::{LedgerTicket, MirrorLedgers};

codetracer-python-recorder/src/runtime/io_capture/fd_mirror/mod.rs

@@ -0,0 +1,56 @@
+use crate::runtime::io_capture::events::IoStream;
+use crate::runtime::io_capture::sink::IoChunkConsumer;
+use std::sync::Arc;
+
+#[derive(Debug, Clone)]
+pub struct FdMirrorError {
+    message: String,
+}
+
+impl FdMirrorError {
+    pub fn new(message: impl Into<String>) -> Self {
+        Self {
+            message: message.into(),
+        }
+    }
+}
+
+impl std::fmt::Display for FdMirrorError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.message)
+    }
+}
+
+impl std::error::Error for FdMirrorError {}
+
+#[derive(Default)]
+pub struct MirrorLedgerSet;
+
+pub struct LedgerTicket;
+
+pub struct FdMirrorController;
+
+impl MirrorLedgerSet {
+    pub fn new() -> Self {
+        Self
+    }
+
+    pub fn begin_proxy_write(&self, _: IoStream, _: &[u8]) -> Option<LedgerTicket> {
+        None
+    }
+}
+
+impl LedgerTicket {
+    pub fn commit(self) {}
+}
+
+impl FdMirrorController {
+    pub fn new(
+        _: Arc<MirrorLedgerSet>,
+        _: Arc<dyn IoChunkConsumer>,
+    ) -> Result<Self, FdMirrorError> {
+        Ok(Self)
+    }
+
+    pub fn shutdown(&mut self) {}
+}