visibility control

Author: Migorithm

duva/src/adapters/loggers/disk_based.rs

@@ -179,7 +179,7 @@ impl Segment {
 }
 
 impl FileOpLogs {
-    pub fn new<P: AsRef<Path>>(path: P) -> Result<Self> {
+    pub(super) fn new<P: AsRef<Path>>(path: P) -> Result<Self> {
         let path = path.as_ref().to_path_buf();
         Self::validate_folder(&path)?;
         let segment_paths = Self::detect_and_sort_existing_segments(&path)?;

duva/src/adapters/loggers/memory_based.rs

@@ -5,7 +5,7 @@ use anyhow::Result;
 
 #[derive(Default, Clone, Debug)]
 pub struct MemoryOpLogs {
-    pub writer: Vec<WriteOperation>,
+    writer: Vec<WriteOperation>,
 }
 
 impl TWriteAheadLog for MemoryOpLogs {

duva/src/domains/cluster_actors/actor.rs

@@ -6,7 +6,6 @@ use super::hash_ring::HashRing;
 pub mod client_sessions;
 pub(crate) mod heartbeat_scheduler;
 use super::*;
-use crate::adapters::loggers::op_logs::OperationLogs;
 use crate::domains::QueryIO;
 use crate::domains::TAsyncReadWrite;
 use crate::domains::caches::cache_manager::CacheManager;
@@ -59,7 +58,7 @@ const FANOUT: usize = 2;
 #[derive(Debug)]
 pub struct ClusterActor {
     pub(crate) members: BTreeMap<PeerIdentifier, Peer>,
-    pub(crate) replication: Replication<OperationLogs>,
+    pub(crate) replication: Replication,
     pub(crate) consensus_tracker: LogConsensusTracker,
     pub(crate) receiver: ClusterActorReceiver,
     pub(crate) self_handler: ClusterActorSender,
@@ -98,7 +97,7 @@ impl ClusterActor {
     pub(crate) fn run(
         topology_writer: std::fs::File,
         heartbeat_interval: u64,
-        replication: Replication<OperationLogs>,
+        replication: Replication,
         cache_manager: CacheManager,
     ) -> ClusterActorSender {
         let cluster_actor =
@@ -113,27 +112,27 @@ impl ClusterActor {
     }
 
     fn new(
-        init_repl_state: Replication<OperationLogs>,
+        init_repl: Replication,
         heartbeat_interval_in_mills: u64,
         topology_writer: File,
         cache_manager: CacheManager,
     ) -> Self {
         let (self_handler, receiver) = ClusterActorQueue::create(2000);
         let heartbeat_scheduler = HeartBeatScheduler::run(
             self_handler.clone(),
-            init_repl_state.is_leader(),
+            init_repl.is_leader(),
             heartbeat_interval_in_mills,
         );
 
         let (tx, _) = tokio::sync::broadcast::channel::<Topology>(100);
         let hash_ring = HashRing::default().add_partitions(vec![(
-            init_repl_state.clone_state().replid.clone(),
-            init_repl_state.self_identifier(),
+            init_repl.clone_state().replid.clone(),
+            init_repl.self_identifier(),
         )]);
 
         Self {
             heartbeat_scheduler,
-            replication: init_repl_state,
+            replication: init_repl,
             receiver,
             self_handler,
             topology_writer,

duva/src/domains/cluster_actors/actor/tests/mod.rs

@@ -9,6 +9,7 @@ use crate::CacheManager;
 use crate::Replication;
 use crate::ReplicationId;
 
+use crate::adapters::loggers::op_logs::OperationLogs;
 use crate::domains::TSerdeDynamicRead;
 use crate::domains::TSerdeDynamicWrite;
 use crate::domains::caches::actor::CacheCommandSender;

duva/src/domains/cluster_actors/actor/tests/replications.rs

@@ -1,3 +1,5 @@
+use crate::adapters::loggers::op_logs::OperationLogs;
+
 use super::*;
 
 #[test]
@@ -356,7 +358,8 @@ async fn follower_truncates_log_on_term_mismatch() {
     //prefill
 
     inmemory
-        .write_many(vec![Helper::write(2, 1, "key1", "val1"), Helper::write(3, 1, "key2", "val2")]);
+        .write_many(vec![Helper::write(2, 1, "key1", "val1"), Helper::write(3, 1, "key2", "val2")])
+        .unwrap();
 
     let mut cluster_actor = Helper::cluster_actor(ReplicationRole::Leader).await;
     cluster_actor.replication.set_target(inmemory);

duva/src/domains/replications/replication.rs

@@ -16,18 +16,18 @@ use std::sync::atomic::AtomicU64;
 use std::sync::atomic::Ordering;
 
 #[derive(Debug)]
-pub(crate) struct Replication<T> {
+pub(crate) struct Replication {
     pub(crate) self_port: u16,
     election_votes: ElectionVotes,
-    logger: T,
+    logger: OperationLogs,
     state: ReplicationState,
     last_log_term: u64,
     con_idx: Arc<AtomicU64>,
     last_applied: u64,
     in_mem_buffer: Vec<WriteOperation>,
 }
 
-impl Replication<OperationLogs> {
+impl Replication {
     pub(crate) fn new(self_port: u16, logger: OperationLogs, state: ReplicationState) -> Self {
         Self {
             election_votes: ElectionVotes::default(),