diff --git a/lib/src/client/builder.rs b/lib/src/client/builder.rs index e802f15c1..d0d8f71f4 100644 --- a/lib/src/client/builder.rs +++ b/lib/src/client/builder.rs @@ -208,6 +208,13 @@ impl ClientBuilder { self } + /// Sets whether the client should transfer (or recreate if transfers are + /// not supported) subscriptions to the new session when reconnecting. + pub fn transfer_on_reconnect(mut self, transfer_on_reconnect: bool) -> Self { + self.config.transfer_on_reconnect = transfer_on_reconnect; + self + } + /// Initial time between retries when backing off on session reconnects. pub fn session_retry_initial(mut self, session_retry_initial: Duration) -> Self { self.config.session_retry_initial = session_retry_initial; diff --git a/lib/src/client/config.rs b/lib/src/client/config.rs index 33dccc03f..ab1bdb34d 100644 --- a/lib/src/client/config.rs +++ b/lib/src/client/config.rs @@ -82,17 +82,18 @@ impl ClientUserToken { ); valid = false; } - } else { - if self.cert_path.is_none() && self.private_key_path.is_none() { - error!( - "User token {} fails to provide a password or certificate info.", - self.user - ); - valid = false; - } else if self.cert_path.is_none() || self.private_key_path.is_none() { - error!("User token {} fails to provide both a certificate path and a private key path.", self.user); - valid = false; - } + } else if self.cert_path.is_none() && self.private_key_path.is_none() { + error!( + "User token {} fails to provide a password or certificate info.", + self.user + ); + valid = false; + } else if self.cert_path.is_none() || self.private_key_path.is_none() { + error!( + "User token {} fails to provide both a certificate path and a private key path.", + self.user + ); + valid = false; } valid } @@ -202,6 +203,9 @@ pub struct ClientConfig { /// Maximum number of times to attempt to reconnect to the server before giving up. /// -1 retries forever pub(crate) session_retry_limit: i32, + /// Transfer (or recreate if transfers are not supported) subscriptions to + /// the new session when reconnecting. + pub(crate) transfer_on_reconnect: bool, /// Initial delay for exponential backoff when reconnecting to the server. pub(crate) session_retry_initial: Duration, @@ -215,8 +219,9 @@ pub struct ClientConfig { /// Timeout for publish requests, separate from normal timeout since /// subscriptions are often more time sensitive. pub(crate) publish_timeout: Duration, - /// Minimum publish interval. Setting this higher will make sure that subscriptions - /// publish together, which may reduce the number of publish requests if you have a lot of subscriptions. + /// Minimum publish interval. Setting this higher will make sure that + /// subscriptions publish together, which may reduce the number of publish + /// requests if you have a lot of subscriptions. pub(crate) min_publish_interval: Duration, /// Maximum number of inflight publish requests before further requests are skipped. pub(crate) max_inflight_publish: usize, @@ -354,6 +359,7 @@ impl ClientConfig { session_retry_limit: SessionRetryPolicy::DEFAULT_RETRY_LIMIT as i32, session_retry_initial: Duration::from_secs(1), session_retry_max: Duration::from_secs(30), + transfer_on_reconnect: true, keep_alive_interval: Duration::from_secs(10), request_timeout: Duration::from_secs(60), min_publish_interval: Duration::from_secs(1), diff --git a/lib/src/client/session/connect.rs b/lib/src/client/session/connect.rs index 1ea509840..b23b49ba9 100644 --- a/lib/src/client/session/connect.rs +++ b/lib/src/client/session/connect.rs @@ -102,7 +102,9 @@ impl SessionConnector { } }; - self.inner.transfer_subscriptions_from_old_session().await; + if self.inner.transfer_on_reconnect { + self.inner.transfer_subscriptions_from_old_session().await; + } Ok(reconnect) } diff --git a/lib/src/client/session/session.rs b/lib/src/client/session/session.rs index 9402ac5aa..d83f01c9b 100644 --- a/lib/src/client/session/session.rs +++ b/lib/src/client/session/session.rs @@ -51,10 +51,11 @@ pub struct Session { pub(super) application_description: ApplicationDescription, pub(super) request_timeout: Duration, pub(super) publish_timeout: Duration, - pub(super) recreate_monitored_items_chunk: usize, pub(super) session_timeout: f64, pub(super) max_inflight_publish: usize, + pub(super) recreate_monitored_items_chunk: usize, pub subscription_state: Mutex, + pub(super) transfer_on_reconnect: bool, pub(super) monitored_item_handle: AtomicHandle, pub(super) trigger_publish_tx: tokio::sync::watch::Sender, } @@ -106,6 +107,7 @@ impl Session { max_inflight_publish: config.max_inflight_publish, recreate_monitored_items_chunk: config.performance.recreate_monitored_items_chunk, subscription_state: Mutex::new(SubscriptionState::new(config.min_publish_interval)), + transfer_on_reconnect: config.transfer_on_reconnect, monitored_item_handle: AtomicHandle::new(1000), trigger_publish_tx, }); diff --git a/samples/client.conf b/samples/client.conf index 8fb8f9b07..597f457b4 100644 --- a/samples/client.conf +++ b/samples/client.conf @@ -46,6 +46,7 @@ decoding_options: max_byte_string_length: 65535 max_array_length: 1000 session_retry_limit: 10 +transfer_on_reconnect: true session_retry_initial: secs: 1 nanos: 0