Reconnect control with callback WStompConfig::on_reconnect

.github/workflows/pipeline.yaml

@@ -1,36 +1,34 @@
 name: Build
-on: [push]
+on: [push, pull_request]
 
 jobs:
   nightly-clippy:
     name: Nightly Clippy
     runs-on: ubuntu-latest
     steps:
       - name: Checkout code
-        uses: actions/checkout@v2
+        uses: actions/checkout@v6
 
       - name: Cache Cargo registry
-        uses: actions/cache@v3
+        uses: actions/cache@v5
         with:
           path: ~/.cargo/registry
           key: ${{ runner.os }}-cargo-registry-${{ hashFiles('**/Cargo.lock') }}
           restore-keys: |
             ${{ runner.os }}-cargo-registry-
 
       - name: Cache Cargo build
-        uses: actions/cache@v3
+        uses: actions/cache@v5
         with:
           path: target
           key: ${{ runner.os }}-cargo-build-${{ hashFiles('**/Cargo.lock') }}
           restore-keys: |
             ${{ runner.os }}-cargo-build-
 
       - name: Set up Rust Nightly
-        uses: actions-rs/toolchain@v1
+        uses: dtolnay/rust-toolchain@nightly
         with:
-          toolchain: nightly
           components: clippy
-          override: true
 
       - name: Run Clippy
         run: cargo clippy -- -D warnings
@@ -40,29 +38,26 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Checkout code
-        uses: actions/checkout@v2
+        uses: actions/checkout@v6
 
       - name: Cache Cargo registry
-        uses: actions/cache@v3
+        uses: actions/cache@v5
         with:
           path: ~/.cargo/registry
           key: ${{ runner.os }}-cargo-registry-${{ hashFiles('**/Cargo.lock') }}
           restore-keys: |
             ${{ runner.os }}-cargo-registry-
 
       - name: Cache Cargo build
-        uses: actions/cache@v3
+        uses: actions/cache@v5
         with:
           path: target
           key: ${{ runner.os }}-cargo-build-${{ hashFiles('**/Cargo.lock') }}
           restore-keys: |
             ${{ runner.os }}-cargo-build-
 
       - name: Set up Rust Nightly
-        uses: actions-rs/toolchain@v1
-        with:
-          toolchain: nightly
-          override: true
+        uses: dtolnay/rust-toolchain@nightly
 
       - name: Run Cargo Tests
         run: cargo test --all-features
@@ -72,30 +67,28 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Checkout code
-        uses: actions/checkout@v2
+        uses: actions/checkout@v6
 
       - name: Cache Cargo registry
-        uses: actions/cache@v3
+        uses: actions/cache@v5
         with:
           path: ~/.cargo/registry
           key: ${{ runner.os }}-cargo-registry-${{ hashFiles('**/Cargo.lock') }}
           restore-keys: |
             ${{ runner.os }}-cargo-registry-
 
       - name: Cache Cargo build
-        uses: actions/cache@v3
+        uses: actions/cache@v5
         with:
           path: target
           key: ${{ runner.os }}-cargo-build-${{ hashFiles('**/Cargo.lock') }}
           restore-keys: |
             ${{ runner.os }}-cargo-build-
 
       - name: Set up Rust Stable
-        uses: actions-rs/toolchain@v1
+        uses: dtolnay/rust-toolchain@stable
         with:
-          toolchain: stable
           components: clippy
-          override: true
 
       - name: Run Clippy
         run: cargo clippy -- -D warnings
@@ -105,29 +98,26 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Checkout code
-        uses: actions/checkout@v2
+        uses: actions/checkout@v6
 
       - name: Cache Cargo registry
-        uses: actions/cache@v3
+        uses: actions/cache@v5
         with:
           path: ~/.cargo/registry
           key: ${{ runner.os }}-cargo-registry-${{ hashFiles('**/Cargo.lock') }}
           restore-keys: |
             ${{ runner.os }}-cargo-registry-
 
       - name: Cache Cargo build
-        uses: actions/cache@v3
+        uses: actions/cache@v5
         with:
           path: target
           key: ${{ runner.os }}-cargo-build-${{ hashFiles('**/Cargo.lock') }}
           restore-keys: |
             ${{ runner.os }}-cargo-build-
 
       - name: Set up Rust Stable
-        uses: actions-rs/toolchain@v1
-        with:
-          toolchain: stable
-          override: true
+        uses: dtolnay/rust-toolchain@stable
 
       - name: Run Cargo Tests
         run: cargo test --all-features

CHANGES.md

