feat!: add ClientBuilder::connect_lazy

Author: paolobarbolini

watermelon/src/client/builder.rs

@@ -209,6 +209,17 @@ impl ClientBuilder {
     pub async fn connect(self, addr: ServerAddr) -> Result<Client, ConnectHandlerError> {
         Client::connect(addr, self).await
     }
+
+    /// Creates a new [`Client`], connecting to the given address in the background.
+    ///
+    /// This method is for applications that want to construct a client without
+    /// waiting for the server connection to be established. This may result in
+    /// the connection never succeeding, despite the continuous attempts
+    /// made by the client, and the [`Client`] buffer filling up with requests
+    /// and blocking all subsequent commands forever.
+    pub fn connect_lazy(self, addr: ServerAddr) -> Client {
+        Client::connect_lazy(addr, self)
+    }
 }
 
 impl Default for ClientBuilder {

watermelon/src/client/mod.rs

@@ -144,34 +144,13 @@ impl Client {
                     HandlerOutput::ServerError
                     | HandlerOutput::Disconnected
                     | HandlerOutput::UnexpectedState => {
-                        let mut recycle = handle.recycle().await;
-
-                        let mut delay = MIN_RECONNECT_DELAY;
-
-                        loop {
-                            select! {
-                                biased;
-                                () = recycle.wait_shutdown() => {
-                                    return;
-                                },
-                                () = sleep(delay) => {},
-                            }
-
-                            match Handler::connect(&addr, &builder, recycle).await {
-                                Ok(Some(new_handle)) => {
-                                    handle = new_handle;
-                                    break;
-                                }
-                                Ok(None) => {
-                                    // shutdown
-                                    return;
-                                }
-                                Err((_err, prev_recycle)) => {
-                                    recycle = prev_recycle;
-                                    delay *= 2;
-                                    delay = delay.min(MAX_RECONNECT_DELAY);
-                                }
-                            }
+                        let recycle = handle.recycle().await;
+                        if let Some(new_handle) =
+                            connect(&addr, &builder, recycle, MIN_RECONNECT_DELAY).await
+                        {
+                            handle = new_handle;
+                        } else {
+                            break;
                         }
                     }
                     HandlerOutput::Closed => break,
@@ -194,6 +173,63 @@ impl Client {
         })
     }
 
+    pub(crate) fn connect_lazy(addr: ServerAddr, builder: ClientBuilder) -> Self {
+        let (sender, receiver) = mpsc::channel(CLIENT_OP_CHANNEL_SIZE);
+
+        let (shutdown_sender, shutdown_receiver) = mpsc::channel(1);
+
+        let quick_info = Arc::new(RawQuickInfo::new());
+
+        let recycle = RecycledHandler::new(
+            receiver,
+            Arc::clone(&quick_info),
+            &builder,
+            shutdown_receiver,
+        );
+        let info = Arc::clone(recycle.info());
+        let multiplexed_subscription_prefix = recycle.multiplexed_subscription_prefix().clone();
+        let inbox_prefix = builder.inbox_prefix.clone();
+        let default_response_timeout = builder.default_response_timeout;
+
+        let handler = tokio::spawn(async move {
+            let Some(mut handle) = connect(&addr, &builder, recycle, Duration::ZERO).await else {
+                return;
+            };
+            #[expect(clippy::while_let_loop)]
+            loop {
+                match (&mut handle).await {
+                    HandlerOutput::ServerError
+                    | HandlerOutput::Disconnected
+                    | HandlerOutput::UnexpectedState => {
+                        let recycle = handle.recycle().await;
+                        if let Some(new_handle) =
+                            connect(&addr, &builder, recycle, MIN_RECONNECT_DELAY).await
+                        {
+                            handle = new_handle;
+                        } else {
+                            break;
+                        }
+                    }
+                    HandlerOutput::Closed => break,
+                }
+            }
+        });
+
+        Self {
+            inner: Arc::new(ClientInner {
+                info,
+                sender,
+                quick_info,
+                multiplexed_subscription_prefix,
+                next_subscription_id: AtomicU64::new(u64::from(MULTIPLEXED_SUBSCRIPTION_ID) + 1),
+                inbox_prefix,
+                default_response_timeout,
+                handler,
+                shutdown_sender,
+            }),
+        }
+    }
+
     #[cfg(test)]
     pub(crate) fn test(client_to_handler_chan_size: usize) -> (Self, TestHandler) {
         let builder = Self::builder();
@@ -417,13 +453,13 @@ impl Client {
     ///
     /// Consider calling [`Client::quick_info`] if you only need
     /// information about Lame Duck Mode.
+    ///
+    /// Returns `None` if the client was created using
+    /// [`ClientBuilder::connect_lazy`] AND the connection
+    /// has not been successfully established yet.
     #[must_use]
-    #[expect(
-        clippy::missing_panics_doc,
-        reason = "we don't expect the panic to ever happen"
-    )]
-    pub fn server_info(&self) -> Arc<ServerInfo> {
-        self.inner.info.load_full().expect("never connected")
+    pub fn server_info(&self) -> Option<Arc<ServerInfo>> {
+        self.inner.info.load_full()
     }
 
     /// Get information about the client
@@ -535,3 +571,41 @@ pub(crate) fn create_inbox_subject(prefix: &Subject) -> Subject {
 
     Subject::from_dangerous_value(subject.into())
 }
+
+async fn connect(
+    addr: &ServerAddr,
+    builder: &ClientBuilder,
+    mut recycle: RecycledHandler,
+    initial_delay: Duration,
+) -> Option<Handler> {
+    let mut delay = initial_delay;
+
+    loop {
+        select! {
+            biased;
+            () = recycle.wait_shutdown() => {
+                return None;
+            },
+            () = sleep(delay) => {},
+        }
+
+        match Handler::connect(addr, builder, recycle).await {
+            Ok(Some(new_handle)) => {
+                return Some(new_handle);
+            }
+            Ok(None) => {
+                // shutdown
+                return None;
+            }
+            Err((_err, prev_recycle)) => {
+                recycle = prev_recycle;
+                if delay < MIN_RECONNECT_DELAY {
+                    delay = MIN_RECONNECT_DELAY;
+                } else {
+                    delay *= 2;
+                    delay = delay.min(MAX_RECONNECT_DELAY);
+                }
+            }
+        }
+    }
+}

watermelon/src/handler/mod.rs

@@ -691,6 +691,14 @@ impl RecycledHandler {
         }
     }
 
+    pub(crate) fn info(&self) -> &Arc<ArcSwapOption<ServerInfo>> {
+        &self.info
+    }
+
+    pub(crate) fn multiplexed_subscription_prefix(&self) -> &Subject {
+        &self.multiplexed_subscription_prefix
+    }
+
     pub(crate) async fn wait_shutdown(&mut self) {
         let _ = self.shutdown_recv.recv().await;
     }