This is an automated email from the ASF dual-hosted git repository.
piotr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 1dffd350 feat(sdk): Support `from_connection_string()` for
`QuicClient` and `HttpClient` (#1803)
1dffd350 is described below
commit 1dffd3503e3760d140ef63250c697185c2a8a075
Author: I-Hsin Cheng <[email protected]>
AuthorDate: Thu May 29 21:19:53 2025 +0800
feat(sdk): Support `from_connection_string()` for `QuicClient` and
`HttpClient` (#1803)
## Summary
This PR aims to provide the following things:
* Implement `from_connection_string()` for `QuicClient` and `HttpClient`
* Support PAT for `from_connection_string()`
* Implement Unit Tests for the usage of `from_connection_string()` from
different protocol client
* Refactor `ConnectionString` to make it type-sensitive so different
clients only have to implement their own `*ConnectionOptions` when
trying to support connection string features.
* Move `QuicClientConfig` , `HttpClientConfig` and their related structs to
iggy_common
## Discussion
QUIC requires to know client address, which connection string doesn't
provide.
Maybe we should try to use some helpers to get it, or we should just
neglect client address when using QUIC with connection string ?
## Related Issue
#1769
---------
Signed-off-by: I Hsin Cheng <[email protected]>
---
core/common/src/lib.rs | 6 +-
.../configuration/auth_config/connection_string.rs | 349 ++++++++++++++-------
.../auth_config/connection_string_options.rs | 74 +----
.../{config.rs => http_client_config.rs} | 38 +--
.../http_config/http_client_config_builder.rs} | 18 +-
.../http_config/http_connection_string_options.rs | 87 +++++
.../src/types/configuration/http_config/mod.rs | 4 +-
.../src/types/configuration/quick_config/mod.rs | 1 +
.../quick_config/quic_client_config.rs | 27 +-
.../quick_config/quic_connection_string_options.rs | 281 +++++++++++++++++
.../src/types/configuration/tcp_config/mod.rs | 1 +
.../configuration/tcp_config/tcp_client_config.rs | 6 +-
.../tcp_config/tcp_connection_string_options.rs | 173 ++++++++++
core/integration/src/http_client.rs | 3 +-
core/integration/src/quic_client.rs | 3 +-
core/sdk/src/client_provider.rs | 9 +-
core/sdk/src/clients/client.rs | 201 +++++++++++-
core/sdk/src/clients/client_builder.rs | 210 ++++++++++++-
core/sdk/src/http/http_client.rs | 222 ++++++++++++-
core/sdk/src/http/mod.rs | 1 -
core/sdk/src/prelude.rs | 17 +-
core/sdk/src/quic/mod.rs | 1 -
core/sdk/src/quic/quick_client.rs | 314 +++++++++++++++++-
core/sdk/src/quic/quick_config.rs | 232 --------------
core/sdk/src/tcp/tcp_client.rs | 285 ++++++++++++++++-
25 files changed, 2063 insertions(+), 500 deletions(-)
diff --git a/core/common/src/lib.rs b/core/common/src/lib.rs
index bad2ca67..bd83fa66 100644
--- a/core/common/src/lib.rs
+++ b/core/common/src/lib.rs
@@ -52,13 +52,17 @@ pub use types::configuration::auth_config::auto_login::*;
pub use types::configuration::auth_config::connection_string::*;
pub use types::configuration::auth_config::connection_string_options::*;
pub use types::configuration::auth_config::credentials::*;
-pub use types::configuration::http_config::config::*;
+pub use types::configuration::http_config::http_client_config::*;
+pub use types::configuration::http_config::http_client_config_builder::*;
+pub use types::configuration::http_config::http_connection_string_options::*;
pub use types::configuration::quick_config::quic_client_config::*;
pub use types::configuration::quick_config::quic_client_config_builder::*;
pub use types::configuration::quick_config::quic_client_reconnection_config::*;
+pub use types::configuration::quick_config::quic_connection_string_options::*;
pub use types::configuration::tcp_config::tcp_client_config::*;
pub use types::configuration::tcp_config::tcp_client_config_builder::*;
pub use types::configuration::tcp_config::tcp_client_reconnection_config::*;
+pub use types::configuration::tcp_config::tcp_connection_string_options::*;
pub use types::confirmation::*;
pub use types::consumer::consumer_group::*;
pub use types::consumer::consumer_kind::*;
diff --git
a/core/common/src/types/configuration/auth_config/connection_string.rs
b/core/common/src/types/configuration/auth_config/connection_string.rs
index 3f53194d..b8854efa 100644
--- a/core/common/src/types/configuration/auth_config/connection_string.rs
+++ b/core/common/src/types/configuration/auth_config/connection_string.rs
@@ -16,22 +16,21 @@
* under the License.
*/
-use crate::{
- AutoLogin, ConnectionStringOptions, Credentials, IggyDuration, IggyError,
- TcpClientReconnectionConfig,
-};
+use crate::{AutoLogin, ConnectionStringOptions, Credentials, IggyError};
use std::str::FromStr;
+use strum::{Display, EnumString, IntoStaticStr};
-const CONNECTION_STRING_PREFIX: &str = "iggy://";
+const DEFAULT_CONNECTION_STRING_PREFIX: &str = "iggy://";
+const CONNECTION_STRING_PREFIX: &str = "iggy+";
#[derive(Debug)]
-pub struct ConnectionString {
+pub struct ConnectionString<T: ConnectionStringOptions + Default> {
server_address: String,
auto_login: AutoLogin,
- options: ConnectionStringOptions,
+ options: T,
}
-impl ConnectionString {
+impl<T: ConnectionStringOptions + Default> ConnectionString<T> {
pub fn server_address(&self) -> &str {
&self.server_address
}
@@ -40,36 +39,32 @@ impl ConnectionString {
&self.auto_login
}
- pub fn options(&self) -> &ConnectionStringOptions {
+ pub fn options(&self) -> &T {
&self.options
}
-}
-impl ConnectionString {
pub fn new(connection_string: &str) -> Result<Self, IggyError> {
- if connection_string.is_empty() {
- return Err(IggyError::InvalidConnectionString);
- }
-
- if !connection_string.starts_with(CONNECTION_STRING_PREFIX) {
- return Err(IggyError::InvalidConnectionString);
- }
-
- let connection_string =
connection_string.replace(CONNECTION_STRING_PREFIX, "");
+ let connection_string =
connection_string.split("://").collect::<Vec<&str>>()[1];
let parts = connection_string.split('@').collect::<Vec<&str>>();
+ let mut username = "";
+ let mut password = "";
+ let mut pat_token = "";
if parts.len() != 2 {
return Err(IggyError::InvalidConnectionString);
}
let credentials = parts[0].split(':').collect::<Vec<&str>>();
- if credentials.len() != 2 {
- return Err(IggyError::InvalidConnectionString);
- }
+ if credentials.len() == 1 {
+ pat_token = credentials[0];
+ } else if credentials.len() == 2 {
+ username = credentials[0];
+ password = credentials[1];
- let username = credentials[0];
- let password = credentials[1];
- if username.is_empty() || password.is_empty() {
+ if username.is_empty() || password.is_empty() {
+ return Err(IggyError::InvalidConnectionString);
+ }
+ } else {
return Err(IggyError::InvalidConnectionString);
}
@@ -83,7 +78,7 @@ impl ConnectionString {
return Err(IggyError::InvalidConnectionString);
}
- if !server_address.contains(':') {
+ if !server_address.contains(':') || server_address.starts_with(':') {
return Err(IggyError::InvalidConnectionString);
}
@@ -98,9 +93,19 @@ impl ConnectionString {
let connection_string_options;
if let Some(options) = server_and_options.get(1) {
- connection_string_options =
ConnectionString::parse_options(options)?;
+ connection_string_options = T::parse_options(options)?;
} else {
- connection_string_options = ConnectionStringOptions::default();
+ connection_string_options = T::default();
+ }
+
+ if credentials.len() == 1 {
+ return Ok(ConnectionString {
+ server_address: server_address.to_owned(),
+ auto_login:
AutoLogin::Enabled(Credentials::PersonalAccessToken(
+ pat_token.to_owned(),
+ )),
+ options: connection_string_options,
+ });
}
Ok(ConnectionString {
@@ -112,89 +117,219 @@ impl ConnectionString {
options: connection_string_options,
})
}
+}
- fn parse_options(options: &str) -> Result<ConnectionStringOptions,
IggyError> {
- let options = options.split('&').collect::<Vec<&str>>();
- let mut tls_enabled = false;
- let mut tls_domain = "localhost".to_string();
- let mut tls_ca_file = None;
- let mut reconnection_retries = "unlimited".to_owned();
- let mut reconnection_interval = "1s".to_owned();
- let mut reestablish_after = "5s".to_owned();
- let mut heartbeat_interval = "5s".to_owned();
- let mut nodelay = false;
-
- for option in options {
- let option_parts = option.split('=').collect::<Vec<&str>>();
- if option_parts.len() != 2 {
- return Err(IggyError::InvalidConnectionString);
- }
- match option_parts[0] {
- "tls" => {
- tls_enabled = option_parts[1] == "true";
- }
- "tls_domain" => {
- tls_domain = option_parts[1].to_string();
- }
- "tls_ca_file" => {
- tls_ca_file = Some(option_parts[1].to_string());
- }
- "reconnection_retries" => {
- reconnection_retries = option_parts[1].to_string();
- }
- "reconnection_interval" => {
- reconnection_interval = option_parts[1].to_string();
- }
- "reestablish_after" => {
- reestablish_after = option_parts[1].to_string();
- }
- "heartbeat_interval" => {
- heartbeat_interval = option_parts[1].to_string();
- }
- "nodelay" => {
- nodelay = option_parts[1] == "true";
- }
- _ => {
- return Err(IggyError::InvalidConnectionString);
- }
- }
+impl<T: ConnectionStringOptions + Default> FromStr for ConnectionString<T> {
+ type Err = IggyError;
+ fn from_str(s: &str) -> Result<Self, Self::Err> {
+ ConnectionString::<T>::new(s)
+ }
+}
+
+/// ConnectionStringUtils is a utility struct for connection strings.
+pub struct ConnectionStringUtils;
+
+#[derive(Clone, Copy, Debug, Default, Display, PartialEq, EnumString,
IntoStaticStr)]
+#[strum(serialize_all = "snake_case")]
+pub enum TransportProtocol {
+ #[default]
+ #[strum(to_string = "tcp")]
+ Tcp,
+ #[strum(to_string = "quic")]
+ Quic,
+ #[strum(to_string = "http")]
+ Http,
+}
+
+impl TransportProtocol {
+ pub fn as_str(&self) -> &'static str {
+ self.into()
+ }
+}
+
+impl ConnectionStringUtils {
+ pub fn parse_protocol(connection_string: &str) ->
Result<TransportProtocol, IggyError> {
+ if connection_string.is_empty() {
+ return Err(IggyError::InvalidConnectionString);
+ }
+
+ if connection_string.starts_with(DEFAULT_CONNECTION_STRING_PREFIX) {
+ return Ok(TransportProtocol::Tcp);
+ }
+
+ if !connection_string.starts_with(CONNECTION_STRING_PREFIX) {
+ return Err(IggyError::InvalidConnectionString);
}
- let reconnection = TcpClientReconnectionConfig {
- enabled: true,
- max_retries: match reconnection_retries.as_str() {
- "unlimited" => None,
- _ => Some(
- reconnection_retries
- .parse()
- .map_err(|_| IggyError::InvalidNumberValue)?,
- ),
- },
- interval: IggyDuration::from_str(reconnection_interval.as_str())
- .map_err(|_| IggyError::InvalidConnectionString)?,
- reestablish_after:
IggyDuration::from_str(reestablish_after.as_str())
- .map_err(|_| IggyError::InvalidConnectionString)?,
- };
-
- let heartbeat_interval =
IggyDuration::from_str(heartbeat_interval.as_str())
- .map_err(|_| IggyError::InvalidConnectionString)?;
-
- let connection_string_options = ConnectionStringOptions::new(
- tls_enabled,
- tls_domain,
- tls_ca_file,
- reconnection,
- heartbeat_interval,
- nodelay,
- );
-
- Ok(connection_string_options)
+ let connection_string =
connection_string.replace(CONNECTION_STRING_PREFIX, "");
+
TransportProtocol::from_str(connection_string.split("://").collect::<Vec<&str>>()[0])
+ .map_err(|_| IggyError::InvalidConnectionString)
}
}
-impl FromStr for ConnectionString {
- type Err = IggyError;
- fn from_str(s: &str) -> Result<Self, Self::Err> {
- ConnectionString::new(s)
+/// Unit tests for ConnectionString and ConnectionStringUtils,
+/// common behavior which isn't type-specific will use
TcpConnectionStringOptions
+/// as default.
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::IggyDuration;
+ use crate::TcpConnectionStringOptions;
+
+ #[test]
+ fn should_fail_without_username() {
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "";
+ let password = "secret";
+ let value = format!(
+
"{DEFAULT_CONNECTION_STRING_PREFIX}{username}:{password}@{server_address}:{port}"
+ );
+ let connection_string =
ConnectionString::<TcpConnectionStringOptions>::new(&value);
+ assert!(connection_string.is_err());
+ }
+
+ #[test]
+ fn should_fail_without_password() {
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "";
+ let value = format!(
+
"{DEFAULT_CONNECTION_STRING_PREFIX}{username}:{password}@{server_address}:{port}"
+ );
+ let connection_string =
ConnectionString::<TcpConnectionStringOptions>::new(&value);
+ assert!(connection_string.is_err());
+ }
+
+ #[test]
+ fn should_fail_without_server_address() {
+ let server_address = "";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{DEFAULT_CONNECTION_STRING_PREFIX}{username}:{password}@{server_address}:{port}"
+ );
+ let connection_string =
ConnectionString::<TcpConnectionStringOptions>::new(&value);
+ assert!(connection_string.is_err());
+ }
+
+ #[test]
+ fn should_fail_without_port() {
+ let server_address = "127.0.0.1";
+ let port = "";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{DEFAULT_CONNECTION_STRING_PREFIX}{username}:{password}@{server_address}:{port}"
+ );
+ let connection_string =
ConnectionString::<TcpConnectionStringOptions>::new(&value);
+ assert!(connection_string.is_err());
+ }
+
+ #[test]
+ fn should_fail_with_invalid_options() {
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{DEFAULT_CONNECTION_STRING_PREFIX}{username}:{password}@{server_address}:{port}?invalid_option=invalid"
+ );
+ let connection_string =
ConnectionString::<TcpConnectionStringOptions>::new(&value);
+ assert!(connection_string.is_err());
+ }
+
+ #[test]
+ fn should_succeed_without_options() {
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{DEFAULT_CONNECTION_STRING_PREFIX}{username}:{password}@{server_address}:{port}"
+ );
+ let connection_string =
ConnectionString::<TcpConnectionStringOptions>::new(&value);
+ assert!(connection_string.is_ok());
+
+ let connection_string = connection_string.unwrap();
+ assert_eq!(
+ connection_string.server_address,
+ format!("{server_address}:{port}")
+ );
+ assert_eq!(
+ connection_string.auto_login,
+ AutoLogin::Enabled(Credentials::UsernamePassword(
+ username.to_string(),
+ password.to_string()
+ ))
+ );
+
+ assert!(connection_string.options.retries().is_none());
+ assert_eq!(
+ connection_string.options.heartbeat_interval(),
+ IggyDuration::from_str("5s").unwrap()
+ );
+ }
+
+ #[test]
+ fn should_succeed_with_options() {
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let retries = "3";
+ let heartbeat_interval = "10s";
+ let value = format!(
+
"{DEFAULT_CONNECTION_STRING_PREFIX}{username}:{password}@{server_address}:{port}?reconnection_retries={retries}&heartbeat_interval={heartbeat_interval}"
+ );
+ let connection_string =
ConnectionString::<TcpConnectionStringOptions>::new(&value);
+ assert!(connection_string.is_ok());
+
+ let connection_string = connection_string.unwrap();
+ assert_eq!(
+ connection_string.server_address,
+ format!("{server_address}:{port}")
+ );
+ assert_eq!(
+ connection_string.auto_login,
+ AutoLogin::Enabled(Credentials::UsernamePassword(
+ username.to_string(),
+ password.to_string()
+ ))
+ );
+
+ assert_eq!(connection_string.options.retries().unwrap(), 3);
+ assert_eq!(
+ connection_string.options.heartbeat_interval(),
+ IggyDuration::from_str("10s").unwrap()
+ );
+ }
+
+ #[test]
+ fn should_succeed_with_pat() {
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let pat = "iggypat-1234567890abcdef";
+ let value =
format!("{DEFAULT_CONNECTION_STRING_PREFIX}{pat}@{server_address}:{port}");
+ let connection_string =
ConnectionString::<TcpConnectionStringOptions>::new(&value);
+ assert!(connection_string.is_ok());
+
+ let connection_string = connection_string.unwrap();
+ assert_eq!(
+ connection_string.server_address,
+ format!("{server_address}:{port}")
+ );
+ assert_eq!(
+ connection_string.auto_login,
+
AutoLogin::Enabled(Credentials::PersonalAccessToken(pat.to_string()))
+ );
+
+ assert!(connection_string.options.retries().is_none());
+ assert_eq!(
+ connection_string.options.heartbeat_interval(),
+ IggyDuration::from_str("5s").unwrap()
+ );
}
}
diff --git
a/core/common/src/types/configuration/auth_config/connection_string_options.rs
b/core/common/src/types/configuration/auth_config/connection_string_options.rs
index b279fda7..b93f212f 100644
---
a/core/common/src/types/configuration/auth_config/connection_string_options.rs
+++
b/core/common/src/types/configuration/auth_config/connection_string_options.rs
@@ -15,74 +15,14 @@
* specific language governing permissions and limitations
* under the License.
*/
-use crate::{IggyDuration, TcpClientReconnectionConfig};
-use std::str::FromStr;
+use crate::{IggyDuration, IggyError};
-#[derive(Debug)]
-pub struct ConnectionStringOptions {
- tls_enabled: bool,
- tls_domain: String,
- tls_ca_file: Option<String>,
- reconnection: TcpClientReconnectionConfig,
- heartbeat_interval: IggyDuration,
- nodelay: bool,
-}
-
-impl ConnectionStringOptions {
- pub fn new(
- tls_enabled: bool,
- tls_domain: String,
- tls_ca_file: Option<String>,
- reconnection: TcpClientReconnectionConfig,
- heartbeat_interval: IggyDuration,
- nodelay: bool,
- ) -> Self {
- Self {
- tls_enabled,
- tls_domain,
- tls_ca_file,
- reconnection,
- heartbeat_interval,
- nodelay,
- }
- }
-}
-
-impl ConnectionStringOptions {
- pub fn tls_enabled(&self) -> bool {
- self.tls_enabled
- }
-
- pub fn tls_domain(&self) -> &str {
- &self.tls_domain
- }
+pub trait ConnectionStringOptions {
+ fn retries(&self) -> Option<u32>;
- pub fn tls_ca_file(&self) -> &Option<String> {
- &self.tls_ca_file
- }
-
- pub fn reconnection(&self) -> &TcpClientReconnectionConfig {
- &self.reconnection
- }
-
- pub fn heartbeat_interval(&self) -> IggyDuration {
- self.heartbeat_interval
- }
-
- pub fn nodelay(&self) -> bool {
- self.nodelay
- }
-}
+ fn heartbeat_interval(&self) -> IggyDuration;
-impl Default for ConnectionStringOptions {
- fn default() -> Self {
- ConnectionStringOptions {
- tls_enabled: false,
- tls_domain: "".to_string(),
- tls_ca_file: None,
- reconnection: Default::default(),
- heartbeat_interval: IggyDuration::from_str("5s").unwrap(),
- nodelay: false,
- }
- }
+ fn parse_options(options: &str) -> Result<Self, IggyError>
+ where
+ Self: Sized;
}
diff --git a/core/common/src/types/configuration/http_config/config.rs
b/core/common/src/types/configuration/http_config/http_client_config.rs
similarity index 57%
rename from core/common/src/types/configuration/http_config/config.rs
rename to core/common/src/types/configuration/http_config/http_client_config.rs
index 2b08f899..06ec4182 100644
--- a/core/common/src/types/configuration/http_config/config.rs
+++ b/core/common/src/types/configuration/http_config/http_client_config.rs
@@ -16,6 +16,8 @@
* under the License.
*/
+use crate::{ConnectionString, ConnectionStringOptions,
HttpConnectionStringOptions};
+
/// Configuration for the HTTP client.
#[derive(Debug, Clone)]
pub struct HttpClientConfig {
@@ -34,35 +36,11 @@ impl Default for HttpClientConfig {
}
}
-/// The builder for the `HttpClientConfig` configuration.
-/// Allows configuring the HTTP client with custom settings or using defaults:
-/// - `api_url`: Default is "http://127.0.0.1:3000";
-/// - `retries`: Default is 3.
-#[derive(Debug, Default)]
-pub struct HttpClientConfigBuilder {
- config: HttpClientConfig,
-}
-
-impl HttpClientConfigBuilder {
- /// Create a new `HttpClientConfigBuilder` with default settings.
- pub fn new() -> Self {
- HttpClientConfigBuilder::default()
- }
-
- /// Sets the API URL for the HTTP client.
- pub fn with_api_url(mut self, url: String) -> Self {
- self.config.api_url = url;
- self
- }
-
- /// Sets the number of retries for the HTTP client.
- pub fn with_retries(mut self, retries: u32) -> Self {
- self.config.retries = retries;
- self
- }
-
- /// Builds the `HttpClientConfig` instance.
- pub fn build(self) -> HttpClientConfig {
- self.config
+impl From<ConnectionString<HttpConnectionStringOptions>> for HttpClientConfig {
+ fn from(connection_string: ConnectionString<HttpConnectionStringOptions>)
-> Self {
+ HttpClientConfig {
+ api_url: format!("http://{}";, connection_string.server_address()),
+ retries: connection_string.options().retries().unwrap(),
+ }
}
}
diff --git a/core/sdk/src/http/http_config.rs
b/core/common/src/types/configuration/http_config/http_client_config_builder.rs
similarity index 79%
rename from core/sdk/src/http/http_config.rs
rename to
core/common/src/types/configuration/http_config/http_client_config_builder.rs
index 2b08f899..097df0da 100644
--- a/core/sdk/src/http/http_config.rs
+++
b/core/common/src/types/configuration/http_config/http_client_config_builder.rs
@@ -16,23 +16,7 @@
* under the License.
*/
-/// Configuration for the HTTP client.
-#[derive(Debug, Clone)]
-pub struct HttpClientConfig {
- /// The URL of the Iggy API.
- pub api_url: String,
- /// The number of retries to perform on transient errors.
- pub retries: u32,
-}
-
-impl Default for HttpClientConfig {
- fn default() -> HttpClientConfig {
- HttpClientConfig {
- api_url: "http://127.0.0.1:3000".to_string(),
- retries: 3,
- }
- }
-}
+use crate::HttpClientConfig;
/// The builder for the `HttpClientConfig` configuration.
/// Allows configuring the HTTP client with custom settings or using defaults:
diff --git
a/core/common/src/types/configuration/http_config/http_connection_string_options.rs
b/core/common/src/types/configuration/http_config/http_connection_string_options.rs
new file mode 100644
index 00000000..f2e4f15d
--- /dev/null
+++
b/core/common/src/types/configuration/http_config/http_connection_string_options.rs
@@ -0,0 +1,87 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use crate::{ConnectionStringOptions, IggyDuration, IggyError};
+use std::str::FromStr;
+
+#[derive(Debug)]
+pub struct HttpConnectionStringOptions {
+ heartbeat_interval: IggyDuration,
+ retries: u32,
+}
+
+impl ConnectionStringOptions for HttpConnectionStringOptions {
+ fn retries(&self) -> Option<u32> {
+ Some(self.retries)
+ }
+
+ fn heartbeat_interval(&self) -> IggyDuration {
+ self.heartbeat_interval
+ }
+
+ fn parse_options(options: &str) -> Result<HttpConnectionStringOptions,
IggyError> {
+ let options = options.split('&').collect::<Vec<&str>>();
+ let mut heartbeat_interval = "5s".to_owned();
+ let mut retries = 3;
+
+ for option in options {
+ let option_parts = option.split('=').collect::<Vec<&str>>();
+ if option_parts.len() != 2 {
+ return Err(IggyError::InvalidConnectionString);
+ }
+ match option_parts[0] {
+ "heartbeat_interval" => {
+ heartbeat_interval = option_parts[1].to_string();
+ }
+ "retries" => {
+ retries = option_parts[1]
+ .parse::<u32>()
+ .map_err(|_| IggyError::InvalidConnectionString)?;
+ }
+ _ => {
+ return Err(IggyError::InvalidConnectionString);
+ }
+ }
+ }
+
+ let heartbeat_interval =
IggyDuration::from_str(heartbeat_interval.as_str())
+ .map_err(|_| IggyError::InvalidConnectionString)?;
+
+ let connection_string_options =
+ HttpConnectionStringOptions::new(heartbeat_interval, retries);
+ Ok(connection_string_options)
+ }
+}
+
+impl HttpConnectionStringOptions {
+ pub fn new(heartbeat_interval: IggyDuration, retries: u32) -> Self {
+ Self {
+ heartbeat_interval,
+ retries,
+ }
+ }
+}
+
+impl Default for HttpConnectionStringOptions {
+ fn default() -> Self {
+ Self {
+ heartbeat_interval: IggyDuration::from_str("5s").unwrap(),
+ retries: 3,
+ }
+ }
+}
diff --git a/core/common/src/types/configuration/http_config/mod.rs
b/core/common/src/types/configuration/http_config/mod.rs
index 624e11d9..1afc01fc 100644
--- a/core/common/src/types/configuration/http_config/mod.rs
+++ b/core/common/src/types/configuration/http_config/mod.rs
@@ -15,4 +15,6 @@
// specific language governing permissions and limitations
// under the License.
-pub(crate) mod config;
+pub(crate) mod http_client_config;
+pub(crate) mod http_client_config_builder;
+pub(crate) mod http_connection_string_options;
diff --git a/core/common/src/types/configuration/quick_config/mod.rs
b/core/common/src/types/configuration/quick_config/mod.rs
index 038ff150..ed0a4d5e 100644
--- a/core/common/src/types/configuration/quick_config/mod.rs
+++ b/core/common/src/types/configuration/quick_config/mod.rs
@@ -18,3 +18,4 @@
pub(crate) mod quic_client_config;
pub(crate) mod quic_client_config_builder;
pub(crate) mod quic_client_reconnection_config;
+pub(crate) mod quic_connection_string_options;
diff --git
a/core/common/src/types/configuration/quick_config/quic_client_config.rs
b/core/common/src/types/configuration/quick_config/quic_client_config.rs
index fa067952..237a80d7 100644
--- a/core/common/src/types/configuration/quick_config/quic_client_config.rs
+++ b/core/common/src/types/configuration/quick_config/quic_client_config.rs
@@ -16,7 +16,10 @@
* under the License.
*/
-use crate::{AutoLogin, IggyDuration, QuicClientReconnectionConfig};
+use crate::{
+ AutoLogin, ConnectionString, ConnectionStringOptions, IggyDuration,
+ QuicClientReconnectionConfig, QuicConnectionStringOptions,
+};
use std::str::FromStr;
/// Configuration for the QUIC client.
@@ -75,3 +78,25 @@ impl Default for QuicClientConfig {
}
}
}
+
+impl From<ConnectionString<QuicConnectionStringOptions>> for QuicClientConfig {
+ fn from(connection_string: ConnectionString<QuicConnectionStringOptions>)
-> Self {
+ QuicClientConfig {
+ client_address: "127.0.0.1:0".to_string(),
+ server_address: connection_string.server_address().into(),
+ server_name: "localhost".to_string(),
+ auto_login: connection_string.auto_login().to_owned(),
+ reconnection:
connection_string.options().reconnection().to_owned(),
+ response_buffer_size:
connection_string.options().response_buffer_size(),
+ max_concurrent_bidi_streams:
connection_string.options().max_concurrent_bidi_streams(),
+ datagram_send_buffer_size:
connection_string.options().datagram_send_buffer_size(),
+ initial_mtu: connection_string.options().initial_mtu(),
+ send_window: connection_string.options().send_window(),
+ receive_window: connection_string.options().receive_window(),
+ keep_alive_interval:
connection_string.options().keep_alive_interval(),
+ max_idle_timeout: connection_string.options().max_idle_timeout(),
+ validate_certificate:
connection_string.options().validate_certificate(),
+ heartbeat_interval:
connection_string.options().heartbeat_interval(),
+ }
+ }
+}
diff --git
a/core/common/src/types/configuration/quick_config/quic_connection_string_options.rs
b/core/common/src/types/configuration/quick_config/quic_connection_string_options.rs
new file mode 100644
index 00000000..7108dc2f
--- /dev/null
+++
b/core/common/src/types/configuration/quick_config/quic_connection_string_options.rs
@@ -0,0 +1,281 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied. See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*/
+
+use crate::{ConnectionStringOptions, IggyDuration, IggyError,
QuicClientReconnectionConfig};
+use std::str::FromStr;
+
+#[derive(Debug)]
+pub struct QuicConnectionStringOptions {
+ reconnection: QuicClientReconnectionConfig,
+ response_buffer_size: u64,
+ max_concurrent_bidi_streams: u64,
+ datagram_send_buffer_size: u64,
+ initial_mtu: u16,
+ send_window: u64,
+ receive_window: u64,
+ keep_alive_interval: u64,
+ max_idle_timeout: u64,
+ validate_certificate: bool,
+ heartbeat_interval: IggyDuration,
+}
+
+impl QuicConnectionStringOptions {
+ pub fn reconnection(&self) -> &QuicClientReconnectionConfig {
+ &self.reconnection
+ }
+
+ pub fn response_buffer_size(&self) -> u64 {
+ self.response_buffer_size
+ }
+
+ pub fn max_concurrent_bidi_streams(&self) -> u64 {
+ self.max_concurrent_bidi_streams
+ }
+
+ pub fn datagram_send_buffer_size(&self) -> u64 {
+ self.datagram_send_buffer_size
+ }
+
+ pub fn initial_mtu(&self) -> u16 {
+ self.initial_mtu
+ }
+
+ pub fn send_window(&self) -> u64 {
+ self.send_window
+ }
+
+ pub fn receive_window(&self) -> u64 {
+ self.receive_window
+ }
+
+ pub fn keep_alive_interval(&self) -> u64 {
+ self.keep_alive_interval
+ }
+
+ pub fn max_idle_timeout(&self) -> u64 {
+ self.max_idle_timeout
+ }
+
+ pub fn validate_certificate(&self) -> bool {
+ self.validate_certificate
+ }
+}
+
+impl ConnectionStringOptions for QuicConnectionStringOptions {
+ fn retries(&self) -> Option<u32> {
+ self.reconnection.max_retries
+ }
+
+ fn heartbeat_interval(&self) -> IggyDuration {
+ self.heartbeat_interval
+ }
+
+ fn parse_options(options: &str) -> Result<QuicConnectionStringOptions,
IggyError> {
+ let options = options.split('&').collect::<Vec<&str>>();
+ let mut response_buffer_size = 1000 * 1000 * 10;
+ let mut max_concurrent_bidi_streams = 10000;
+ let mut datagram_send_buffer_size = 100_000;
+ let mut initial_mtu = 1200;
+ let mut send_window = 100_000;
+ let mut receive_window = 100_000;
+ let mut keep_alive_interval = 5000;
+ let mut max_idle_timeout = 10000;
+ let mut validate_certificate = false;
+ let mut heartbeat_interval = "5s".to_owned();
+
+ // For reconnection config
+ let mut reconnection_max_retries = "unlimited".to_owned();
+ let mut reconnection_interval = "1s".to_owned();
+ let mut reconnection_reestablish_after = "5s".to_owned();
+
+ for option in options {
+ let option_parts = option.split('=').collect::<Vec<&str>>();
+ if option_parts.len() != 2 {
+ return Err(IggyError::InvalidConnectionString);
+ }
+ match option_parts[0] {
+ "response_buffer_size" => match option_parts[1].parse::<u64>()
{
+ Ok(value) => {
+ response_buffer_size = value;
+ }
+ Err(_) => {
+ return Err(IggyError::InvalidConnectionString);
+ }
+ },
+ "max_concurrent_bidi_streams" => match
option_parts[1].parse::<u64>() {
+ Ok(value) => {
+ max_concurrent_bidi_streams = value;
+ }
+ Err(_) => {
+ return Err(IggyError::InvalidConnectionString);
+ }
+ },
+ "datagram_send_buffer_size" => match
option_parts[1].parse::<u64>() {
+ Ok(value) => {
+ datagram_send_buffer_size = value;
+ }
+ Err(_) => {
+ return Err(IggyError::InvalidConnectionString);
+ }
+ },
+ "initial_mtu" => match option_parts[1].parse::<u16>() {
+ Ok(value) => {
+ initial_mtu = value;
+ }
+ Err(_) => {
+ return Err(IggyError::InvalidConnectionString);
+ }
+ },
+ "send_window" => match option_parts[1].parse::<u64>() {
+ Ok(value) => {
+ send_window = value;
+ }
+ Err(_) => {
+ return Err(IggyError::InvalidConnectionString);
+ }
+ },
+ "receive_window" => match option_parts[1].parse::<u64>() {
+ Ok(value) => {
+ receive_window = value;
+ }
+ Err(_) => {
+ return Err(IggyError::InvalidConnectionString);
+ }
+ },
+ "keep_alive_interval" => match option_parts[1].parse::<u64>() {
+ Ok(value) => {
+ keep_alive_interval = value;
+ }
+ Err(_) => {
+ return Err(IggyError::InvalidConnectionString);
+ }
+ },
+ "max_idle_timeout" => match option_parts[1].parse::<u64>() {
+ Ok(value) => {
+ max_idle_timeout = value;
+ }
+ Err(_) => {
+ return Err(IggyError::InvalidConnectionString);
+ }
+ },
+ "validate_certificate" => {
+ validate_certificate = option_parts[1] == "true";
+ }
+ "heartbeat_interval" => {
+ heartbeat_interval = option_parts[1].to_string();
+ }
+ "reconnection_max_retries" => {
+ reconnection_max_retries = option_parts[1].to_string();
+ }
+ "reconnection_interval" => {
+ reconnection_interval = option_parts[1].to_string();
+ }
+ "reconnection_reestablish_after" => {
+ reconnection_reestablish_after =
option_parts[1].to_string();
+ }
+ _ => {
+ return Err(IggyError::InvalidConnectionString);
+ }
+ }
+ }
+
+ let reconnection = QuicClientReconnectionConfig {
+ enabled: true,
+ max_retries: match reconnection_max_retries.as_str() {
+ "unlimited" => None,
+ _ => Some(
+ reconnection_max_retries
+ .parse()
+ .map_err(|_| IggyError::InvalidNumberValue)?,
+ ),
+ },
+ interval: IggyDuration::from_str(reconnection_interval.as_str())
+ .map_err(|_| IggyError::InvalidConnectionString)?,
+ reestablish_after:
IggyDuration::from_str(reconnection_reestablish_after.as_str())
+ .map_err(|_| IggyError::InvalidConnectionString)?,
+ };
+
+ let heartbeat_interval =
IggyDuration::from_str(heartbeat_interval.as_str())
+ .map_err(|_| IggyError::InvalidConnectionString)?;
+
+ let connection_string_options = QuicConnectionStringOptions::new(
+ reconnection,
+ response_buffer_size,
+ max_concurrent_bidi_streams,
+ datagram_send_buffer_size,
+ initial_mtu,
+ send_window,
+ receive_window,
+ keep_alive_interval,
+ max_idle_timeout,
+ validate_certificate,
+ heartbeat_interval,
+ );
+
+ Ok(connection_string_options)
+ }
+}
+
+impl QuicConnectionStringOptions {
+ #[allow(clippy::too_many_arguments)]
+ pub fn new(
+ reconnection: QuicClientReconnectionConfig,
+ response_buffer_size: u64,
+ max_concurrent_bidi_streams: u64,
+ datagram_send_buffer_size: u64,
+ initial_mtu: u16,
+ send_window: u64,
+ receive_window: u64,
+ keep_alive_interval: u64,
+ max_idle_timeout: u64,
+ validate_certificate: bool,
+ heartbeat_interval: IggyDuration,
+ ) -> Self {
+ Self {
+ reconnection,
+ response_buffer_size,
+ max_concurrent_bidi_streams,
+ datagram_send_buffer_size,
+ initial_mtu,
+ send_window,
+ receive_window,
+ keep_alive_interval,
+ max_idle_timeout,
+ validate_certificate,
+ heartbeat_interval,
+ }
+ }
+}
+
+impl Default for QuicConnectionStringOptions {
+ fn default() -> Self {
+ QuicConnectionStringOptions {
+ reconnection: Default::default(),
+ response_buffer_size: 1000 * 1000 * 10,
+ max_concurrent_bidi_streams: 10000,
+ datagram_send_buffer_size: 100_000,
+ initial_mtu: 1200,
+ send_window: 100_000,
+ receive_window: 100_000,
+ keep_alive_interval: 5000,
+ max_idle_timeout: 10000,
+ validate_certificate: false,
+ heartbeat_interval: IggyDuration::from_str("5s").unwrap(),
+ }
+ }
+}
diff --git a/core/common/src/types/configuration/tcp_config/mod.rs
b/core/common/src/types/configuration/tcp_config/mod.rs
index 592b6409..7e3fbb6e 100644
--- a/core/common/src/types/configuration/tcp_config/mod.rs
+++ b/core/common/src/types/configuration/tcp_config/mod.rs
@@ -18,3 +18,4 @@
pub(crate) mod tcp_client_config;
pub(crate) mod tcp_client_config_builder;
pub(crate) mod tcp_client_reconnection_config;
+pub(crate) mod tcp_connection_string_options;
diff --git
a/core/common/src/types/configuration/tcp_config/tcp_client_config.rs
b/core/common/src/types/configuration/tcp_config/tcp_client_config.rs
index 18319896..9ebc6c1c 100644
--- a/core/common/src/types/configuration/tcp_config/tcp_client_config.rs
+++ b/core/common/src/types/configuration/tcp_config/tcp_client_config.rs
@@ -16,6 +16,8 @@
* under the License.
*/
use
crate::types::configuration::auth_config::connection_string::ConnectionString;
+use
crate::types::configuration::auth_config::connection_string_options::ConnectionStringOptions;
+use
crate::types::configuration::tcp_config::tcp_connection_string_options::TcpConnectionStringOptions;
use crate::{AutoLogin, IggyDuration, TcpClientReconnectionConfig};
use std::str::FromStr;
@@ -55,8 +57,8 @@ impl Default for TcpClientConfig {
}
}
-impl From<ConnectionString> for TcpClientConfig {
- fn from(connection_string: ConnectionString) -> Self {
+impl From<ConnectionString<TcpConnectionStringOptions>> for TcpClientConfig {
+ fn from(connection_string: ConnectionString<TcpConnectionStringOptions>)
-> Self {
TcpClientConfig {
server_address: connection_string.server_address().into(),
auto_login: connection_string.auto_login().to_owned(),
diff --git
a/core/common/src/types/configuration/tcp_config/tcp_connection_string_options.rs
b/core/common/src/types/configuration/tcp_config/tcp_connection_string_options.rs
new file mode 100644
index 00000000..bb4815ef
--- /dev/null
+++
b/core/common/src/types/configuration/tcp_config/tcp_connection_string_options.rs
@@ -0,0 +1,173 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied. See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*/
+
+use crate::{ConnectionStringOptions, IggyDuration, IggyError,
TcpClientReconnectionConfig};
+use std::str::FromStr;
+
+#[derive(Debug)]
+pub struct TcpConnectionStringOptions {
+ tls_enabled: bool,
+ tls_domain: String,
+ tls_ca_file: Option<String>,
+ reconnection: TcpClientReconnectionConfig,
+ heartbeat_interval: IggyDuration,
+ nodelay: bool,
+}
+
+impl TcpConnectionStringOptions {
+ pub fn tls_enabled(&self) -> bool {
+ self.tls_enabled
+ }
+
+ pub fn tls_domain(&self) -> &str {
+ &self.tls_domain
+ }
+
+ pub fn tls_ca_file(&self) -> &Option<String> {
+ &self.tls_ca_file
+ }
+
+ pub fn reconnection(&self) -> &TcpClientReconnectionConfig {
+ &self.reconnection
+ }
+
+ pub fn nodelay(&self) -> bool {
+ self.nodelay
+ }
+}
+
+impl ConnectionStringOptions for TcpConnectionStringOptions {
+ fn retries(&self) -> Option<u32> {
+ self.reconnection.max_retries
+ }
+
+ fn heartbeat_interval(&self) -> IggyDuration {
+ self.heartbeat_interval
+ }
+
+ fn parse_options(options: &str) -> Result<TcpConnectionStringOptions,
IggyError> {
+ let options = options.split('&').collect::<Vec<&str>>();
+ let mut tls_enabled = false;
+ let mut tls_domain = "".to_string();
+ let mut tls_ca_file = None;
+ let mut reconnection_retries = "unlimited".to_owned();
+ let mut reconnection_interval = "1s".to_owned();
+ let mut reestablish_after = "5s".to_owned();
+ let mut heartbeat_interval = "5s".to_owned();
+ let mut nodelay = false;
+
+ for option in options {
+ let option_parts = option.split('=').collect::<Vec<&str>>();
+ if option_parts.len() != 2 {
+ return Err(IggyError::InvalidConnectionString);
+ }
+ match option_parts[0] {
+ "tls" => {
+ tls_enabled = option_parts[1] == "true";
+ }
+ "tls_domain" => {
+ tls_domain = option_parts[1].to_string();
+ }
+ "tls_ca_file" => {
+ tls_ca_file = Some(option_parts[1].to_string());
+ }
+ "reconnection_retries" => {
+ reconnection_retries = option_parts[1].to_string();
+ }
+ "reconnection_interval" => {
+ reconnection_interval = option_parts[1].to_string();
+ }
+ "reestablish_after" => {
+ reestablish_after = option_parts[1].to_string();
+ }
+ "heartbeat_interval" => {
+ heartbeat_interval = option_parts[1].to_string();
+ }
+ "nodelay" => {
+ nodelay = option_parts[1] == "true";
+ }
+ _ => {
+ return Err(IggyError::InvalidConnectionString);
+ }
+ }
+ }
+
+ let reconnection = TcpClientReconnectionConfig {
+ enabled: true,
+ max_retries: match reconnection_retries.as_str() {
+ "unlimited" => None,
+ _ => Some(
+ reconnection_retries
+ .parse()
+ .map_err(|_| IggyError::InvalidNumberValue)?,
+ ),
+ },
+ interval: IggyDuration::from_str(reconnection_interval.as_str())
+ .map_err(|_| IggyError::InvalidConnectionString)?,
+ reestablish_after:
IggyDuration::from_str(reestablish_after.as_str())
+ .map_err(|_| IggyError::InvalidConnectionString)?,
+ };
+
+ let heartbeat_interval =
IggyDuration::from_str(heartbeat_interval.as_str())
+ .map_err(|_| IggyError::InvalidConnectionString)?;
+
+ let connection_string_options = TcpConnectionStringOptions::new(
+ tls_enabled,
+ tls_domain,
+ tls_ca_file,
+ reconnection,
+ heartbeat_interval,
+ nodelay,
+ );
+
+ Ok(connection_string_options)
+ }
+}
+
+impl TcpConnectionStringOptions {
+ pub fn new(
+ tls_enabled: bool,
+ tls_domain: String,
+ tls_ca_file: Option<String>,
+ reconnection: TcpClientReconnectionConfig,
+ heartbeat_interval: IggyDuration,
+ nodelay: bool,
+ ) -> Self {
+ Self {
+ tls_enabled,
+ tls_domain,
+ tls_ca_file,
+ reconnection,
+ heartbeat_interval,
+ nodelay,
+ }
+ }
+}
+
+impl Default for TcpConnectionStringOptions {
+ fn default() -> Self {
+ TcpConnectionStringOptions {
+ tls_enabled: false,
+ tls_domain: "".to_string(),
+ tls_ca_file: None,
+ reconnection: Default::default(),
+ heartbeat_interval: IggyDuration::from_str("5s").unwrap(),
+ nodelay: false,
+ }
+ }
+}
diff --git a/core/integration/src/http_client.rs
b/core/integration/src/http_client.rs
index b4a63d62..afd566ed 100644
--- a/core/integration/src/http_client.rs
+++ b/core/integration/src/http_client.rs
@@ -19,8 +19,7 @@
use crate::test_server::ClientFactory;
use async_trait::async_trait;
use iggy::http::http_client::HttpClient;
-use iggy::http::http_config::HttpClientConfig;
-use iggy::prelude::Client;
+use iggy::prelude::{Client, HttpClientConfig};
use std::sync::Arc;
#[derive(Debug, Clone)]
diff --git a/core/integration/src/quic_client.rs
b/core/integration/src/quic_client.rs
index 16928502..a8ef2446 100644
--- a/core/integration/src/quic_client.rs
+++ b/core/integration/src/quic_client.rs
@@ -18,9 +18,8 @@
use crate::test_server::ClientFactory;
use async_trait::async_trait;
-use iggy::prelude::Client;
+use iggy::prelude::{Client, QuicClientConfig};
use iggy::quic::quick_client::QuicClient;
-use iggy::quic::quick_config::QuicClientConfig;
use std::sync::Arc;
#[derive(Debug, Clone)]
diff --git a/core/sdk/src/client_provider.rs b/core/sdk/src/client_provider.rs
index ce6a404a..9bad4f16 100644
--- a/core/sdk/src/client_provider.rs
+++ b/core/sdk/src/client_provider.rs
@@ -19,12 +19,11 @@
#[allow(deprecated)]
use crate::clients::client::IggyClient;
use crate::http::http_client::HttpClient;
-use crate::http::http_config::HttpClientConfig;
-use crate::prelude::ClientError;
-use crate::prelude::IggyDuration;
-use crate::prelude::{TcpClientConfig, TcpClientReconnectionConfig};
+use crate::prelude::{
+ ClientError, HttpClientConfig, IggyDuration, QuicClientConfig,
QuicClientReconnectionConfig,
+ TcpClientConfig, TcpClientReconnectionConfig,
+};
use crate::quic::quick_client::QuicClient;
-use crate::quic::quick_config::{QuicClientConfig,
QuicClientReconnectionConfig};
use crate::tcp::tcp_client::TcpClient;
use iggy_binary_protocol::Client;
use iggy_common::{AutoLogin, Credentials};
diff --git a/core/sdk/src/clients/client.rs b/core/sdk/src/clients/client.rs
index 3e6e1c2b..4896e629 100644
--- a/core/sdk/src/clients/client.rs
+++ b/core/sdk/src/clients/client.rs
@@ -19,15 +19,19 @@
use crate::clients::client_builder::IggyClientBuilder;
use iggy_common::locking::{IggySharedMut, IggySharedMutFn};
+use crate::http::http_client::HttpClient;
use crate::prelude::EncryptorKind;
use crate::prelude::IggyConsumerBuilder;
use crate::prelude::IggyError;
use crate::prelude::IggyProducerBuilder;
+use crate::quic::quick_client::QuicClient;
use crate::tcp::tcp_client::TcpClient;
use async_broadcast::Receiver;
use async_trait::async_trait;
use iggy_binary_protocol::Client;
-use iggy_common::{Consumer, DiagnosticEvent, Partitioner};
+use iggy_common::{
+ ConnectionStringUtils, Consumer, DiagnosticEvent, Partitioner,
TransportProtocol,
+};
use std::fmt::Debug;
use std::sync::Arc;
use tokio::spawn;
@@ -76,8 +80,17 @@ impl IggyClient {
}
pub fn from_connection_string(connection_string: &str) -> Result<Self,
IggyError> {
- let client =
Box::new(TcpClient::from_connection_string(connection_string)?);
- Ok(IggyClient::new(client))
+ match ConnectionStringUtils::parse_protocol(connection_string)? {
+ TransportProtocol::Tcp => Ok(IggyClient::new(Box::new(
+ TcpClient::from_connection_string(connection_string)?,
+ ))),
+ TransportProtocol::Quic => Ok(IggyClient::new(Box::new(
+ QuicClient::from_connection_string(connection_string)?,
+ ))),
+ TransportProtocol::Http => Ok(IggyClient::new(Box::new(
+ HttpClient::from_connection_string(connection_string)?,
+ ))),
+ }
}
/// Creates a new `IggyClient` with the provided client implementation for
the specific transport and the optional implementations for the `partitioner`
and `encryptor`.
@@ -200,3 +213,185 @@ impl Client for IggyClient {
self.client.read().await.subscribe_events().await
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn should_fail_with_empty_connection_string() {
+ let value = "";
+ let client = IggyClient::from_connection_string(value);
+ assert!(client.is_err());
+ }
+
+ #[test]
+ fn should_fail_without_username() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Tcp;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let client = IggyClient::from_connection_string(&value);
+ assert!(client.is_err());
+ }
+
+ #[test]
+ fn should_fail_without_password() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Tcp;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let client = IggyClient::from_connection_string(&value);
+ assert!(client.is_err());
+ }
+
+ #[test]
+ fn should_fail_without_server_address() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Tcp;
+ let server_address = "";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let client = IggyClient::from_connection_string(&value);
+ assert!(client.is_err());
+ }
+
+ #[test]
+ fn should_fail_without_port() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Tcp;
+ let server_address = "127.0.0.1";
+ let port = "";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let client = IggyClient::from_connection_string(&value);
+ assert!(client.is_err());
+ }
+
+ #[test]
+ fn should_fail_with_invalid_prefix() {
+ let connection_string_prefix = "invalid+";
+ let protocol = TransportProtocol::Tcp;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let client = IggyClient::from_connection_string(&value);
+ assert!(client.is_err());
+ }
+
+ #[test]
+ fn should_succeed_with_default_prefix() {
+ let default_connection_string_prefix = "iggy://";
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{default_connection_string_prefix}{username}:{password}@{server_address}:{port}"
+ );
+ let client = IggyClient::from_connection_string(&value);
+ assert!(client.is_ok());
+ }
+
+ #[test]
+ fn should_succeed_with_tcp_protocol() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Tcp;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let client = IggyClient::from_connection_string(&value);
+ assert!(client.is_ok());
+ }
+
+ #[test]
+ fn should_succeed_with_tcp_protocol_using_pat() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Tcp;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let pat = "iggypat-1234567890abcdef";
+ let value =
format!("{connection_string_prefix}{protocol}://{pat}@{server_address}:{port}");
+ let client = IggyClient::from_connection_string(&value);
+ assert!(client.is_ok());
+ }
+
+ #[tokio::test]
+ async fn should_succeed_with_quic_protocol() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Quic;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let client = IggyClient::from_connection_string(&value);
+ assert!(client.is_ok());
+ }
+
+ #[tokio::test]
+ async fn should_succeed_with_quic_protocol_using_pat() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Quic;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let pat = "iggypat-1234567890abcdef";
+ let value =
format!("{connection_string_prefix}{protocol}://{pat}@{server_address}:{port}");
+ let client = IggyClient::from_connection_string(&value);
+ assert!(client.is_ok());
+ }
+
+ #[test]
+ fn should_succeed_with_http_protocol() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Http;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let client = IggyClient::from_connection_string(&value);
+ assert!(client.is_ok());
+ }
+
+ #[test]
+ fn should_succeed_with_http_protocol_with_pat() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Http;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let pat = "iggypat-1234567890abcdef";
+ let value =
format!("{connection_string_prefix}{protocol}://{pat}@{server_address}:{port}");
+ let client = IggyClient::from_connection_string(&value);
+ assert!(client.is_ok());
+ }
+}
diff --git a/core/sdk/src/clients/client_builder.rs
b/core/sdk/src/clients/client_builder.rs
index 2c048399..04f8cbea 100644
--- a/core/sdk/src/clients/client_builder.rs
+++ b/core/sdk/src/clients/client_builder.rs
@@ -18,13 +18,13 @@
use crate::clients::client::IggyClient;
use crate::http::http_client::HttpClient;
-use crate::http::http_config::HttpClientConfigBuilder;
use crate::prelude::{
- AutoLogin, Client, EncryptorKind, IggyDuration, IggyError, Partitioner,
TcpClientConfigBuilder,
+ AutoLogin, Client, EncryptorKind, HttpClientConfigBuilder, IggyDuration,
IggyError,
+ Partitioner, QuicClientConfigBuilder, TcpClientConfigBuilder,
};
use crate::quic::quick_client::QuicClient;
-use crate::quic::quick_config::QuicClientConfigBuilder;
use crate::tcp::tcp_client::TcpClient;
+use iggy_common::{ConnectionStringUtils, TransportProtocol};
use std::sync::Arc;
use tracing::error;
@@ -45,9 +45,25 @@ impl IggyClientBuilder {
pub fn from_connection_string(connection_string: &str) -> Result<Self,
IggyError> {
let mut builder = Self::new();
- builder.client = Some(Box::new(TcpClient::from_connection_string(
- connection_string,
- )?));
+
+ match ConnectionStringUtils::parse_protocol(connection_string)? {
+ TransportProtocol::Tcp => {
+ builder.client =
Some(Box::new(TcpClient::from_connection_string(
+ connection_string,
+ )?));
+ }
+ TransportProtocol::Quic => {
+ builder.client =
Some(Box::new(QuicClient::from_connection_string(
+ connection_string,
+ )?));
+ }
+ TransportProtocol::Http => {
+ builder.client =
Some(Box::new(HttpClient::from_connection_string(
+ connection_string,
+ )?));
+ }
+ }
+
Ok(builder)
}
@@ -255,3 +271,185 @@ impl HttpClientBuilder {
Ok(client)
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn should_fail_with_empty_connection_string() {
+ let value = "";
+ let client_builder = IggyClientBuilder::from_connection_string(value);
+ assert!(client_builder.is_err());
+ }
+
+ #[test]
+ fn should_fail_without_username() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Tcp;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let client_builder = IggyClientBuilder::from_connection_string(&value);
+ assert!(client_builder.is_err());
+ }
+
+ #[test]
+ fn should_fail_without_password() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Tcp;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let client_builder = IggyClientBuilder::from_connection_string(&value);
+ assert!(client_builder.is_err());
+ }
+
+ #[test]
+ fn should_fail_without_server_address() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Tcp;
+ let server_address = "";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let client_builder = IggyClientBuilder::from_connection_string(&value);
+ assert!(client_builder.is_err());
+ }
+
+ #[test]
+ fn should_fail_without_port() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Tcp;
+ let server_address = "127.0.0.1";
+ let port = "";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let client_builder = IggyClientBuilder::from_connection_string(&value);
+ assert!(client_builder.is_err());
+ }
+
+ #[test]
+ fn should_fail_with_invalid_prefix() {
+ let connection_string_prefix = "invalid+";
+ let protocol = TransportProtocol::Tcp;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let client_builder = IggyClientBuilder::from_connection_string(&value);
+ assert!(client_builder.is_err());
+ }
+
+ #[test]
+ fn should_succeed_with_default_prefix() {
+ let default_connection_string_prefix = "iggy://";
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{default_connection_string_prefix}{username}:{password}@{server_address}:{port}"
+ );
+ let client_builder = IggyClientBuilder::from_connection_string(&value);
+ assert!(client_builder.is_ok());
+ }
+
+ #[test]
+ fn should_succeed_with_tcp_protocol() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Tcp;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let client_builder = IggyClientBuilder::from_connection_string(&value);
+ assert!(client_builder.is_ok());
+ }
+
+ #[test]
+ fn should_succeed_with_tcp_protocol_using_pat() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Tcp;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let pat = "iggypat-1234567890abcdef";
+ let value =
format!("{connection_string_prefix}{protocol}://{pat}@{server_address}:{port}");
+ let client_builder = IggyClientBuilder::from_connection_string(&value);
+ assert!(client_builder.is_ok());
+ }
+
+ #[tokio::test]
+ async fn should_succeed_with_quic_protocol() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Quic;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let client_builder = IggyClientBuilder::from_connection_string(&value);
+ assert!(client_builder.is_ok());
+ }
+
+ #[tokio::test]
+ async fn should_succeed_with_quic_protocol_using_pat() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Quic;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let pat = "iggypat-1234567890abcdef";
+ let value =
format!("{connection_string_prefix}{protocol}://{pat}@{server_address}:{port}");
+ let client_builder = IggyClientBuilder::from_connection_string(&value);
+ assert!(client_builder.is_ok());
+ }
+
+ #[test]
+ fn should_succeed_with_http_protocol() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Http;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let client_builder = IggyClientBuilder::from_connection_string(&value);
+ assert!(client_builder.is_ok());
+ }
+
+ #[test]
+ fn should_succeed_with_http_protocol_with_pat() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Http;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let pat = "iggypat-1234567890abcdef";
+ let value =
format!("{connection_string_prefix}{protocol}://{pat}@{server_address}:{port}");
+ let client_builder = IggyClientBuilder::from_connection_string(&value);
+ assert!(client_builder.is_ok());
+ }
+}
diff --git a/core/sdk/src/http/http_client.rs b/core/sdk/src/http/http_client.rs
index 8ff9ec25..be619dad 100644
--- a/core/sdk/src/http/http_client.rs
+++ b/core/sdk/src/http/http_client.rs
@@ -16,16 +16,15 @@
* under the License.
*/
-use crate::http::http_config::HttpClientConfig;
use crate::http::http_transport::HttpTransport;
-use crate::prelude::Client;
-use crate::prelude::IggyDuration;
-use crate::prelude::IggyError;
+use crate::prelude::{Client, HttpClientConfig, IggyDuration, IggyError};
use async_broadcast::{Receiver, Sender, broadcast};
use async_trait::async_trait;
-use iggy_common::DiagnosticEvent;
-use iggy_common::IdentityInfo;
use iggy_common::locking::{IggySharedMut, IggySharedMutFn};
+use iggy_common::{
+ ConnectionString, ConnectionStringUtils, DiagnosticEvent,
HttpConnectionStringOptions,
+ IdentityInfo, TransportProtocol,
+};
use reqwest::{Response, StatusCode, Url};
use reqwest_middleware::{ClientBuilder, ClientWithMiddleware};
use reqwest_retry::{RetryTransientMiddleware, policies::ExponentialBackoff};
@@ -286,6 +285,17 @@ impl HttpClient {
})
}
+ /// Create a new HttpClient from a connection string.
+ pub fn from_connection_string(connection_string: &str) -> Result<Self,
IggyError> {
+ if ConnectionStringUtils::parse_protocol(connection_string)? !=
TransportProtocol::Http {
+ return Err(IggyError::InvalidConnectionString);
+ }
+
+ Self::create(Arc::new(
+
ConnectionString::<HttpConnectionStringOptions>::from_str(connection_string)?.into(),
+ ))
+ }
+
async fn handle_response(response: Response) -> Result<Response,
IggyError> {
let status = response.status();
match status.is_success() {
@@ -325,3 +335,203 @@ impl HttpClient {
struct RefreshToken {
token: String,
}
+
+/// Unit tests for HttpClient.
+/// Currently only tests for "from_connection_string()" are implemented.
+/// TODO: Add complete unit tests for HttpClient.
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn should_fail_with_empty_connection_string() {
+ let value = "";
+ let http_client = HttpClient::from_connection_string(value);
+ assert!(http_client.is_err());
+ }
+
+ #[test]
+ fn should_fail_without_username() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Http;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let http_client = HttpClient::from_connection_string(&value);
+ assert!(http_client.is_err());
+ }
+
+ #[test]
+ fn should_fail_without_password() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Http;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let http_client = HttpClient::from_connection_string(&value);
+ assert!(http_client.is_err());
+ }
+
+ #[test]
+ fn should_fail_without_server_address() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Http;
+ let server_address = "";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let http_client = HttpClient::from_connection_string(&value);
+ assert!(http_client.is_err());
+ }
+
+ #[test]
+ fn should_fail_without_port() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Http;
+ let server_address = "127.0.0.1";
+ let port = "";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let http_client = HttpClient::from_connection_string(&value);
+ assert!(http_client.is_err());
+ }
+
+ #[test]
+ fn should_fail_with_invalid_prefix() {
+ let connection_string_prefix = "invalid+";
+ let protocol = TransportProtocol::Http;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let http_client = HttpClient::from_connection_string(&value);
+ assert!(http_client.is_err());
+ }
+
+ #[test]
+ fn should_fail_with_unmatch_protocol() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Quic;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let http_client = HttpClient::from_connection_string(&value);
+ assert!(http_client.is_err());
+ }
+
+ #[test]
+ fn should_fail_with_default_prefix() {
+ let default_connection_string_prefix = "iggy://";
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{default_connection_string_prefix}{username}:{password}@{server_address}:{port}"
+ );
+ let http_client = HttpClient::from_connection_string(&value);
+ assert!(http_client.is_err());
+ }
+
+ #[test]
+ fn should_fail_with_invalid_options() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Http;
+ let server_address = "127.0.0.1";
+ let port = "";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}?invalid_option=invalid"
+ );
+ let http_client = HttpClient::from_connection_string(&value);
+ assert!(http_client.is_err());
+ }
+
+ #[test]
+ fn should_succeed_without_options() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Http;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let http_client = HttpClient::from_connection_string(&value);
+ assert!(http_client.is_ok());
+
+ assert_eq!(
+ http_client.as_ref().unwrap().api_url.to_string(),
+ format!("{protocol}://{server_address}:{port}/")
+ );
+ assert_eq!(
+ http_client.as_ref().unwrap().heartbeat_interval,
+ IggyDuration::from_str("5s").unwrap()
+ );
+ }
+
+ #[test]
+ fn should_succeed_with_options() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Http;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let retries = "10";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}?retries={retries}"
+ );
+ let http_client = HttpClient::from_connection_string(&value);
+ assert!(http_client.is_ok());
+
+ assert_eq!(
+ http_client.as_ref().unwrap().api_url.to_string(),
+ format!("{protocol}://{server_address}:{port}/")
+ );
+ }
+
+ #[test]
+ fn should_succeed_with_pat() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Http;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let pat = "iggypat-1234567890abcdef";
+ let value =
format!("{connection_string_prefix}{protocol}://{pat}@{server_address}:{port}");
+ let http_client = HttpClient::from_connection_string(&value);
+ assert!(http_client.is_ok());
+
+ assert_eq!(
+ http_client.as_ref().unwrap().api_url.to_string(),
+ format!("{protocol}://{server_address}:{port}/")
+ );
+ assert_eq!(
+ http_client.as_ref().unwrap().heartbeat_interval,
+ IggyDuration::from_str("5s").unwrap()
+ );
+ }
+}
diff --git a/core/sdk/src/http/mod.rs b/core/sdk/src/http/mod.rs
index 791a28b9..bd8dc649 100644
--- a/core/sdk/src/http/mod.rs
+++ b/core/sdk/src/http/mod.rs
@@ -28,5 +28,4 @@ pub mod binary_topics;
pub mod binary_users;
#[allow(deprecated)]
pub mod http_client;
-pub mod http_config;
mod http_transport;
diff --git a/core/sdk/src/prelude.rs b/core/sdk/src/prelude.rs
index 3d87cc82..97b9a984 100644
--- a/core/sdk/src/prelude.rs
+++ b/core/sdk/src/prelude.rs
@@ -53,14 +53,15 @@ pub use iggy_common::{
Aes256GcmEncryptor, Args, ArgsOptional, AutoLogin, BytesSerializable,
CacheMetrics,
CacheMetricsKey, ClientError, ClientInfoDetails, CompressionAlgorithm,
Confirmation, Consumer,
ConsumerGroupDetails, ConsumerKind, EncryptorKind, FlushUnsavedBuffer,
GlobalPermissions,
- HeaderKey, HeaderValue, IdKind, Identifier, IdentityInfo, IggyByteSize,
IggyDuration,
- IggyError, IggyExpiry, IggyIndexView, IggyMessage, IggyMessageHeader,
IggyMessageHeaderView,
- IggyMessageView, IggyMessageViewIterator, IggyTimestamp, MaxTopicSize,
Partition, Partitioner,
- Partitioning, Permissions, PersonalAccessTokenExpiry, PollMessages,
PolledMessages,
- PollingKind, PollingStrategy, SendMessages, Sizeable, SnapshotCompression,
Stats, Stream,
- StreamPermissions, SystemSnapshotType, TcpClientConfig,
TcpClientConfigBuilder,
- TcpClientReconnectionConfig, Topic, TopicPermissions, UserId, UserStatus,
Validatable,
- defaults, locking,
+ HeaderKey, HeaderValue, HttpClientConfig, HttpClientConfigBuilder, IdKind,
Identifier,
+ IdentityInfo, IggyByteSize, IggyDuration, IggyError, IggyExpiry,
IggyIndexView, IggyMessage,
+ IggyMessageHeader, IggyMessageHeaderView, IggyMessageView,
IggyMessageViewIterator,
+ IggyTimestamp, MaxTopicSize, Partition, Partitioner, Partitioning,
Permissions,
+ PersonalAccessTokenExpiry, PollMessages, PolledMessages, PollingKind,
PollingStrategy,
+ QuicClientConfig, QuicClientConfigBuilder, QuicClientReconnectionConfig,
SendMessages,
+ Sizeable, SnapshotCompression, Stats, Stream, StreamPermissions,
SystemSnapshotType,
+ TcpClientConfig, TcpClientConfigBuilder, TcpClientReconnectionConfig,
Topic, TopicPermissions,
+ UserId, UserStatus, Validatable, defaults, locking,
};
pub use iggy_common::{
IGGY_MESSAGE_CHECKSUM_OFFSET_RANGE, IGGY_MESSAGE_HEADER_SIZE,
diff --git a/core/sdk/src/quic/mod.rs b/core/sdk/src/quic/mod.rs
index 8994b599..0ffce248 100644
--- a/core/sdk/src/quic/mod.rs
+++ b/core/sdk/src/quic/mod.rs
@@ -17,5 +17,4 @@
*/
pub mod quick_client;
-pub mod quick_config;
mod skip_server_verification;
diff --git a/core/sdk/src/quic/quick_client.rs
b/core/sdk/src/quic/quick_client.rs
index 2977c0a2..7d0211ae 100644
--- a/core/sdk/src/quic/quick_client.rs
+++ b/core/sdk/src/quic/quick_client.rs
@@ -21,19 +21,20 @@ use iggy_binary_protocol::{
BinaryClient, BinaryTransport, Client, PersonalAccessTokenClient,
UserClient,
};
-use crate::prelude::IggyDuration;
-use crate::prelude::IggyError;
-use crate::prelude::IggyTimestamp;
-use crate::quic::quick_config::QuicClientConfig;
+use crate::prelude::{IggyDuration, IggyError, IggyTimestamp, QuicClientConfig};
use crate::quic::skip_server_verification::SkipServerVerification;
use async_broadcast::{Receiver, Sender, broadcast};
use async_trait::async_trait;
use bytes::Bytes;
-use iggy_common::{ClientState, Command, Credentials, DiagnosticEvent};
+use iggy_common::{
+ ClientState, Command, ConnectionString, ConnectionStringUtils,
Credentials, DiagnosticEvent,
+ QuicConnectionStringOptions, TransportProtocol,
+};
use quinn::crypto::rustls::QuicClientConfig as QuinnQuicClientConfig;
use quinn::{ClientConfig, Connection, Endpoint, IdleTimeout, RecvStream,
VarInt};
use rustls::crypto::CryptoProvider;
use std::net::SocketAddr;
+use std::str::FromStr;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Mutex;
@@ -202,6 +203,17 @@ impl QuicClient {
})
}
+ /// Creates a new QUIC client from a connection string.
+ pub fn from_connection_string(connection_string: &str) -> Result<Self,
IggyError> {
+ if ConnectionStringUtils::parse_protocol(connection_string)? !=
TransportProtocol::Quic {
+ return Err(IggyError::InvalidConnectionString);
+ }
+
+ Self::create(Arc::new(
+
ConnectionString::<QuicConnectionStringOptions>::from_str(connection_string)?.into(),
+ ))
+ }
+
async fn handle_response(&self, recv: &mut RecvStream) -> Result<Bytes,
IggyError> {
let buffer = recv
.read_to_end(self.config.response_buffer_size as usize)
@@ -536,3 +548,295 @@ fn configure(config: &QuicClientConfig) ->
Result<ClientConfig, IggyError> {
client_config.transport_config(Arc::new(transport));
Ok(client_config)
}
+
+/// Unit tests for QuicClient.
+/// Currently only tests for "from_connection_string()" are implemented.
+/// TODO: Add complete unit tests for QuicClient.
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[tokio::test]
+ async fn should_fail_with_empty_connection_string() {
+ let value = "";
+ let quic_client = QuicClient::from_connection_string(value);
+ assert!(quic_client.is_err());
+ }
+
+ #[tokio::test]
+ async fn should_fail_without_username() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Quic;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let quic_client = QuicClient::from_connection_string(&value);
+ assert!(quic_client.is_err());
+ }
+
+ #[tokio::test]
+ async fn should_fail_without_password() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Quic;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let quic_client = QuicClient::from_connection_string(&value);
+ assert!(quic_client.is_err());
+ }
+
+ #[tokio::test]
+ async fn should_fail_without_server_address() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Quic;
+ let server_address = "";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let quic_client = QuicClient::from_connection_string(&value);
+ assert!(quic_client.is_err());
+ }
+
+ #[tokio::test]
+ async fn should_fail_without_port() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Quic;
+ let server_address = "127.0.0.1";
+ let port = "";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let quic_client = QuicClient::from_connection_string(&value);
+ assert!(quic_client.is_err());
+ }
+
+ #[tokio::test]
+ async fn should_fail_with_invalid_prefix() {
+ let connection_string_prefix = "invalid+";
+ let protocol = TransportProtocol::Quic;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let quic_client = QuicClient::from_connection_string(&value);
+ assert!(quic_client.is_err());
+ }
+
+ #[tokio::test]
+ async fn should_fail_with_unmatch_protocol() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Tcp;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let quic_client = QuicClient::from_connection_string(&value);
+ assert!(quic_client.is_err());
+ }
+
+ #[tokio::test]
+ async fn should_fail_with_default_prefix() {
+ let default_connection_string_prefix = "iggy://";
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{default_connection_string_prefix}{username}:{password}@{server_address}:{port}"
+ );
+ let quic_client = QuicClient::from_connection_string(&value);
+ assert!(quic_client.is_err());
+ }
+
+ #[tokio::test]
+ async fn should_fail_with_invalid_options() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Quic;
+ let server_address = "127.0.0.1";
+ let port = "";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}?invalid_option=invalid"
+ );
+ let quic_client = QuicClient::from_connection_string(&value);
+ assert!(quic_client.is_err());
+ }
+
+ #[tokio::test]
+ async fn should_succeed_without_options() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Quic;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let quic_client = QuicClient::from_connection_string(&value);
+ assert!(quic_client.is_ok());
+
+ let quic_client_config = quic_client.unwrap().config;
+ assert_eq!(
+ quic_client_config.server_address,
+ format!("{server_address}:{port}")
+ );
+ assert_eq!(
+ quic_client_config.auto_login,
+ AutoLogin::Enabled(Credentials::UsernamePassword(
+ username.to_string(),
+ password.to_string()
+ ))
+ );
+
+ assert_eq!(quic_client_config.response_buffer_size, 10_000_000);
+ assert_eq!(quic_client_config.max_concurrent_bidi_streams, 10_000);
+ assert_eq!(quic_client_config.datagram_send_buffer_size, 100_000);
+ assert_eq!(quic_client_config.initial_mtu, 1200);
+ assert_eq!(quic_client_config.send_window, 100_000);
+ assert_eq!(quic_client_config.receive_window, 100_000);
+ assert_eq!(quic_client_config.keep_alive_interval, 5000);
+ assert_eq!(quic_client_config.max_idle_timeout, 10_000);
+ assert!(!quic_client_config.validate_certificate);
+ assert_eq!(
+ quic_client_config.heartbeat_interval,
+ IggyDuration::from_str("5s").unwrap()
+ );
+
+ assert!(quic_client_config.reconnection.enabled);
+ assert!(quic_client_config.reconnection.max_retries.is_none());
+ assert_eq!(
+ quic_client_config.reconnection.interval,
+ IggyDuration::from_str("1s").unwrap()
+ );
+ assert_eq!(
+ quic_client_config.reconnection.reestablish_after,
+ IggyDuration::from_str("5s").unwrap()
+ );
+ }
+
+ #[tokio::test]
+ async fn should_succeed_with_options() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Quic;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let initial_mtu = "3000";
+ let reconnection_interval = "5s";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}?initial_mtu={initial_mtu}&reconnection_interval={reconnection_interval}"
+ );
+ let quic_client = QuicClient::from_connection_string(&value);
+ assert!(quic_client.is_ok());
+
+ let quic_client_config = quic_client.unwrap().config;
+ assert_eq!(
+ quic_client_config.server_address,
+ format!("{server_address}:{port}")
+ );
+ assert_eq!(
+ quic_client_config.auto_login,
+ AutoLogin::Enabled(Credentials::UsernamePassword(
+ username.to_string(),
+ password.to_string()
+ ))
+ );
+
+ assert_eq!(quic_client_config.response_buffer_size, 10_000_000);
+ assert_eq!(quic_client_config.max_concurrent_bidi_streams, 10_000);
+ assert_eq!(quic_client_config.datagram_send_buffer_size, 100_000);
+ assert_eq!(
+ quic_client_config.initial_mtu,
+ initial_mtu.parse::<u16>().unwrap()
+ );
+ assert_eq!(quic_client_config.send_window, 100_000);
+ assert_eq!(quic_client_config.receive_window, 100_000);
+ assert_eq!(quic_client_config.keep_alive_interval, 5000);
+ assert_eq!(quic_client_config.max_idle_timeout, 10_000);
+ assert!(!quic_client_config.validate_certificate);
+ assert_eq!(
+ quic_client_config.heartbeat_interval,
+ IggyDuration::from_str("5s").unwrap()
+ );
+
+ assert!(quic_client_config.reconnection.enabled);
+ assert!(quic_client_config.reconnection.max_retries.is_none());
+ assert_eq!(
+ quic_client_config.reconnection.interval,
+ IggyDuration::from_str(reconnection_interval).unwrap()
+ );
+ assert_eq!(
+ quic_client_config.reconnection.reestablish_after,
+ IggyDuration::from_str("5s").unwrap()
+ );
+ }
+
+ #[tokio::test]
+ async fn should_succeed_with_pat() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Quic;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let pat = "iggypat-1234567890abcdef";
+ let value =
format!("{connection_string_prefix}{protocol}://{pat}@{server_address}:{port}");
+ let quic_client = QuicClient::from_connection_string(&value);
+ assert!(quic_client.is_ok());
+
+ let quic_client_config = quic_client.unwrap().config;
+ assert_eq!(
+ quic_client_config.server_address,
+ format!("{server_address}:{port}")
+ );
+ assert_eq!(
+ quic_client_config.auto_login,
+
AutoLogin::Enabled(Credentials::PersonalAccessToken(pat.to_string()))
+ );
+
+ assert_eq!(quic_client_config.response_buffer_size, 10_000_000);
+ assert_eq!(quic_client_config.max_concurrent_bidi_streams, 10_000);
+ assert_eq!(quic_client_config.datagram_send_buffer_size, 100_000);
+ assert_eq!(quic_client_config.initial_mtu, 1200);
+ assert_eq!(quic_client_config.send_window, 100_000);
+ assert_eq!(quic_client_config.receive_window, 100_000);
+ assert_eq!(quic_client_config.keep_alive_interval, 5000);
+ assert_eq!(quic_client_config.max_idle_timeout, 10_000);
+ assert!(!quic_client_config.validate_certificate);
+ assert_eq!(
+ quic_client_config.heartbeat_interval,
+ IggyDuration::from_str("5s").unwrap()
+ );
+
+ assert!(quic_client_config.reconnection.enabled);
+ assert!(quic_client_config.reconnection.max_retries.is_none());
+ assert_eq!(
+ quic_client_config.reconnection.interval,
+ IggyDuration::from_str("1s").unwrap()
+ );
+ assert_eq!(
+ quic_client_config.reconnection.reestablish_after,
+ IggyDuration::from_str("5s").unwrap()
+ );
+ }
+}
diff --git a/core/sdk/src/quic/quick_config.rs
b/core/sdk/src/quic/quick_config.rs
deleted file mode 100644
index f96f2e8b..00000000
--- a/core/sdk/src/quic/quick_config.rs
+++ /dev/null
@@ -1,232 +0,0 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-use crate::prelude::AutoLogin;
-use crate::prelude::IggyDuration;
-use std::str::FromStr;
-
-/// Configuration for the QUIC client.
-#[derive(Debug, Clone)]
-pub struct QuicClientConfig {
- /// The address to bind the QUIC client to.
- pub client_address: String,
- /// The address of the QUIC server to connect to.
- pub server_address: String,
- /// The server name to use.
- pub server_name: String,
- /// Whether to automatically login user after establishing connection.
- pub auto_login: AutoLogin,
- // Whether to automatically reconnect when disconnected.
- pub reconnection: QuicClientReconnectionConfig,
- /// The size of the response buffer.
- pub response_buffer_size: u64,
- /// The maximum number of concurrent bidirectional streams.
- pub max_concurrent_bidi_streams: u64,
- /// The size of the datagram send buffer.
- pub datagram_send_buffer_size: u64,
- /// The initial MTU.
- pub initial_mtu: u16,
- /// The send window.
- pub send_window: u64,
- /// The receive window.
- pub receive_window: u64,
- /// The keep alive interval.
- pub keep_alive_interval: u64,
- /// The maximum idle timeout.
- pub max_idle_timeout: u64,
- /// Whether to validate the server certificate.
- pub validate_certificate: bool,
- /// Interval of heartbeats sent by the client
- pub heartbeat_interval: IggyDuration,
-}
-
-#[derive(Debug, Clone)]
-pub struct QuicClientReconnectionConfig {
- pub enabled: bool,
- pub max_retries: Option<u32>,
- pub interval: IggyDuration,
- pub reestablish_after: IggyDuration,
-}
-
-impl Default for QuicClientReconnectionConfig {
- fn default() -> QuicClientReconnectionConfig {
- QuicClientReconnectionConfig {
- enabled: true,
- max_retries: None,
- interval: IggyDuration::from_str("1s").unwrap(),
- reestablish_after: IggyDuration::from_str("5s").unwrap(),
- }
- }
-}
-
-impl Default for QuicClientConfig {
- fn default() -> QuicClientConfig {
- QuicClientConfig {
- client_address: "127.0.0.1:0".to_string(),
- server_address: "127.0.0.1:8080".to_string(),
- server_name: "localhost".to_string(),
- auto_login: AutoLogin::Disabled,
- heartbeat_interval: IggyDuration::from_str("5s").unwrap(),
- reconnection: QuicClientReconnectionConfig::default(),
- response_buffer_size: 1000 * 1000 * 10,
- max_concurrent_bidi_streams: 10000,
- datagram_send_buffer_size: 100_000,
- initial_mtu: 1200,
- send_window: 100_000,
- receive_window: 100_000,
- keep_alive_interval: 5000,
- max_idle_timeout: 10000,
- validate_certificate: false,
- }
- }
-}
-
-/// Builder for the QUIC client configuration.
-///
-/// Allows configuring the QUIC client with custom settings or using defaults:
-/// - `client_address`: Default is "127.0.0.1:0" (binds to any available port).
-/// - `server_address`: Default is "127.0.0.1:8080".
-/// - `server_name`: Default is "localhost".
-/// - `auto_login`: Default is AutoLogin::Disabled.
-/// - `reconnection`: Default is enabled unlimited retries and 1 second
interval.
-/// - `response_buffer_size`: Default is 10MB (10,000,000 bytes).
-/// - `max_concurrent_bidi_streams`: Default is 10,000 streams.
-/// - `datagram_send_buffer_size`: Default is 100,000 bytes.
-/// - `initial_mtu`: Default is 1200 bytes.
-/// - `send_window`: Default is 100,000 bytes.
-/// - `receive_window`: Default is 100,000 bytes.
-/// - `keep_alive_interval`: Default is 5000 milliseconds.
-/// - `max_idle_timeout`: Default is 10,000 milliseconds.
-/// - `validate_certificate`: Default is false (certificate validation is
disabled).
-#[derive(Debug, Default)]
-pub struct QuicClientConfigBuilder {
- config: QuicClientConfig,
-}
-
-impl QuicClientConfigBuilder {
- /// Creates a new builder instance with default configuration values.
- pub fn new() -> Self {
- QuicClientConfigBuilder::default()
- }
-
- /// Sets the client address. Defaults to "127.0.0.1:0".
- pub fn with_client_address(mut self, client_address: String) -> Self {
- self.config.client_address = client_address;
- self
- }
-
- /// Sets the server address. Defaults to "127.0.0.1:8080".
- pub fn with_server_address(mut self, server_address: String) -> Self {
- self.config.server_address = server_address;
- self
- }
-
- /// Sets the auto sign in during connection.
- pub fn with_auto_sign_in(mut self, auto_sign_in: AutoLogin) -> Self {
- self.config.auto_login = auto_sign_in;
- self
- }
-
- /// Sets the server name. Defaults to "localhost".
- pub fn with_server_name(mut self, server_name: String) -> Self {
- self.config.server_name = server_name;
- self
- }
-
- pub fn with_enabled_reconnection(mut self) -> Self {
- self.config.reconnection.enabled = true;
- self
- }
-
- /// Sets the number of retries when connecting to the server.
- pub fn with_reconnection_max_retries(mut self, max_retries: Option<u32>)
-> Self {
- self.config.reconnection.max_retries = max_retries;
- self
- }
-
- /// Sets the interval between retries when connecting to the server.
- pub fn with_reconnection_interval(mut self, interval: IggyDuration) ->
Self {
- self.config.reconnection.interval = interval;
- self
- }
-
- /// Sets the response buffer size in bytes. Defaults to 10MB (10,000,000
bytes).
- pub fn with_response_buffer_size(mut self, response_buffer_size: u64) ->
Self {
- self.config.response_buffer_size = response_buffer_size;
- self
- }
-
- /// Sets the maximum number of concurrent bidirectional streams. Defaults
to 10,000.
- pub fn with_max_concurrent_bidi_streams(mut self,
max_concurrent_bidi_streams: u64) -> Self {
- self.config.max_concurrent_bidi_streams = max_concurrent_bidi_streams;
- self
- }
-
- /// Sets the datagram send buffer size in bytes. Defaults to 100,000 bytes.
- pub fn with_datagram_send_buffer_size(mut self, datagram_send_buffer_size:
u64) -> Self {
- self.config.datagram_send_buffer_size = datagram_send_buffer_size;
- self
- }
-
- /// Sets the initial MTU (Maximum Transmission Unit) in bytes. Defaults to
1200 bytes.
- pub fn with_initial_mtu(mut self, initial_mtu: u16) -> Self {
- self.config.initial_mtu = initial_mtu;
- self
- }
-
- /// Sets the send window size in bytes. Defaults to 100,000 bytes.
- pub fn with_send_window(mut self, send_window: u64) -> Self {
- self.config.send_window = send_window;
- self
- }
-
- /// Sets the receive window size in bytes. Defaults to 100,000 bytes.
- pub fn with_receive_window(mut self, receive_window: u64) -> Self {
- self.config.receive_window = receive_window;
- self
- }
-
- /// Sets the keep-alive interval in milliseconds. Defaults to 5000ms.
- pub fn with_keep_alive_interval(mut self, keep_alive_interval: u64) ->
Self {
- self.config.keep_alive_interval = keep_alive_interval;
- self
- }
-
- /// Sets the maximum idle timeout in milliseconds. Defaults to 10,000ms.
- pub fn with_max_idle_timeout(mut self, max_idle_timeout: u64) -> Self {
- self.config.max_idle_timeout = max_idle_timeout;
- self
- }
-
- /// Enables or disables certificate validation. Defaults to false
(disabled).
- pub fn with_validate_certificate(mut self, validate_certificate: bool) ->
Self {
- self.config.validate_certificate = validate_certificate;
- self
- }
-
- /// Sets the heartbeat interval. Defaults to 5000ms.
- pub fn with_heartbeat_interval(mut self, interval: IggyDuration) -> Self {
- self.config.heartbeat_interval = interval;
- self
- }
-
- /// Finalizes the builder and returns the `QuicClientConfig`.
- pub fn build(self) -> QuicClientConfig {
- self.config
- }
-}
diff --git a/core/sdk/src/tcp/tcp_client.rs b/core/sdk/src/tcp/tcp_client.rs
index 8d0d556a..3ea37500 100644
--- a/core/sdk/src/tcp/tcp_client.rs
+++ b/core/sdk/src/tcp/tcp_client.rs
@@ -26,8 +26,9 @@ use async_trait::async_trait;
use bytes::{BufMut, Bytes, BytesMut};
use iggy_binary_protocol::{BinaryClient, BinaryTransport,
PersonalAccessTokenClient, UserClient};
use iggy_common::{
- AutoLogin, ClientState, Command, ConnectionString, Credentials,
DiagnosticEvent, IggyDuration,
- IggyError, IggyErrorDiscriminants, IggyTimestamp,
+ AutoLogin, ClientState, Command, ConnectionString, ConnectionStringUtils,
Credentials,
+ DiagnosticEvent, IggyDuration, IggyError, IggyErrorDiscriminants,
IggyTimestamp,
+ TcpConnectionStringOptions, TransportProtocol,
};
use rustls::pki_types::{CertificateDer, ServerName, pem::PemObject};
use std::net::SocketAddr;
@@ -178,8 +179,12 @@ impl TcpClient {
}
pub fn from_connection_string(connection_string: &str) -> Result<Self,
IggyError> {
+ if ConnectionStringUtils::parse_protocol(connection_string)? !=
TransportProtocol::Tcp {
+ return Err(IggyError::InvalidConnectionString);
+ }
+
Self::create(Arc::new(
- ConnectionString::from_str(connection_string)?.into(),
+
ConnectionString::<TcpConnectionStringOptions>::from_str(connection_string)?.into(),
))
}
@@ -526,3 +531,277 @@ impl TcpClient {
}
}
}
+
+/// Unit tests for TcpClient.
+/// Currently only tests for "from_connection_string()" are implemented.
+/// TODO: Add complete unit tests for TcpClient.
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn should_fail_with_empty_connection_string() {
+ let value = "";
+ let tcp_client = TcpClient::from_connection_string(value);
+ assert!(tcp_client.is_err());
+ }
+
+ #[test]
+ fn should_fail_without_username() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Tcp;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let tcp_client = TcpClient::from_connection_string(&value);
+ assert!(tcp_client.is_err());
+ }
+
+ #[test]
+ fn should_fail_without_password() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Tcp;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let tcp_client = TcpClient::from_connection_string(&value);
+ assert!(tcp_client.is_err());
+ }
+
+ #[test]
+ fn should_fail_without_server_address() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Tcp;
+ let server_address = "";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let tcp_client = TcpClient::from_connection_string(&value);
+ assert!(tcp_client.is_err());
+ }
+
+ #[test]
+ fn should_fail_without_port() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Tcp;
+ let server_address = "127.0.0.1";
+ let port = "";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let tcp_client = TcpClient::from_connection_string(&value);
+ assert!(tcp_client.is_err());
+ }
+
+ #[test]
+ fn should_fail_with_invalid_prefix() {
+ let connection_string_prefix = "invalid+";
+ let protocol = TransportProtocol::Tcp;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let tcp_client = TcpClient::from_connection_string(&value);
+ assert!(tcp_client.is_err());
+ }
+
+ #[test]
+ fn should_fail_with_unmatch_protocol() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Quic;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let tcp_client = TcpClient::from_connection_string(&value);
+ assert!(tcp_client.is_err());
+ }
+
+ #[test]
+ fn should_succeed_with_default_prefix() {
+ let default_connection_string_prefix = "iggy://";
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{default_connection_string_prefix}{username}:{password}@{server_address}:{port}"
+ );
+ let tcp_client = TcpClient::from_connection_string(&value);
+ assert!(tcp_client.is_ok());
+ }
+
+ #[test]
+ fn should_fail_with_invalid_options() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Tcp;
+ let server_address = "127.0.0.1";
+ let port = "";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}?invalid_option=invalid"
+ );
+ let tcp_client = TcpClient::from_connection_string(&value);
+ assert!(tcp_client.is_err());
+ }
+
+ #[test]
+ fn should_succeed_without_options() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Tcp;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}"
+ );
+ let tcp_client = TcpClient::from_connection_string(&value);
+ assert!(tcp_client.is_ok());
+
+ let tcp_client_config = tcp_client.unwrap().config;
+ assert_eq!(
+ tcp_client_config.server_address,
+ format!("{server_address}:{port}")
+ );
+ assert_eq!(
+ tcp_client_config.auto_login,
+ AutoLogin::Enabled(Credentials::UsernamePassword(
+ username.to_string(),
+ password.to_string()
+ ))
+ );
+
+ assert!(!tcp_client_config.tls_enabled);
+ assert!(tcp_client_config.tls_domain.is_empty());
+ assert!(tcp_client_config.tls_ca_file.is_none());
+ assert_eq!(
+ tcp_client_config.heartbeat_interval,
+ IggyDuration::from_str("5s").unwrap()
+ );
+
+ assert!(tcp_client_config.reconnection.enabled);
+ assert!(tcp_client_config.reconnection.max_retries.is_none());
+ assert_eq!(
+ tcp_client_config.reconnection.interval,
+ IggyDuration::from_str("1s").unwrap()
+ );
+ assert_eq!(
+ tcp_client_config.reconnection.reestablish_after,
+ IggyDuration::from_str("5s").unwrap()
+ );
+ }
+
+ #[test]
+ fn should_succeed_with_options() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Tcp;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let username = "user";
+ let password = "secret";
+ let heartbeat_interval = "10s";
+ let reconnection_retries = "10";
+ let value = format!(
+
"{connection_string_prefix}{protocol}://{username}:{password}@{server_address}:{port}?heartbeat_interval={heartbeat_interval}&reconnection_retries={reconnection_retries}"
+ );
+ let tcp_client = TcpClient::from_connection_string(&value);
+ assert!(tcp_client.is_ok());
+
+ let tcp_client_config = tcp_client.unwrap().config;
+ assert_eq!(
+ tcp_client_config.server_address,
+ format!("{server_address}:{port}")
+ );
+ assert_eq!(
+ tcp_client_config.auto_login,
+ AutoLogin::Enabled(Credentials::UsernamePassword(
+ username.to_string(),
+ password.to_string()
+ ))
+ );
+
+ assert!(!tcp_client_config.tls_enabled);
+ assert!(tcp_client_config.tls_domain.is_empty());
+ assert!(tcp_client_config.tls_ca_file.is_none());
+ assert_eq!(
+ tcp_client_config.heartbeat_interval,
+ IggyDuration::from_str(heartbeat_interval).unwrap()
+ );
+
+ assert!(tcp_client_config.reconnection.enabled);
+ assert_eq!(
+ tcp_client_config.reconnection.max_retries.unwrap(),
+ reconnection_retries.parse::<u32>().unwrap()
+ );
+ assert_eq!(
+ tcp_client_config.reconnection.interval,
+ IggyDuration::from_str("1s").unwrap()
+ );
+ assert_eq!(
+ tcp_client_config.reconnection.reestablish_after,
+ IggyDuration::from_str("5s").unwrap()
+ );
+ }
+
+ #[test]
+ fn should_succeed_with_pat() {
+ let connection_string_prefix = "iggy+";
+ let protocol = TransportProtocol::Tcp;
+ let server_address = "127.0.0.1";
+ let port = "1234";
+ let pat = "iggypat-1234567890abcdef";
+ let value =
format!("{connection_string_prefix}{protocol}://{pat}@{server_address}:{port}");
+ let tcp_client = TcpClient::from_connection_string(&value);
+ assert!(tcp_client.is_ok());
+
+ let tcp_client_config = tcp_client.unwrap().config;
+ assert_eq!(
+ tcp_client_config.server_address,
+ format!("{server_address}:{port}")
+ );
+ assert_eq!(
+ tcp_client_config.auto_login,
+
AutoLogin::Enabled(Credentials::PersonalAccessToken(pat.to_string()))
+ );
+
+ assert!(!tcp_client_config.tls_enabled);
+ assert!(tcp_client_config.tls_domain.is_empty());
+ assert!(tcp_client_config.tls_ca_file.is_none());
+ assert_eq!(
+ tcp_client_config.heartbeat_interval,
+ IggyDuration::from_str("5s").unwrap()
+ );
+
+ assert!(tcp_client_config.reconnection.enabled);
+ assert!(tcp_client_config.reconnection.max_retries.is_none());
+ assert_eq!(
+ tcp_client_config.reconnection.interval,
+ IggyDuration::from_str("1s").unwrap()
+ );
+ assert_eq!(
+ tcp_client_config.reconnection.reestablish_after,
+ IggyDuration::from_str("5s").unwrap()
+ );
+ }
+}
