refactor: add Builder for Session to avoid too many arguments

Author: kezhuw

src/client/mod.rs

@@ -11,8 +11,7 @@ use either::{Either, Left, Right};
 use ignore_result::Ignore;
 use thiserror::Error;
 use tokio::sync::mpsc;
-use tracing::field::display;
-use tracing::{instrument, Span};
+use tracing::instrument;
 
 pub use self::watcher::{OneshotWatcher, PersistentWatcher, StateWatcher};
 use super::session::{Depot, MarshalledRequest, Session, SessionOperation, WatchReceiver};
@@ -1649,32 +1648,17 @@ impl Connector {
     #[instrument(name = "connect", skip_all, fields(session))]
     async fn connect_internally(&mut self, secure: bool, cluster: &str) -> Result<Client> {
         let (endpoints, chroot) = endpoint::parse_connect_string(cluster, secure)?;
-        if let Some(session) = self.session.as_ref() {
-            if session.is_readonly() {
-                return Err(Error::new_other(
-                    format!("can't reestablish readonly and hence local session {}", session.id()),
-                    None,
-                ));
-            }
-            Span::current().record("session", display(session.id()));
-        }
-        if self.session_timeout < Duration::ZERO {
-            return Err(Error::BadArguments(&"session timeout must not be negative"));
-        } else if self.connection_timeout < Duration::ZERO {
-            return Err(Error::BadArguments(&"connection timeout must not be negative"));
-        }
-        let tls_config = self.tls.take().unwrap_or_default().into_config()?;
-        let (mut session, state_receiver) = Session::new(
-            self.session.take(),
-            &self.authes,
-            self.readonly,
-            self.detached,
-            tls_config,
-            #[cfg(any(feature = "sasl-digest-md5", feature = "sasl-gssapi"))]
-            self.sasl.take(),
-            self.session_timeout,
-            self.connection_timeout,
-        );
+        let builder = Session::builder()
+            .with_tls(self.tls.take())
+            .with_session(self.session.take())
+            .with_authes(&self.authes)
+            .with_readonly(self.readonly)
+            .with_detached(self.detached)
+            .with_session_timeout(self.session_timeout)
+            .with_connection_timeout(self.connection_timeout);
+        #[cfg(any(feature = "sasl-digest-md5", feature = "sasl-gssapi"))]
+        let builder = builder.with_sasl(self.sasl.take());
+        let (mut session, state_receiver) = builder.build()?;
         let mut endpoints = IterableEndpoints::from(endpoints.as_slice());
         endpoints.reset();
         if !self.fail_eagerly {

src/session/mod.rs

@@ -10,7 +10,6 @@ use std::io;
 use std::time::Duration;
 
 use ignore_result::Ignore;
-use rustls::ClientConfig;
 use tokio::select;
 use tokio::sync::mpsc;
 use tokio::time::{self, Instant};
@@ -39,6 +38,7 @@ use crate::proto::{AuthPacket, ConnectRequest, ConnectResponse, ErrorCode, OpCod
 use crate::record;
 #[cfg(any(feature = "sasl-digest-md5", feature = "sasl-gssapi"))]
 use crate::sasl::{SaslInitiator, SaslOptions, SaslSession};
+use crate::tls::TlsOptions;
 
 pub const PASSWORD_LEN: usize = 16;
 pub const DEFAULT_SESSION_TIMEOUT: Duration = Duration::from_secs(6);
@@ -59,59 +59,81 @@ impl RequestOperation for (WatcherId, StateResponser) {
     }
 }
 
-pub struct Session {
+#[derive(Default)]
+pub struct Builder {
+    tls: Option<TlsOptions>,
+    #[cfg(any(feature = "sasl-digest-md5", feature = "sasl-gssapi"))]
+    sasl: Option<SaslOptions>,
+    authes: Vec<MarshalledRequest>,
     readonly: bool,
     detached: bool,
+    session: Option<SessionInfo>,
+    session_timeout: Duration,
+    connection_timeout: Duration,
+}
 
-    connector: Connector,
+impl Builder {
+    pub fn with_tls(self, tls: Option<TlsOptions>) -> Self {
+        Self { tls, ..self }
+    }
 
-    configured_connection_timeout: Duration,
+    #[cfg(any(feature = "sasl-digest-md5", feature = "sasl-gssapi"))]
+    pub fn with_sasl(self, sasl: Option<SaslOptions>) -> Self {
+        Self { sasl, ..self }
+    }
 
-    last_zxid: i64,
-    last_recv: Instant,
-    last_send: Instant,
-    last_ping: Option<Instant>,
-    tick_timeout: Duration,
-    ping_timeout: Duration,
-    session_expired_timeout: Duration,
+    pub fn with_authes(self, authes: &[AuthPacket]) -> Self {
+        Self { authes: authes.iter().map(|auth| MarshalledRequest::new(OpCode::Auth, auth)).collect(), ..self }
+    }
 
-    pub session: SessionInfo,
-    session_state: SessionState,
-    pub session_timeout: Duration,
+    pub fn with_readonly(self, readonly: bool) -> Self {
+        Self { readonly, ..self }
+    }
 
-    #[cfg(any(feature = "sasl-digest-md5", feature = "sasl-gssapi"))]
-    sasl_options: Option<SaslOptions>,
-    #[cfg(any(feature = "sasl-digest-md5", feature = "sasl-gssapi"))]
-    sasl_session: Option<SaslSession>,
+    pub fn with_detached(self, detached: bool) -> Self {
+        Self { detached, ..self }
+    }
 
-    pub authes: Vec<MarshalledRequest>,
-    state_sender: tokio::sync::watch::Sender<SessionState>,
+    pub fn with_session(self, session: Option<SessionInfo>) -> Self {
+        Self { session, ..self }
+    }
 
-    watch_manager: WatchManager,
-    unwatch_receiver: Option<mpsc::UnboundedReceiver<(WatcherId, StateResponser)>>,
-}
+    pub fn with_session_timeout(self, session_timeout: Duration) -> Self {
+        Self { session_timeout, ..self }
+    }
 
-impl Session {
-    #[allow(clippy::too_many_arguments)]
-    pub fn new(
-        session: Option<SessionInfo>,
-        authes: &[AuthPacket],
-        readonly: bool,
-        detached: bool,
-        tls_config: ClientConfig,
-        #[cfg(any(feature = "sasl-digest-md5", feature = "sasl-gssapi"))] sasl_options: Option<SaslOptions>,
-        session_timeout: Duration,
-        connection_timeout: Duration,
-    ) -> (Session, tokio::sync::watch::Receiver<SessionState>) {
-        let session = session.unwrap_or_else(|| SessionInfo::new(SessionId(0), Vec::with_capacity(PASSWORD_LEN)));
+    pub fn with_connection_timeout(self, connection_timeout: Duration) -> Self {
+        Self { connection_timeout, ..self }
+    }
+
+    pub fn build(self) -> Result<(Session, tokio::sync::watch::Receiver<SessionState>), Error> {
+        let session = match self.session {
+            Some(session) => {
+                if session.is_readonly() {
+                    return Err(Error::new_other(
+                        format!("can't reestablish readonly and hence local session {}", session.id()),
+                        None,
+                    ));
+                }
+                Span::current().record("session", display(session.id()));
+                session
+            },
+            None => SessionInfo::new(SessionId(0), Vec::with_capacity(PASSWORD_LEN)),
+        };
+        if self.session_timeout < Duration::ZERO {
+            return Err(Error::BadArguments(&"session timeout must not be negative"));
+        } else if self.connection_timeout < Duration::ZERO {
+            return Err(Error::BadArguments(&"connection timeout must not be negative"));
+        }
+        let tls_config = self.tls.unwrap_or_default().into_config()?;
         let (state_sender, state_receiver) = tokio::sync::watch::channel(SessionState::Disconnected);
         let now = Instant::now();
         let (watch_manager, unwatch_receiver) = WatchManager::new();
         let mut session = Session {
-            readonly,
-            detached,
+            readonly: self.readonly,
+            detached: self.detached,
 
-            configured_connection_timeout: connection_timeout,
+            configured_connection_timeout: self.connection_timeout,
 
             last_zxid: session.last_zxid,
             last_recv: now,
@@ -122,22 +144,60 @@ impl Session {
             session_expired_timeout: Duration::ZERO,
             connector: Connector::new(tls_config),
             #[cfg(any(feature = "sasl-digest-md5", feature = "sasl-gssapi"))]
-            sasl_options,
+            sasl_options: self.sasl,
             #[cfg(any(feature = "sasl-digest-md5", feature = "sasl-gssapi"))]
             sasl_session: None,
 
             session,
-            session_timeout,
+            session_timeout: self.session_timeout,
             session_state: SessionState::Disconnected,
 
-            authes: authes.iter().map(|auth| MarshalledRequest::new(OpCode::Auth, auth)).collect(),
+            authes: self.authes,
             state_sender,
             watch_manager,
             unwatch_receiver: Some(unwatch_receiver),
         };
-        let timeout = if session_timeout.is_zero() { DEFAULT_SESSION_TIMEOUT } else { session_timeout };
+        let timeout = if self.session_timeout.is_zero() { DEFAULT_SESSION_TIMEOUT } else { self.session_timeout };
         session.reset_timeout(timeout);
-        (session, state_receiver)
+        Ok((session, state_receiver))
+    }
+}
+
+pub struct Session {
+    readonly: bool,
+    detached: bool,
+
+    connector: Connector,
+
+    configured_connection_timeout: Duration,
+
+    last_zxid: i64,
+    last_recv: Instant,
+    last_send: Instant,
+    last_ping: Option<Instant>,
+    tick_timeout: Duration,
+    ping_timeout: Duration,
+    session_expired_timeout: Duration,
+
+    pub session: SessionInfo,
+    session_state: SessionState,
+    pub session_timeout: Duration,
+
+    #[cfg(any(feature = "sasl-digest-md5", feature = "sasl-gssapi"))]
+    sasl_options: Option<SaslOptions>,
+    #[cfg(any(feature = "sasl-digest-md5", feature = "sasl-gssapi"))]
+    sasl_session: Option<SaslSession>,
+
+    pub authes: Vec<MarshalledRequest>,
+    state_sender: tokio::sync::watch::Sender<SessionState>,
+
+    watch_manager: WatchManager,
+    unwatch_receiver: Option<mpsc::UnboundedReceiver<(WatcherId, StateResponser)>>,
+}
+
+impl Session {
+    pub fn builder() -> Builder {
+        Builder::default()
     }
 
     fn is_readonly_allowed(&self) -> bool {