@@ -0,0 +1,21 @@
+<!-- markdownlint-configure-file { "no-duplicate-heading": { "siblings_only": true } } -->
+
+<!-- markdownlint-disable-next-line first-line-h1 -->
+## 0.2.0 - Unreleased
+
+### Added
+
+* Reconnect with callback installed via `WStompConfig::on_reconnect`.
+  The reconnect loop is then spawned by `WStompConfig::build_and_connect`,
+  which in the reconnecting state returns a `WStompReconnectHandle`
+  instead of a `WStompClient`.
+
+## 0.1.0 - 2025-11-18
+
+* Initial release
+* Connects to STOMP servers over WebSocket using [`awc`](https://crates.io/crates/awc).
+* Handles all STOMP protocol encoding and decoding via [`async-stomp`](https://crates.io/crates/async-stomp).
+* Manages WebSocket ping/pong heartbeats automatically in a background task.
+* Provides a simple `tokio::mpsc` channel-based API ([`WStompClient`]) for sending and receiving STOMP frames.
+* Connection helpers for various authentication methods
+* Optional `rustls` feature for SSL connections, with helpers that force HTTP/1.1 for compatibility with servers like SockJS.

README.md

@@ -21,6 +21,7 @@ This crate provides a simple client to connect to a STOMP-enabled WebSocket serv
   * [`connect`]: Anonymous connection.
   * [`connect_with_pass`]: Login and passcode authentication.
   * [`connect_with_token`]: Authentication using an authorization token header.
+  * [`WStompConfig::auth_token_fn`]: Dynamic token provider — refreshes credentials on every (re)connect.
 * Optional `rustls` feature for SSL connections, with helpers that force HTTP/1.1 for compatibility with servers like SockJS.
 
 ## Installation
@@ -42,7 +43,7 @@ wstomp = { version = "0.1.0", features = ["rustls"] }
 
 ## Usage
 
-Here is a basic example of connecting, subscribing to a topic, and receiving messages.
+Here is a basic example of connecting, subscribing to a topic and receiving messages.
 
 ```rust,no_run
 use wstomp::{
@@ -182,31 +183,42 @@ async fn main() {
 
 ### Auto-reconnect
 
-Use [`WStompConfig::build_and_connect_with_reconnection_cb`] method to automatically perform a full reconnect upon errors.
+Install a callback via [`WStompConfig::on_reconnect`] and call `build_and_connect` to automatically perform a full reconnect upon errors. The callback receives the result of each connection attempt and returns a [`ReconnectControl`] value that drives the loop.
 
 ```rust,no_run
-use wstomp::{WStompClient, WStompConfig, WStompConnectError};
+use wstomp::{ReconnectControl, WStompClient, WStompConfig, WStompConnectError, WStompReconnectHandle};
 
 #[actix_rt::main]
 async fn main() {
     let url = "wss://secure-server.com/ws";
     let session_token = "session_token";
 
-    let cb = {
-        move |wstomp_client_res: Result<WStompClient, WStompConnectError>| {
-            async move {
-                // Unwrap wstomp client here or react to an error.
-                // Upon an error you can return from the callback to make wstomp library a re-connection attempt
+    let cb = move |wstomp_client_res: Result<WStompClient, WStompConnectError>| {
+        async move {
+            match wstomp_client_res {
+                Ok(client) => {
+                    // Use the client; when the session ends, Continue triggers a reconnect.
+                    ReconnectControl::Continue
+                }
+                Err(err) => {
+                    eprintln!("Connection failed: {}", err);
+                    // Return Stop to give up, or Continue to keep retrying with backoff.
+                    ReconnectControl::Continue
+                }
             }
         }
     };
 
-    let res = WStompConfig::new(url)
+    // Returns immediately with a handle; the reconnection loop runs in the background.
+    let _handle: WStompReconnectHandle = WStompConfig::new(url)
         .ssl()
         .auth_token(session_token)
-        .build_and_connect_with_reconnection_cb(cb);
+        .on_reconnect(cb)
+        .build_and_connect()
+        .expect("Failed to start reconnection loop");
 
-    // ... do different stuff here, but don't exit immediately as this will terminate wstomp loop.
+    // Keep _handle alive for as long as the loop should run.
+    // Dropping it (or calling _handle.abort()) stops the loop.
 }
 ```
 

src/client.rs

@@ -33,7 +33,8 @@ impl WStompClient {
     /// It also manages websocket ping-pong heartbeat.
     ///
     /// NOTE: This method does not perform automatic reconnection.
-    /// Use [WStompConfig::build_and_connect_with_reconnection_cb] to auto-reconnect.
+    /// Install a callback via [WStompConfig::on_reconnect] and call
+    /// `build_and_connect` to auto-reconnect.
     pub fn from_framed(ws_framed: Framed<BoxedSocket, Codec>) -> Self {
         // Channel for you to send STOMP frames to the handler task
         let (app_tx, app_rx) = mpsc::channel::<Message<ToServer>>(100);

src/config.rs

@@ -1,17 +1,36 @@
-use std::sync::Arc;
+use std::{future::Future, pin::Pin, sync::Arc};
 
 #[cfg(feature = "rustls")]
 use tokio_rustls::rustls::pki_types::{CertificateDer, PrivateKeyDer};
 
-pub struct WStompConfig<U> {
-    url: U,
-    opts: WStompConfigOpts,
+/// Closure type for dynamic auth token retrieval. Invoked on every
+/// (re)connection attempt so the token can be refreshed out-of-band by the
+/// application (e.g. via a background token-refresh task) without recreating
+/// the wstomp client.
+pub type AuthTokenFn =
+    Arc<dyn Fn() -> Pin<Box<dyn Future<Output = Option<String>> + Send>> + Send + Sync>;
+
+/// Marker for a [`WStompConfig`] without a reconnection callback. In this
+/// state [`WStompConfig::build_and_connect`] is a one-shot async method that
+/// resolves to a [`crate::WStompClient`].
+pub struct NoReconnect;
+
+/// Marker for a [`WStompConfig`] with a reconnection callback installed via
+/// [`WStompConfig::on_reconnect`]. In this state
+/// [`WStompConfig::build_and_connect`] spawns the reconnect loop and returns
+/// a [`crate::WStompReconnectHandle`].
+pub struct Reconnecting<F>(pub(crate) F);
+
+pub struct WStompConfig<U, R = NoReconnect> {
+    pub(crate) url: U,
+    pub(crate) opts: WStompConfigOpts,
+    pub(crate) reconnect: R,
 }
 
 #[derive(Clone)]
 pub struct WStompConfigOpts {
     pub ssl: bool,
-    pub auth_token: Option<String>,
+    pub auth_token: Option<AuthTokenFn>,
     pub login: Option<String>,
     pub passcode: Option<String>,
     #[cfg(feature = "rustls")]
@@ -54,14 +73,17 @@ impl Default for WStompConfigOpts {
     }
 }
 
-impl<U> WStompConfig<U> {
+impl<U> WStompConfig<U, NoReconnect> {
     pub fn new(url: U) -> Self {
         Self {
             url,
             opts: WStompConfigOpts::default(),
+            reconnect: NoReconnect,
         }
     }
+}
 
+impl<U, R> WStompConfig<U, R> {
     /// Get url to which this config is assigned to use.
     pub fn get_url(&self) -> &U {
         &self.url
@@ -88,9 +110,29 @@ impl<U> WStompConfig<U> {
         self
     }
 
-    /// Sets the authentication token for the connection.
+    /// Sets a static authentication token for the connection. The token is
+    /// fixed for the lifetime of the client — use [Self::auth_token_fn] when
+    /// the token needs to be refreshed between reconnects.
     pub fn auth_token(mut self, auth_token: impl Into<String>) -> Self {
-        self.opts.auth_token = Some(auth_token.into());
+        let token = auth_token.into();
+        self.opts.auth_token = Some(Arc::new(move || {
+            let token = token.clone();
+            Box::pin(async move { Some(token) })
+        }));
+        self
+    }
+
+    /// Sets a dynamic authentication token provider. `f` is invoked on every
+    /// (re)connection attempt, so returning a fresh token allows the app to
+    /// refresh credentials without tearing down the wstomp client.
+    ///
+    /// Returning `None` from `f` omits the `Authorization` header entirely.
+    pub fn auth_token_fn<F, Fut>(mut self, f: F) -> Self
+    where
+        F: Fn() -> Fut + Send + Sync + 'static,
+        Fut: Future<Output = Option<String>> + Send + 'static,
+    {
+        self.opts.auth_token = Some(Arc::new(move || Box::pin(f())));
         self
     }
 
@@ -145,35 +187,36 @@ impl<U> WStompConfig<U> {
         self
     }
 
-    /// If [Self::build_and_connect_with_reconnection_cb] method is used,
-    /// sets the initial retry interval in seconds.
+    /// When [`Self::on_reconnect`] is used to install a reconnection
+    /// callback, sets the initial retry interval in seconds.
     ///
     /// Example: Start retrying after 3 seconds.
     pub fn retry_initial_interval(mut self, seconds: u64) -> Self {
         self.opts.retry_initial_interval = seconds;
         self
     }
 
-    /// If [Self::build_and_connect_with_reconnection_cb] method is used,
-    /// sets the maximum retry interval in seconds.
+    /// When [`Self::on_reconnect`] is used to install a reconnection
+    /// callback, sets the maximum retry interval in seconds.
     ///
     /// Example: Cap the wait time at 30 seconds.
     pub fn retry_max_interval(mut self, seconds: u64) -> Self {
         self.opts.retry_max_interval = seconds;
         self
     }
 
-    /// If [Self::build_and_connect_with_reconnection_cb] method is used,
-    /// sets the multiplier for the backoff.
+    /// When [`Self::on_reconnect`] is used to install a reconnection
+    /// callback, sets the multiplier for the backoff.
     ///
     /// Example: 2.0 doubles the wait time after every failure.
     pub fn retry_multiplier(mut self, multiplier: f64) -> Self {
         self.opts.retry_multiplier = multiplier;
         self
     }
 
-    /// If [Self::build_and_connect_with_reconnection_cb] method is used,
-    /// sets a maximum total time to try reconnecting before giving up.
+    /// When [`Self::on_reconnect`] is used to install a reconnection
+    /// callback, sets a maximum total time to try reconnecting before giving
+    /// up.
     ///
     /// Defaults to no limit if method not invoked.
     pub fn retry_max_elapsed_time(mut self, seconds: u64) -> Self {