Feature: Raft::client_write_ff() ff for fire-and-forget

`Raft<C>::client_write_ff() -> C::Responder::Receiver` submit a client
request to Raft to update the state machine, returns an application
defined response receiver `Responder::Receiver` to receive the response.

`_ff` means fire and forget.

It is same as [`Raft::client_write`] but does not wait for the response.
When using this method, it is the application's responsibility for
defining mechanism building and consuming the `Responder::Receiver`.

- Part of #1068

openraft/src/raft/mod.rs

@@ -689,15 +689,29 @@ where C: RaftTypeConfig
  ResponderReceiverOf<C>: Future<Output = Result<WriteResult<C>, E>>,
  E: Error + OptionalSend,
  {
- let (app_data, tx, rx) = ResponderOf::<C>::from_app_data(app_data);
+ let rx = self.client_write_ff(app_data).await?;
 
- self.inner.send_msg(RaftMsg::ClientWriteRequest { app_data, tx }).await?;
  let res: WriteResult<C> = self.inner.recv_msg(rx).await?;
 
  let client_write_response = res.map_err(|e| RaftError::APIError(e))?;
  Ok(client_write_response)
  }
 
+ /// Submit a mutating client request to Raft to update the state machine, returns an application
+ /// defined response receiver [`Responder::Receiver`].
+ ///
+ /// `_ff` means fire and forget.
+ ///
+ /// It is same as [`Raft::client_write`] but does not wait for the response.
+ #[tracing::instrument(level = "debug", skip(self, app_data))]
+ pub async fn client_write_ff(&self, app_data: C::D) -> Result<ResponderReceiverOf<C>, Fatal<C>> {
+ let (app_data, tx, rx) = ResponderOf::<C>::from_app_data(app_data);
+
+ self.inner.send_msg(RaftMsg::ClientWriteRequest { app_data, tx }).await?;
+
+ Ok(rx)
+ }
+
  /// Return `true` if this node is already initialized and can not be initialized again with
  /// [`Raft::initialize`]
  pub async fn is_initialized(&self) -> Result<bool, Fatal<C>> {

tests/tests/client_api/t10_client_writes.rs

@@ -3,18 +3,18 @@ use std::sync::Arc;
 use anyhow::Result;
 use futures::prelude::*;
 use maplit::btreeset;
+use openraft::raft::ClientWriteResponse;
 use openraft::CommittedLeaderId;
 use openraft::Config;
 use openraft::LogId;
 use openraft::SnapshotPolicy;
+use openraft_memstore::ClientRequest;
+use openraft_memstore::IntoMemClientRequest;
+use openraft_memstore::TypeConfig;
 
 use crate::fixtures::init_default_ut_tracing;
 use crate::fixtures::RaftRouter;
 
-/// Client write tests.
-///
-/// What does this test do?
-///
 /// - create a stable 3-node cluster.
 /// - write a lot of data to it.
 /// - assert that the cluster stayed stable and has all of the expected data.
@@ -63,3 +63,35 @@ async fn client_writes() -> Result<()> {
 
  Ok(())
 }
+
+/// Test Raft::client_write_ff,
+///
+/// Manually receive the client-write response via the returned `Responder::Receiver`
+#[async_entry::test(worker_threads = 4, init = "init_default_ut_tracing()", tracing_span = "debug")]
+async fn client_write_ff() -> Result<()> {
+ let config = Arc::new(
+ Config {
+ enable_tick: false,
+ ..Default::default()
+ }
+ .validate()?,
+ );
+
+ let mut router = RaftRouter::new(config.clone());
+
+ tracing::info!("--- initializing cluster");
+ let log_index = router.new_cluster(btreeset! {0,1,2}, btreeset! {}).await?;
+ let _ = log_index;
+
+ let n0 = router.get_raft_handle(&0)?;
+
+ let resp_rx = n0.client_write_ff(ClientRequest::make_request("foo", 2)).await?;
+ let got: ClientWriteResponse<TypeConfig> = resp_rx.await??;
+ assert_eq!(None, got.response().0.as_deref());
+
+ let resp_rx = n0.client_write_ff(ClientRequest::make_request("foo", 3)).await?;
+ let got: ClientWriteResponse<TypeConfig> = resp_rx.await??;
+ assert_eq!(Some("request-2"), got.response().0.as_deref());
+
+ Ok(())
+}