Refactor: Initiate replication per event loop

When a replication response is handled, RaftCore should initiate next
replication if there are more logs to to replicate.

Before this commit this is done in response-handler.
In this commit it is moved to `RaftCore::runtime_loop`:
replications are re-initiated when all of the events are processed.

This way the replication response handler is simplified.

openraft/src/core/raft_core.rs

@@ -40,6 +40,7 @@ use crate::display_ext::DisplayInstantExt;
 use crate::display_ext::DisplayOption;
 use crate::display_ext::DisplayOptionExt;
 use crate::display_ext::DisplaySlice;
+use crate::engine::handler::replication_handler::SendNone;
 use crate::engine::Command;
 use crate::engine::Condition;
 use crate::engine::Engine;
@@ -922,6 +923,12 @@ where
                     balancer.increase_raft_msg();
                 }
             }
+
+            // Keep replicating to a target if the replication stream to it is idle.
+            if let Ok(mut lh) = self.engine.leader_handler() {
+                lh.replication_handler().initiate_replication(SendNone::False);
+            }
+            self.run_engine_commands().await?;
         }
     }
 

openraft/src/engine/handler/replication_handler/mod.rs

@@ -1,5 +1,3 @@
-use std::ops::Deref;
-
 use crate::display_ext::DisplayOptionExt;
 use crate::engine::handler::log_handler::LogHandler;
 use crate::engine::handler::snapshot_handler::SnapshotHandler;
@@ -308,21 +306,6 @@ where C: RaftTypeConfig
         // The purge job may be postponed because a replication task is using them.
         // Thus we just try again to purge when progress is updated.
         self.try_purge_log();
-
-        // initialize next replication to this target
-
-        {
-            let p = self.leader.progress.get_mut(&target).unwrap();
-
-            let r = p.next_send(self.state.deref(), self.config.max_payload_entries);
-            tracing::debug!(next_send_res = debug(&r), "next_send");
-
-            if let Ok(inflight) = r {
-                Self::send_to_target(self.output, &target, inflight);
-            } else {
-                tracing::debug!("nothing to send to target={target}, progress:{}", p);
-            }
-        }
     }
 
     /// Update replication streams to reflect replication progress change.