You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by ba...@apache.org on 2025/08/22 02:06:51 UTC
(iggy) 01/01: draft version of sans io tcp client
This is an automated email from the ASF dual-hosted git repository.
bashirbekov pushed a commit to branch draft-sans-new
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 814e4448823284347bf2472efea50559aa240910
Author: haze518 <as...@gmail.com>
AuthorDate: Thu Aug 21 07:09:16 2025 +0600
draft version of sans io tcp client
---
Cargo.lock | 10 +
core/bench/src/utils/mod.rs | 20 +-
core/common/src/error/iggy_error.rs | 14 +
core/integration/src/tcp_client.rs | 22 +-
core/sdk/Cargo.toml | 4 +
core/sdk/src/client_wrappers/binary_client.rs | 4 +
.../binary_consumer_group_client.rs | 27 +
.../binary_consumer_offset_client.rs | 16 +
.../src/client_wrappers/binary_message_client.rs | 23 +
.../src/client_wrappers/binary_partition_client.rs | 10 +
.../binary_personal_access_token_client.rs | 4 +
.../src/client_wrappers/binary_segment_client.rs | 5 +
.../src/client_wrappers/binary_stream_client.rs | 6 +
.../src/client_wrappers/binary_system_client.rs | 7 +
.../sdk/src/client_wrappers/binary_topic_client.rs | 31 +
core/sdk/src/client_wrappers/binary_user_client.rs | 17 +
core/sdk/src/client_wrappers/client_wrapper.rs | 2 +
core/sdk/src/connection/mod.rs | 765 +++++++++++++++++++++
core/sdk/src/lib.rs | 2 +
core/sdk/src/protocol/mod.rs | 289 ++++++++
20 files changed, 1272 insertions(+), 6 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 4db7dd24..f22e116f 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1508,6 +1508,12 @@ dependencies = [
"static_assertions",
]
+[[package]]
+name = "compat"
+version = "0.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1993b0854368a452b0011692f128362d9b42dbe17e009b1b4dd10e91ede39c27"
+
[[package]]
name = "concurrent-queue"
version = "2.5.0"
@@ -3796,6 +3802,7 @@ dependencies = [
"async-trait",
"bon",
"bytes",
+ "compat",
"dashmap",
"flume",
"futures",
@@ -3804,14 +3811,17 @@ dependencies = [
"iggy_common",
"mockall",
"num_cpus",
+ "parking_lot 0.12.4",
"quinn",
"reqwest",
"reqwest-middleware",
"reqwest-retry",
+ "rustc-hash 2.1.1",
"rustls",
"serde",
"tokio",
"tokio-rustls",
+ "tokio-util",
"tracing",
"trait-variant",
"webpki-roots 1.0.2",
diff --git a/core/bench/src/utils/mod.rs b/core/bench/src/utils/mod.rs
index cc6d5fc4..6fe7a1e9 100644
--- a/core/bench/src/utils/mod.rs
+++ b/core/bench/src/utils/mod.rs
@@ -62,15 +62,30 @@ pub fn batch_user_size_bytes(polled_messages: &PolledMessages) -> u64 {
}
pub async fn get_server_stats(client_factory: &Arc<dyn ClientFactory>) -> Result<Stats, IggyError> {
+ println!("DEBUG: get_server_stats - Creating client");
let client = client_factory.create_client().await;
let client = IggyClient::create(client, None, None);
+ println!("DEBUG: get_server_stats - Connecting");
client.connect().await?;
+
+ println!("DEBUG: get_server_stats - Logging in");
client
.login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
.await?;
- client.get_stats().await
+ println!("DEBUG: get_server_stats - Getting stats");
+ let stats = client.get_stats().await?;
+
+ println!("DEBUG: get_server_stats - Disconnecting");
+ // Disconnect the client to ensure proper cleanup with NewTcpClient
+ let _ = client.disconnect().await;
+
+ println!("DEBUG: get_server_stats - Dropping client");
+ drop(client);
+
+ println!("DEBUG: get_server_stats - Completed successfully");
+ Ok(stats)
}
pub async fn collect_server_logs_and_save_to_file(
@@ -93,6 +108,9 @@ pub async fn collect_server_logs_and_save_to_file(
.await?
.0;
+ // Disconnect the client to ensure proper cleanup
+ let _ = client.disconnect().await;
+
fs::write(output_dir.join("server_logs.zip"), snapshot).map_err(|e| {
error!("Failed to write server logs to file: {:?}", e);
IggyError::CannotWriteToFile
diff --git a/core/common/src/error/iggy_error.rs b/core/common/src/error/iggy_error.rs
index cc8a284b..b2209ced 100644
--- a/core/common/src/error/iggy_error.rs
+++ b/core/common/src/error/iggy_error.rs
@@ -459,6 +459,20 @@ pub enum IggyError {
CannotReadIndexPosition = 10011,
#[error("Cannot read index timestamp")]
CannotReadIndexTimestamp = 10012,
+ #[error("Send queue is full")]
+ SendQueueFull = 10050,
+ #[error("Too many pending requests")]
+ TooManyPendingRequests = 10051,
+ #[error("IO Error")]
+ IoError = 10052,
+ #[error("Max number of retry has exceeded")]
+ MaxRetriesExceeded = 10053,
+ #[error("Connection timeout")]
+ ConnectionTimeout = 10054,
+ #[error("Incorrect connection state")]
+ IncorrectConnectionState = 10055,
+ #[error("Connection missed socket")]
+ ConnectionMissedSocket = 10056,
}
impl IggyError {
diff --git a/core/integration/src/tcp_client.rs b/core/integration/src/tcp_client.rs
index 9df8d9b2..7fc6e505 100644
--- a/core/integration/src/tcp_client.rs
+++ b/core/integration/src/tcp_client.rs
@@ -18,8 +18,8 @@
use crate::test_server::{ClientFactory, Transport};
use async_trait::async_trait;
-use iggy::prelude::{Client, ClientWrapper, TcpClient, TcpClientConfig};
-use std::sync::Arc;
+use iggy::{connection::{tokio_tcp, NewTcpClient, SFut, StreamFactory, TokioCompat}, prelude::{Client, ClientWrapper, IggyClient, TcpClient, TcpClientConfig}};
+use std::{net::SocketAddr, sync::Arc};
#[derive(Debug, Clone, Default)]
pub struct TcpClientFactory {
@@ -43,12 +43,24 @@ impl ClientFactory for TcpClientFactory {
tls_validate_certificate: self.tls_validate_certificate,
..TcpClientConfig::default()
};
- let client = TcpClient::create(Arc::new(config)).unwrap_or_else(|e| {
+
+ let factory: StreamFactory<TokioCompat> = Arc::new(move |addr: SocketAddr| -> SFut<TokioCompat> {
+ Box::pin(tokio_tcp(addr))
+ });
+
+ let client = NewTcpClient::create(Arc::new(config), factory).unwrap_or_else(|e| {
panic!(
- "Failed to create TcpClient, iggy-server has address {}, error: {:?}",
+ "Failed to create NewTcpClient, iggy-server has address {}, error: {:?}",
self.server_addr, e
)
});
+ // let client = TcpClient::create(Arc::new(config)).unwrap_or_else(|e| {
+ // panic!(
+ // "Failed to create TcpClient, iggy-server has address {}, error: {:?}",
+ // self.server_addr, e
+ // )
+ // });
+
Client::connect(&client).await.unwrap_or_else(|e| {
if self.tls_enabled {
panic!(
@@ -65,7 +77,7 @@ impl ClientFactory for TcpClientFactory {
)
}
});
- ClientWrapper::Tcp(client)
+ ClientWrapper::New(client)
}
fn transport(&self) -> Transport {
diff --git a/core/sdk/Cargo.toml b/core/sdk/Cargo.toml
index c6b26729..5b973c3d 100644
--- a/core/sdk/Cargo.toml
+++ b/core/sdk/Cargo.toml
@@ -40,6 +40,7 @@ async-dropper = { workspace = true }
async-trait = { workspace = true }
bon = { workspace = true }
bytes = { workspace = true }
+compat = "0.0.0"
dashmap = { workspace = true }
# Can be removed if the feature above gets removed.
#fast-async-mutex = { version = "0.6.7", optional = true }
@@ -49,14 +50,17 @@ futures-util = { workspace = true }
iggy_binary_protocol = { workspace = true }
iggy_common = { workspace = true }
num_cpus = "1.17.0"
+parking_lot = "0.12.4"
quinn = { workspace = true }
reqwest = { workspace = true }
reqwest-middleware = { workspace = true }
reqwest-retry = { workspace = true }
+rustc-hash = "2.1.1"
rustls = { workspace = true }
serde = { workspace = true }
tokio = { workspace = true }
tokio-rustls = { workspace = true }
+tokio-util.workspace = true
tracing = { workspace = true }
trait-variant = { workspace = true }
webpki-roots = { workspace = true }
diff --git a/core/sdk/src/client_wrappers/binary_client.rs b/core/sdk/src/client_wrappers/binary_client.rs
index 41d5744e..05bb3e05 100644
--- a/core/sdk/src/client_wrappers/binary_client.rs
+++ b/core/sdk/src/client_wrappers/binary_client.rs
@@ -30,6 +30,7 @@ impl Client for ClientWrapper {
ClientWrapper::Http(client) => client.connect().await,
ClientWrapper::Tcp(client) => client.connect().await,
ClientWrapper::Quic(client) => client.connect().await,
+ ClientWrapper::New(client) => client.connect().await,
}
}
@@ -39,6 +40,7 @@ impl Client for ClientWrapper {
ClientWrapper::Http(client) => client.disconnect().await,
ClientWrapper::Tcp(client) => client.disconnect().await,
ClientWrapper::Quic(client) => client.disconnect().await,
+ ClientWrapper::New(client) => client.disconnect().await,
}
}
@@ -48,6 +50,7 @@ impl Client for ClientWrapper {
ClientWrapper::Http(client) => client.shutdown().await,
ClientWrapper::Tcp(client) => client.shutdown().await,
ClientWrapper::Quic(client) => client.shutdown().await,
+ ClientWrapper::New(client) => client.shutdown().await,
}
}
@@ -57,6 +60,7 @@ impl Client for ClientWrapper {
ClientWrapper::Http(client) => client.subscribe_events().await,
ClientWrapper::Tcp(client) => client.subscribe_events().await,
ClientWrapper::Quic(client) => client.subscribe_events().await,
+ ClientWrapper::New(client) => client.subscribe_events().await,
}
}
}
diff --git a/core/sdk/src/client_wrappers/binary_consumer_group_client.rs b/core/sdk/src/client_wrappers/binary_consumer_group_client.rs
index 18117588..6a8d8fc4 100644
--- a/core/sdk/src/client_wrappers/binary_consumer_group_client.rs
+++ b/core/sdk/src/client_wrappers/binary_consumer_group_client.rs
@@ -51,6 +51,9 @@ impl ConsumerGroupClient for ClientWrapper {
.get_consumer_group(stream_id, topic_id, group_id)
.await
}
+ ClientWrapper::New(client) => {
+ client.get_consumer_group(stream_id, topic_id, group_id).await
+ }
}
}
@@ -64,6 +67,7 @@ impl ConsumerGroupClient for ClientWrapper {
ClientWrapper::Http(client) => client.get_consumer_groups(stream_id, topic_id).await,
ClientWrapper::Tcp(client) => client.get_consumer_groups(stream_id, topic_id).await,
ClientWrapper::Quic(client) => client.get_consumer_groups(stream_id, topic_id).await,
+ ClientWrapper::New(client) => client.get_consumer_groups(stream_id, topic_id).await,
}
}
@@ -95,6 +99,11 @@ impl ConsumerGroupClient for ClientWrapper {
.create_consumer_group(stream_id, topic_id, name, group_id)
.await
}
+ ClientWrapper::New(client) => {
+ client
+ .create_consumer_group(stream_id, topic_id, name, group_id)
+ .await
+ }
}
}
@@ -125,6 +134,11 @@ impl ConsumerGroupClient for ClientWrapper {
.delete_consumer_group(stream_id, topic_id, group_id)
.await
}
+ ClientWrapper::New(client) => {
+ client
+ .delete_consumer_group(stream_id, topic_id, group_id)
+ .await
+ }
}
}
@@ -155,6 +169,11 @@ impl ConsumerGroupClient for ClientWrapper {
.join_consumer_group(stream_id, topic_id, group_id)
.await
}
+ ClientWrapper::New(client) => {
+ client
+ .join_consumer_group(stream_id, topic_id, group_id)
+ .await
+ }
}
}
@@ -185,6 +204,11 @@ impl ConsumerGroupClient for ClientWrapper {
.leave_consumer_group(stream_id, topic_id, group_id)
.await
}
+ ClientWrapper::New(client) => {
+ client
+ .leave_consumer_group(stream_id, topic_id, group_id)
+ .await
+ }
}
}
}
@@ -205,6 +229,9 @@ impl AsyncDrop for ClientWrapper {
ClientWrapper::Quic(client) => {
let _ = client.logout_user().await;
}
+ ClientWrapper::New(client) => {
+ let _ = client.logout_user().await;
+ }
}
}
}
diff --git a/core/sdk/src/client_wrappers/binary_consumer_offset_client.rs b/core/sdk/src/client_wrappers/binary_consumer_offset_client.rs
index 668f120b..8881967b 100644
--- a/core/sdk/src/client_wrappers/binary_consumer_offset_client.rs
+++ b/core/sdk/src/client_wrappers/binary_consumer_offset_client.rs
@@ -52,6 +52,12 @@ impl ConsumerOffsetClient for ClientWrapper {
.store_consumer_offset(consumer, stream_id, topic_id, partition_id, offset)
.await
}
+ ClientWrapper::New(client) => {
+ client
+ .store_consumer_offset(consumer, stream_id, topic_id, partition_id, offset)
+ .await
+ }
+
}
}
@@ -83,6 +89,11 @@ impl ConsumerOffsetClient for ClientWrapper {
.get_consumer_offset(consumer, stream_id, topic_id, partition_id)
.await
}
+ ClientWrapper::New(client) => {
+ client
+ .get_consumer_offset(consumer, stream_id, topic_id, partition_id)
+ .await
+ }
}
}
@@ -114,6 +125,11 @@ impl ConsumerOffsetClient for ClientWrapper {
.delete_consumer_offset(consumer, stream_id, topic_id, partition_id)
.await
}
+ ClientWrapper::New(client) => {
+ client
+ .delete_consumer_offset(consumer, stream_id, topic_id, partition_id)
+ .await
+ }
}
}
}
diff --git a/core/sdk/src/client_wrappers/binary_message_client.rs b/core/sdk/src/client_wrappers/binary_message_client.rs
index 01e0c166..b357ea4b 100644
--- a/core/sdk/src/client_wrappers/binary_message_client.rs
+++ b/core/sdk/src/client_wrappers/binary_message_client.rs
@@ -88,6 +88,19 @@ impl MessageClient for ClientWrapper {
)
.await
}
+ ClientWrapper::New(client) => {
+ client
+ .poll_messages(
+ stream_id,
+ topic_id,
+ partition_id,
+ consumer,
+ strategy,
+ count,
+ auto_commit,
+ )
+ .await
+ }
}
}
@@ -119,6 +132,11 @@ impl MessageClient for ClientWrapper {
.send_messages(stream_id, topic_id, partitioning, messages)
.await
}
+ ClientWrapper::New(client) => {
+ client
+ .send_messages(stream_id, topic_id, partitioning, messages)
+ .await
+ }
}
}
@@ -150,6 +168,11 @@ impl MessageClient for ClientWrapper {
.flush_unsaved_buffer(stream_id, topic_id, partitioning_id, fsync)
.await
}
+ ClientWrapper::New(client) => {
+ client
+ .flush_unsaved_buffer(stream_id, topic_id, partitioning_id, fsync)
+ .await
+ }
}
}
}
diff --git a/core/sdk/src/client_wrappers/binary_partition_client.rs b/core/sdk/src/client_wrappers/binary_partition_client.rs
index 26ccad1d..1ff6d89d 100644
--- a/core/sdk/src/client_wrappers/binary_partition_client.rs
+++ b/core/sdk/src/client_wrappers/binary_partition_client.rs
@@ -50,6 +50,11 @@ impl PartitionClient for ClientWrapper {
.create_partitions(stream_id, topic_id, partitions_count)
.await
}
+ ClientWrapper::New(client) => {
+ client
+ .create_partitions(stream_id, topic_id, partitions_count)
+ .await
+ }
}
}
@@ -80,6 +85,11 @@ impl PartitionClient for ClientWrapper {
.delete_partitions(stream_id, topic_id, partitions_count)
.await
}
+ ClientWrapper::New(client) => {
+ client
+ .delete_partitions(stream_id, topic_id, partitions_count)
+ .await
+ }
}
}
}
diff --git a/core/sdk/src/client_wrappers/binary_personal_access_token_client.rs b/core/sdk/src/client_wrappers/binary_personal_access_token_client.rs
index 58c6fbed..21a68412 100644
--- a/core/sdk/src/client_wrappers/binary_personal_access_token_client.rs
+++ b/core/sdk/src/client_wrappers/binary_personal_access_token_client.rs
@@ -32,6 +32,7 @@ impl PersonalAccessTokenClient for ClientWrapper {
ClientWrapper::Http(client) => client.get_personal_access_tokens().await,
ClientWrapper::Tcp(client) => client.get_personal_access_tokens().await,
ClientWrapper::Quic(client) => client.get_personal_access_tokens().await,
+ ClientWrapper::New(client) => client.get_personal_access_tokens().await,
}
}
@@ -45,6 +46,7 @@ impl PersonalAccessTokenClient for ClientWrapper {
ClientWrapper::Http(client) => client.create_personal_access_token(name, expiry).await,
ClientWrapper::Tcp(client) => client.create_personal_access_token(name, expiry).await,
ClientWrapper::Quic(client) => client.create_personal_access_token(name, expiry).await,
+ ClientWrapper::New(client) => client.create_personal_access_token(name, expiry).await,
}
}
@@ -54,6 +56,7 @@ impl PersonalAccessTokenClient for ClientWrapper {
ClientWrapper::Http(client) => client.delete_personal_access_token(name).await,
ClientWrapper::Tcp(client) => client.delete_personal_access_token(name).await,
ClientWrapper::Quic(client) => client.delete_personal_access_token(name).await,
+ ClientWrapper::New(client) => client.delete_personal_access_token(name).await,
}
}
@@ -66,6 +69,7 @@ impl PersonalAccessTokenClient for ClientWrapper {
ClientWrapper::Http(client) => client.login_with_personal_access_token(token).await,
ClientWrapper::Tcp(client) => client.login_with_personal_access_token(token).await,
ClientWrapper::Quic(client) => client.login_with_personal_access_token(token).await,
+ ClientWrapper::New(client) => client.login_with_personal_access_token(token).await,
}
}
}
diff --git a/core/sdk/src/client_wrappers/binary_segment_client.rs b/core/sdk/src/client_wrappers/binary_segment_client.rs
index 0ec082ff..2c0c05ef 100644
--- a/core/sdk/src/client_wrappers/binary_segment_client.rs
+++ b/core/sdk/src/client_wrappers/binary_segment_client.rs
@@ -51,6 +51,11 @@ impl SegmentClient for ClientWrapper {
.delete_segments(stream_id, topic_id, partition_id, segments_count)
.await
}
+ ClientWrapper::New(client) => {
+ client
+ .delete_segments(stream_id, topic_id, partition_id, segments_count)
+ .await
+ }
}
}
}
diff --git a/core/sdk/src/client_wrappers/binary_stream_client.rs b/core/sdk/src/client_wrappers/binary_stream_client.rs
index 93d785dd..27e48fd9 100644
--- a/core/sdk/src/client_wrappers/binary_stream_client.rs
+++ b/core/sdk/src/client_wrappers/binary_stream_client.rs
@@ -29,6 +29,7 @@ impl StreamClient for ClientWrapper {
ClientWrapper::Http(client) => client.get_stream(stream_id).await,
ClientWrapper::Tcp(client) => client.get_stream(stream_id).await,
ClientWrapper::Quic(client) => client.get_stream(stream_id).await,
+ ClientWrapper::New(client) => client.get_stream(stream_id).await,
}
}
@@ -38,6 +39,7 @@ impl StreamClient for ClientWrapper {
ClientWrapper::Http(client) => client.get_streams().await,
ClientWrapper::Tcp(client) => client.get_streams().await,
ClientWrapper::Quic(client) => client.get_streams().await,
+ ClientWrapper::New(client) => client.get_streams().await,
}
}
@@ -51,6 +53,7 @@ impl StreamClient for ClientWrapper {
ClientWrapper::Http(client) => client.create_stream(name, stream_id).await,
ClientWrapper::Tcp(client) => client.create_stream(name, stream_id).await,
ClientWrapper::Quic(client) => client.create_stream(name, stream_id).await,
+ ClientWrapper::New(client) => client.create_stream(name, stream_id).await,
}
}
@@ -60,6 +63,7 @@ impl StreamClient for ClientWrapper {
ClientWrapper::Http(client) => client.update_stream(stream_id, name).await,
ClientWrapper::Tcp(client) => client.update_stream(stream_id, name).await,
ClientWrapper::Quic(client) => client.update_stream(stream_id, name).await,
+ ClientWrapper::New(client) => client.update_stream(stream_id, name).await,
}
}
@@ -69,6 +73,7 @@ impl StreamClient for ClientWrapper {
ClientWrapper::Http(client) => client.delete_stream(stream_id).await,
ClientWrapper::Tcp(client) => client.delete_stream(stream_id).await,
ClientWrapper::Quic(client) => client.delete_stream(stream_id).await,
+ ClientWrapper::New(client) => client.delete_stream(stream_id).await,
}
}
@@ -78,6 +83,7 @@ impl StreamClient for ClientWrapper {
ClientWrapper::Http(client) => client.purge_stream(stream_id).await,
ClientWrapper::Tcp(client) => client.purge_stream(stream_id).await,
ClientWrapper::Quic(client) => client.purge_stream(stream_id).await,
+ ClientWrapper::New(client) => client.purge_stream(stream_id).await,
}
}
}
diff --git a/core/sdk/src/client_wrappers/binary_system_client.rs b/core/sdk/src/client_wrappers/binary_system_client.rs
index 825dbd8e..12f5fd28 100644
--- a/core/sdk/src/client_wrappers/binary_system_client.rs
+++ b/core/sdk/src/client_wrappers/binary_system_client.rs
@@ -32,6 +32,7 @@ impl SystemClient for ClientWrapper {
ClientWrapper::Http(client) => client.get_stats().await,
ClientWrapper::Tcp(client) => client.get_stats().await,
ClientWrapper::Quic(client) => client.get_stats().await,
+ ClientWrapper::New(client) => client.get_stats().await,
}
}
@@ -41,6 +42,7 @@ impl SystemClient for ClientWrapper {
ClientWrapper::Http(client) => client.get_me().await,
ClientWrapper::Tcp(client) => client.get_me().await,
ClientWrapper::Quic(client) => client.get_me().await,
+ ClientWrapper::New(client) => client.get_me().await,
}
}
@@ -50,6 +52,7 @@ impl SystemClient for ClientWrapper {
ClientWrapper::Http(client) => client.get_client(client_id).await,
ClientWrapper::Tcp(client) => client.get_client(client_id).await,
ClientWrapper::Quic(client) => client.get_client(client_id).await,
+ ClientWrapper::New(client) => client.get_client(client_id).await,
}
}
@@ -59,6 +62,7 @@ impl SystemClient for ClientWrapper {
ClientWrapper::Http(client) => client.get_clients().await,
ClientWrapper::Tcp(client) => client.get_clients().await,
ClientWrapper::Quic(client) => client.get_clients().await,
+ ClientWrapper::New(client) => client.get_clients().await,
}
}
@@ -68,6 +72,7 @@ impl SystemClient for ClientWrapper {
ClientWrapper::Http(client) => client.ping().await,
ClientWrapper::Tcp(client) => client.ping().await,
ClientWrapper::Quic(client) => client.ping().await,
+ ClientWrapper::New(client) => client.ping().await,
}
}
@@ -77,6 +82,7 @@ impl SystemClient for ClientWrapper {
ClientWrapper::Http(client) => client.heartbeat_interval().await,
ClientWrapper::Tcp(client) => client.heartbeat_interval().await,
ClientWrapper::Quic(client) => client.heartbeat_interval().await,
+ ClientWrapper::New(client) => client.heartbeat_interval().await,
}
}
@@ -90,6 +96,7 @@ impl SystemClient for ClientWrapper {
ClientWrapper::Http(client) => client.snapshot(compression, snapshot_types).await,
ClientWrapper::Tcp(client) => client.snapshot(compression, snapshot_types).await,
ClientWrapper::Quic(client) => client.snapshot(compression, snapshot_types).await,
+ ClientWrapper::New(client) => client.snapshot(compression, snapshot_types).await,
}
}
}
diff --git a/core/sdk/src/client_wrappers/binary_topic_client.rs b/core/sdk/src/client_wrappers/binary_topic_client.rs
index c9685bdb..6a3de86e 100644
--- a/core/sdk/src/client_wrappers/binary_topic_client.rs
+++ b/core/sdk/src/client_wrappers/binary_topic_client.rs
@@ -35,6 +35,7 @@ impl TopicClient for ClientWrapper {
ClientWrapper::Http(client) => client.get_topic(stream_id, topic_id).await,
ClientWrapper::Tcp(client) => client.get_topic(stream_id, topic_id).await,
ClientWrapper::Quic(client) => client.get_topic(stream_id, topic_id).await,
+ ClientWrapper::New(client) => client.get_topic(stream_id, topic_id).await,
}
}
@@ -44,6 +45,7 @@ impl TopicClient for ClientWrapper {
ClientWrapper::Http(client) => client.get_topics(stream_id).await,
ClientWrapper::Tcp(client) => client.get_topics(stream_id).await,
ClientWrapper::Quic(client) => client.get_topics(stream_id).await,
+ ClientWrapper::New(client) => client.get_topics(stream_id).await,
}
}
@@ -115,6 +117,20 @@ impl TopicClient for ClientWrapper {
)
.await
}
+ ClientWrapper::New(client) => {
+ client
+ .create_topic(
+ stream_id,
+ name,
+ partitions_count,
+ compression_algorithm,
+ replication_factor,
+ topic_id,
+ message_expiry,
+ max_topic_size,
+ )
+ .await
+ }
}
}
@@ -181,6 +197,19 @@ impl TopicClient for ClientWrapper {
)
.await
}
+ ClientWrapper::New(client) => {
+ client
+ .update_topic(
+ stream_id,
+ topic_id,
+ name,
+ compression_algorithm,
+ replication_factor,
+ message_expiry,
+ max_topic_size,
+ )
+ .await
+ }
}
}
@@ -194,6 +223,7 @@ impl TopicClient for ClientWrapper {
ClientWrapper::Http(client) => client.delete_topic(stream_id, topic_id).await,
ClientWrapper::Tcp(client) => client.delete_topic(stream_id, topic_id).await,
ClientWrapper::Quic(client) => client.delete_topic(stream_id, topic_id).await,
+ ClientWrapper::New(client) => client.delete_topic(stream_id, topic_id).await,
}
}
@@ -207,6 +237,7 @@ impl TopicClient for ClientWrapper {
ClientWrapper::Http(client) => client.purge_topic(stream_id, topic_id).await,
ClientWrapper::Tcp(client) => client.purge_topic(stream_id, topic_id).await,
ClientWrapper::Quic(client) => client.purge_topic(stream_id, topic_id).await,
+ ClientWrapper::New(client) => client.purge_topic(stream_id, topic_id).await,
}
}
}
diff --git a/core/sdk/src/client_wrappers/binary_user_client.rs b/core/sdk/src/client_wrappers/binary_user_client.rs
index f7289c24..4f0ef5e4 100644
--- a/core/sdk/src/client_wrappers/binary_user_client.rs
+++ b/core/sdk/src/client_wrappers/binary_user_client.rs
@@ -31,6 +31,7 @@ impl UserClient for ClientWrapper {
ClientWrapper::Http(client) => client.get_user(user_id).await,
ClientWrapper::Tcp(client) => client.get_user(user_id).await,
ClientWrapper::Quic(client) => client.get_user(user_id).await,
+ ClientWrapper::New(client) => client.get_user(user_id).await,
}
}
@@ -40,6 +41,7 @@ impl UserClient for ClientWrapper {
ClientWrapper::Http(client) => client.get_users().await,
ClientWrapper::Tcp(client) => client.get_users().await,
ClientWrapper::Quic(client) => client.get_users().await,
+ ClientWrapper::New(client) => client.get_users().await,
}
}
@@ -71,6 +73,11 @@ impl UserClient for ClientWrapper {
.create_user(username, password, status, permissions)
.await
}
+ ClientWrapper::New(client) => {
+ client
+ .create_user(username, password, status, permissions)
+ .await
+ }
}
}
@@ -80,6 +87,7 @@ impl UserClient for ClientWrapper {
ClientWrapper::Tcp(client) => client.delete_user(user_id).await,
ClientWrapper::Quic(client) => client.delete_user(user_id).await,
ClientWrapper::Iggy(client) => client.delete_user(user_id).await,
+ ClientWrapper::New(client) => client.delete_user(user_id).await,
}
}
@@ -94,6 +102,7 @@ impl UserClient for ClientWrapper {
ClientWrapper::Tcp(client) => client.update_user(user_id, username, status).await,
ClientWrapper::Quic(client) => client.update_user(user_id, username, status).await,
ClientWrapper::Iggy(client) => client.update_user(user_id, username, status).await,
+ ClientWrapper::New(client) => client.update_user(user_id, username, status).await,
}
}
@@ -107,6 +116,7 @@ impl UserClient for ClientWrapper {
ClientWrapper::Http(client) => client.update_permissions(user_id, permissions).await,
ClientWrapper::Tcp(client) => client.update_permissions(user_id, permissions).await,
ClientWrapper::Quic(client) => client.update_permissions(user_id, permissions).await,
+ ClientWrapper::New(client) => client.update_permissions(user_id, permissions).await,
}
}
@@ -137,6 +147,11 @@ impl UserClient for ClientWrapper {
.change_password(user_id, current_password, new_password)
.await
}
+ ClientWrapper::New(client) => {
+ client
+ .change_password(user_id, current_password, new_password)
+ .await
+ }
}
}
@@ -146,6 +161,7 @@ impl UserClient for ClientWrapper {
ClientWrapper::Http(client) => client.login_user(username, password).await,
ClientWrapper::Tcp(client) => client.login_user(username, password).await,
ClientWrapper::Quic(client) => client.login_user(username, password).await,
+ ClientWrapper::New(client) => client.login_user(username, password).await,
}
}
@@ -155,6 +171,7 @@ impl UserClient for ClientWrapper {
ClientWrapper::Http(client) => client.logout_user().await,
ClientWrapper::Tcp(client) => client.logout_user().await,
ClientWrapper::Quic(client) => client.logout_user().await,
+ ClientWrapper::New(client) => client.logout_user().await,
}
}
}
diff --git a/core/sdk/src/client_wrappers/client_wrapper.rs b/core/sdk/src/client_wrappers/client_wrapper.rs
index d391bfda..31af756e 100644
--- a/core/sdk/src/client_wrappers/client_wrapper.rs
+++ b/core/sdk/src/client_wrappers/client_wrapper.rs
@@ -17,6 +17,7 @@
*/
use crate::clients::client::IggyClient;
+use crate::connection::{NewTcpClient, NewTokioTcpClient};
use crate::http::http_client::HttpClient;
use crate::quic::quic_client::QuicClient;
use crate::tcp::tcp_client::TcpClient;
@@ -28,4 +29,5 @@ pub enum ClientWrapper {
Http(HttpClient),
Tcp(TcpClient),
Quic(QuicClient),
+ New(NewTokioTcpClient),
}
diff --git a/core/sdk/src/connection/mod.rs b/core/sdk/src/connection/mod.rs
new file mode 100644
index 00000000..f35c3c5b
--- /dev/null
+++ b/core/sdk/src/connection/mod.rs
@@ -0,0 +1,765 @@
+use std::{
+ collections::{HashMap, VecDeque},
+ fmt::Debug,
+ io::{self, IoSlice},
+ mem::MaybeUninit,
+ net::SocketAddr,
+ ops::Deref,
+ pin::Pin,
+ str::FromStr,
+ sync::{
+ Arc,
+ atomic::{AtomicU64, Ordering},
+ },
+ task::{Context, Poll, Waker, ready},
+};
+
+use crate::protocol::{ControlAction, ProtocolCore, ProtocolCoreConfig, Response, TxBuf};
+use async_broadcast::{Receiver, Sender, broadcast};
+use async_trait::async_trait;
+use bytes::{Buf, BufMut, Bytes, BytesMut};
+use futures::{AsyncRead, AsyncWrite, task::AtomicWaker};
+use iggy_binary_protocol::{BinaryClient, BinaryTransport, Client};
+use iggy_common::{
+ AutoLogin, ClientState, Command, Credentials, DiagnosticEvent, IggyDuration, IggyError,
+ TcpClientConfig,
+};
+use parking_lot::Mutex;
+use rustc_hash::FxHashMap;
+use tokio::{
+ net::TcpStream,
+ time::{Sleep, sleep},
+};
+use tokio_util::compat::{Compat, TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};
+use tracing::{debug, error, info, trace, warn};
+
+pub type SFut<S> = Pin<Box<dyn Future<Output = io::Result<S>> + Send>>;
+pub type StreamFactory<S> = Arc<dyn Fn(SocketAddr) -> SFut<S> + Send + Sync>;
+
+pub async fn tokio_tcp(addr: SocketAddr) -> io::Result<Compat<TcpStream>> {
+ let socket = match addr {
+ SocketAddr::V4(_) => tokio::net::TcpSocket::new_v4()?,
+ SocketAddr::V6(_) => tokio::net::TcpSocket::new_v6()?,
+ };
+
+ socket.set_nodelay(true)?;
+ let s = socket.connect(addr).await?;
+ Ok(s.compat())
+}
+
+pub type TokioCompat = Compat<TcpStream>;
+pub type NewTokioTcpClient = NewTcpClient<TokioCompat>;
+
+pub enum ClientCommand {
+ Connect(SocketAddr),
+ Disconnect,
+ Shutdown,
+}
+
+pub trait AsyncIO: AsyncWrite + AsyncRead + Unpin {}
+impl<T: AsyncRead + AsyncWrite + Unpin> AsyncIO for T {}
+
+pub trait Runtime: Send + Sync + Debug {
+ fn spawn(&self, future: Pin<Box<dyn Future<Output = ()> + Send>>);
+}
+
+#[derive(Debug)]
+pub struct TokioRuntime {}
+
+impl Runtime for TokioRuntime {
+ fn spawn(&self, future: Pin<Box<dyn Future<Output = ()> + Send>>) {
+ tokio::spawn(future);
+ }
+}
+
+#[derive(Debug)]
+pub struct ConnectionInner<S: AsyncIO> {
+ pub(crate) state: Mutex<State<S>>,
+}
+
+#[derive(Debug)]
+pub struct ConnectionRef<S: AsyncIO>(Arc<ConnectionInner<S>>);
+
+impl<S: AsyncIO> ConnectionRef<S> {
+ fn new(core: ProtocolCore, stream_factory: StreamFactory<S>) -> Self {
+ Self(Arc::new(ConnectionInner {
+ state: Mutex::new(State {
+ inner: core,
+ driver: None,
+ stream_factory,
+ stream: None,
+ current_send: None,
+ send_offset: 0,
+ recv_buffer: BytesMut::with_capacity(16 * 1024),
+ wait_timer: None,
+ waiters: Arc::new(Waiters {
+ map: Mutex::new(FxHashMap::with_capacity_and_hasher(256, Default::default())),
+ next_id: AtomicU64::new(0),
+ }),
+ requests_to_wait: FxHashMap::with_capacity_and_hasher(256, Default::default()),
+ pending_commands: VecDeque::new(),
+ connect_waiters: Vec::new(),
+ pending_connect: None,
+ }),
+ }))
+ }
+}
+
+impl<S: AsyncIO> ConnectionRef<S> {
+ fn state(&self) -> ClientState {
+ let state = self.0.state.lock();
+ state.inner.state
+ }
+
+ fn set_state(&self, client_state: ClientState) {
+ let mut state = self.0.state.lock();
+ state.inner.state = client_state
+ }
+}
+
+impl<S: AsyncIO> Deref for ConnectionRef<S> {
+ type Target = ConnectionInner<S>;
+ fn deref(&self) -> &Self::Target {
+ &self.0
+ }
+}
+
+impl<S: AsyncIO> Clone for ConnectionRef<S> {
+ fn clone(&self) -> Self {
+ Self(self.0.clone())
+ }
+}
+
+struct WaitEntry<T> {
+ waker: AtomicWaker,
+ result: Option<T>,
+}
+
+struct Waiters<T> {
+ map: Mutex<FxHashMap<u64, WaitEntry<T>>>,
+ next_id: AtomicU64,
+}
+
+impl<T> Waiters<T> {
+ fn alloc(&self) -> u64 {
+ let id = self.next_id.fetch_add(1, Ordering::Relaxed);
+ self.map.lock().insert(
+ id,
+ WaitEntry {
+ waker: AtomicWaker::new(),
+ result: None,
+ },
+ );
+ id
+ }
+
+ fn complete(&self, id: u64, val: T) -> bool {
+ if let Some(entry) = self.map.lock().get_mut(&id) {
+ entry.result = Some(val);
+ entry.waker.wake();
+ true
+ } else {
+ false
+ }
+ }
+}
+
+struct WaitFuture<T> {
+ waiters: Arc<Waiters<T>>,
+ id: u64,
+}
+
+impl<T> Future for WaitFuture<T> {
+ type Output = T;
+
+ fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<T> {
+ let mut map = self.waiters.map.lock();
+
+ let ready: Option<T> = {
+ if let Some(entry) = map.get_mut(&self.id) {
+ if let Some(val) = entry.result.take() {
+ Some(val)
+ } else {
+ entry.waker.register(cx.waker());
+ entry.result.take()
+ }
+ } else {
+ None
+ }
+ };
+
+ if let Some(val) = ready {
+ map.remove(&self.id);
+ Poll::Ready(val)
+ } else {
+ Poll::Pending
+ }
+ }
+}
+
+pub struct State<S: AsyncIO> {
+ inner: ProtocolCore,
+ driver: Option<Waker>,
+ stream_factory: StreamFactory<S>,
+ stream: Option<S>,
+ current_send: Option<TxBuf>,
+ send_offset: usize,
+ recv_buffer: BytesMut,
+
+ wait_timer: Option<Pin<Box<Sleep>>>,
+
+ waiters: Arc<Waiters<Result<Bytes, IggyError>>>,
+ requests_to_wait: FxHashMap<u64, u64>,
+ pending_commands: VecDeque<(u64, ClientCommand)>,
+ connect_waiters: Vec<u64>,
+ pending_connect: Option<Pin<Box<dyn Future<Output = io::Result<S>> + Send>>>,
+}
+
+impl<S> Debug for State<S>
+where
+ S: AsyncIO,
+{
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ write!(f, "test")
+ }
+}
+
+impl<S> State<S>
+where
+ S: AsyncIO,
+{
+ fn take_waker(&mut self) -> Option<Waker> {
+ self.driver.take()
+ }
+
+ fn complete_all_waiters_with_error(&mut self, error: IggyError) {
+ for id in self.connect_waiters.drain(..) {
+ let _ = self.waiters.complete(id, Err(error.clone()));
+ }
+
+ for (_request_id, wait_id) in self.requests_to_wait.drain() {
+ let _ = self.waiters.complete(wait_id, Err(error.clone()));
+ }
+
+ for (wait_id, _cmd) in self.pending_commands.drain(..) {
+ let _ = self.waiters.complete(wait_id, Err(error.clone()));
+ }
+ }
+
+ fn enqueu_command(
+ &mut self,
+ command: ClientCommand,
+ ) -> (WaitFuture<Result<Bytes, IggyError>>, Option<Waker>) {
+ let id = self.waiters.alloc();
+ self.pending_commands.push_back((id, command));
+ let waker = self.take_waker();
+ (
+ WaitFuture {
+ waiters: self.waiters.clone(),
+ id,
+ },
+ waker,
+ )
+ }
+
+ fn enqueue_message(
+ &mut self,
+ code: u32,
+ payload: Bytes,
+ ) -> (WaitFuture<Result<Bytes, IggyError>>, Option<Waker>) {
+ let wait_id = self.waiters.alloc();
+ let waker = match self.inner.send(code, payload) {
+ Ok(protocol_id) => {
+ self.requests_to_wait.insert(protocol_id, wait_id);
+ self.take_waker()
+ }
+ Err(e) => {
+ self.waiters.complete(wait_id, Err(e));
+ None
+ }
+ };
+ (
+ WaitFuture {
+ waiters: self.waiters.clone(),
+ id: wait_id,
+ },
+ waker,
+ )
+ }
+
+ fn drive_client_commands(&mut self) -> io::Result<bool> {
+ let mut made_progress = false;
+ for (request_id, cmd) in self.pending_commands.drain(..) {
+ made_progress = true;
+ match cmd {
+ ClientCommand::Connect(server_address) => {
+ tracing::debug!(
+ "ConnectionDriver: Processing Connect command to {}",
+ server_address
+ );
+ let current_state = self.inner.state;
+
+ if matches!(
+ current_state,
+ ClientState::Connected
+ | ClientState::Authenticating
+ | ClientState::Authenticated
+ ) {
+ tracing::debug!(
+ "ConnectionDriver: Already connected (state: {:?}), completing waiter immediately",
+ current_state
+ );
+ let _ = self.waiters.complete(request_id, Ok(Bytes::new()));
+ continue;
+ }
+
+ if matches!(current_state, ClientState::Connecting) {
+ tracing::debug!("ConnectionDriver: Already connecting, adding to waiters");
+ self.connect_waiters.push(request_id);
+ continue;
+ }
+
+ self.connect_waiters.push(request_id);
+ self.inner.desire_connect(server_address).map_err(|e| {
+ tracing::error!(
+ "ConnectionDriver: desire_connect failed: {}",
+ e.as_string()
+ );
+ io::Error::new(io::ErrorKind::ConnectionAborted, e.as_string())
+ })?;
+ tracing::debug!(
+ "ConnectionDriver: desire_connect successful, state: {:?}",
+ self.inner.state
+ );
+ }
+ ClientCommand::Disconnect => {
+ self.inner.disconnect();
+ self.waiters.complete(request_id, Ok(Bytes::new()));
+ }
+ ClientCommand::Shutdown => {
+ self.inner.shutdown();
+ self.waiters.complete(request_id, Ok(Bytes::new()));
+ }
+ }
+ }
+ Ok(made_progress)
+ }
+
+ fn drive_connect(&mut self, cx: &mut Context<'_>) -> io::Result<bool> {
+ if let Some(fut) = self.pending_connect.as_mut() {
+ match fut.as_mut().poll(cx) {
+ Poll::Pending => return Ok(false),
+ Poll::Ready(Ok(stream)) => {
+ self.stream = Some(stream);
+ self.pending_connect = None;
+ self.inner.on_connected().map_err(|e| {
+ io::Error::new(io::ErrorKind::ConnectionRefused, e.as_string())
+ })?;
+ tracing::debug!(
+ "ConnectionDriver: Connection established, completing {} waiters",
+ self.connect_waiters.len()
+ );
+ if !self.inner.should_wait_auth() {
+ for id in self.connect_waiters.drain(..) {
+ tracing::debug!("ConnectionDriver: Completing connect waiter {}", id);
+ let _ = self.waiters.complete(id, Ok(Bytes::new()));
+ }
+ }
+ return Ok(true);
+ }
+ Poll::Ready(Err(_e)) => {
+ self.pending_connect = None;
+ self.inner.disconnect();
+ for id in self.connect_waiters.drain(..) {
+ let _ = self
+ .waiters
+ .complete(id, Err(IggyError::CannotEstablishConnection));
+ }
+ return Ok(true);
+ }
+ }
+ }
+ Ok(false)
+ }
+
+ fn drive_timer(&mut self, cx: &mut Context<'_>) -> bool {
+ if let Some(t) = &mut self.wait_timer {
+ if t.as_mut().poll(cx).is_pending() {
+ return false;
+ }
+ self.wait_timer = None;
+ return true;
+ }
+ false
+ }
+
+ fn drive_transmit(&mut self, cx: &mut Context<'_>) -> io::Result<bool> {
+ if self.current_send.is_none() {
+ if let Some(tx) = self.inner.poll_transmit() {
+ self.current_send = Some(tx);
+ self.send_offset = 0;
+ } else {
+ return Ok(false);
+ }
+ }
+
+ let stream = self
+ .stream
+ .as_mut()
+ .ok_or_else(|| io::Error::new(io::ErrorKind::NotConnected, "No stream"))?;
+
+ let buf = self.current_send.as_ref().unwrap();
+ let mut offset = self.send_offset;
+
+ while offset < buf.total_len() {
+ let mut storage = [IoSlice::new(&[]), IoSlice::new(&[])];
+
+ let iov = if offset < 8 {
+ storage[0] = IoSlice::new(&buf.header[offset..]);
+ if !buf.payload.is_empty() {
+ storage[1] = IoSlice::new(&buf.payload);
+ &storage[..2]
+ } else {
+ &storage[..1]
+ }
+ } else {
+ let body_off = offset - 8;
+ storage[0] = IoSlice::new(&buf.payload[body_off..]);
+ &storage[..1]
+ };
+
+ let written = match Pin::new(&mut *stream).poll_write_vectored(cx, iov)? {
+ Poll::Ready(0) => {
+ return Err(io::Error::new(
+ io::ErrorKind::WriteZero,
+ "write returned 0 bytes",
+ ));
+ }
+ Poll::Ready(n) => n,
+ Poll::Pending => return Ok(false),
+ };
+
+ offset += written;
+ self.send_offset += written;
+ }
+
+ match Pin::new(stream).poll_flush(cx)? {
+ Poll::Pending => return Ok(false),
+ Poll::Ready(()) => {}
+ }
+
+ self.send_offset = 0;
+ self.current_send = None;
+ Ok(true)
+ }
+
+ fn drive_receive(&mut self, cx: &mut Context<'_>) -> io::Result<bool> {
+ let mut progress = false;
+
+ // todo add some const var
+ for _ in 0..16 {
+ if self.recv_buffer.spare_capacity_mut().is_empty() {
+ self.recv_buffer.reserve(8192);
+ }
+
+ let spare: &mut [MaybeUninit<u8>] = self.recv_buffer.spare_capacity_mut();
+
+ let buf: &mut [u8] = unsafe { &mut *(spare as *mut [MaybeUninit<u8>] as *mut [u8]) };
+
+ let n = {
+ let stream = self
+ .stream
+ .as_mut()
+ .ok_or(io::Error::new(io::ErrorKind::NotConnected, "No stream"))?;
+ match Pin::new(&mut *stream).poll_read(cx, buf)? {
+ Poll::Pending => return Ok(progress),
+ Poll::Ready(0) => {
+ self.inner.disconnect();
+ self.stream = None;
+ self.complete_all_waiters_with_error(IggyError::CannotEstablishConnection);
+ return Ok(true);
+ }
+ Poll::Ready(n) => n,
+ }
+ };
+
+ unsafe {
+ self.recv_buffer.advance_mut(n);
+ }
+
+ self.inner
+ .process_incoming_with(&mut self.recv_buffer, |req_id, status, payload| {
+ if let Some(wait_id) = self.requests_to_wait.remove(&req_id) {
+ let res = if status == 0 {
+ Ok(payload)
+ } else {
+ Err(IggyError::from_code(status))
+ };
+ let _ = self.waiters.complete(wait_id, res);
+ }
+ });
+ progress = true;
+ }
+
+ if let Some(auth_res) = self.inner.take_auth_result() {
+ match auth_res {
+ Ok(()) => {
+ for id in self.connect_waiters.drain(..) {
+ let _ = self.waiters.complete(id, Ok(Bytes::new()));
+ }
+ }
+ Err(e) => {
+ for id in self.connect_waiters.drain(..) {
+ let _ = self.waiters.complete(id, Err(e.clone()));
+ }
+ }
+ }
+ progress = true;
+ }
+
+ Ok(progress)
+ }
+}
+
+struct ConnectionDriver<S: AsyncIO>(ConnectionRef<S>);
+
+impl<S: AsyncIO> Future for ConnectionDriver<S> {
+ type Output = Result<(), io::Error>;
+
+ fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+ let st = &mut *self.0.state.lock();
+ let mut keep_going = false;
+
+ keep_going |= st.drive_client_commands()?;
+
+ let order = st.inner.poll();
+ match order {
+ ControlAction::Wait(dur) => {
+ if st.wait_timer.is_none() {
+ st.wait_timer = Some(Box::pin(tokio::time::sleep(dur.get_duration())));
+ }
+ keep_going |= st.drive_timer(cx);
+ }
+ ControlAction::Connect(server_adress) => {
+ if st.pending_connect.is_none() {
+ st.pending_connect = Some((st.stream_factory)(server_adress));
+ }
+ keep_going |= st.drive_connect(cx)?;
+ }
+ ControlAction::Noop | ControlAction::Authenticate { .. } => {}
+ ControlAction::Error(e) => {
+ if matches!(e, IggyError::ClientShutdown) {
+ tracing::debug!(
+ "ConnectionDriver: Received ClientShutdown, terminating gracefully"
+ );
+ return Poll::Ready(Ok(()));
+ }
+ tracing::error!("ConnectionDriver: Error - {e:?}");
+ return Poll::Ready(Err(io::Error::new(io::ErrorKind::Other, format!("{e:?}"))));
+ }
+ }
+
+ if st.stream.is_some() {
+ match st.drive_transmit(cx) {
+ Ok(progress) => keep_going |= progress,
+ Err(e) => {
+ tracing::debug!("Transmit error, disconnecting: {:?}", e);
+ st.inner.disconnect();
+ st.stream = None;
+ st.complete_all_waiters_with_error(IggyError::CannotEstablishConnection);
+ keep_going = true;
+ }
+ }
+
+ if st.stream.is_some() {
+ match st.drive_receive(cx) {
+ Ok(progress) => keep_going |= progress,
+ Err(e) => {
+ tracing::debug!("Receive error, disconnecting: {:?}", e);
+ st.inner.disconnect();
+ st.stream = None;
+ st.complete_all_waiters_with_error(IggyError::CannotEstablishConnection);
+ keep_going = true;
+ }
+ }
+ }
+ }
+
+ let current_state = st.inner.state;
+ if current_state == ClientState::Shutdown {
+ if st.stream.is_some() {
+ tracing::debug!("ConnectionDriver: Closing stream due to Shutdown");
+ st.stream = None;
+ }
+ return Poll::Ready(Ok(()));
+ }
+
+ if current_state == ClientState::Disconnected && st.stream.is_some() {
+ tracing::debug!(
+ "ConnectionDriver: Closing stream due to Disconnected but keeping driver alive"
+ );
+ st.stream = None;
+ }
+
+ st.driver = Some(cx.waker().clone());
+
+ if keep_going {
+ cx.waker().wake_by_ref();
+ }
+
+ Poll::Pending
+ }
+}
+
+#[derive(Debug)]
+pub struct NewTcpClient<S: AsyncIO> {
+ state: ConnectionRef<S>,
+ config: Arc<TcpClientConfig>,
+ events: (Sender<DiagnosticEvent>, Receiver<DiagnosticEvent>),
+ _driver_handle: tokio::task::JoinHandle<()>,
+}
+
+impl<S: AsyncIO + Send + Sync + 'static> NewTcpClient<S> {
+ pub fn create(
+ config: Arc<TcpClientConfig>,
+ factory: StreamFactory<S>,
+ ) -> Result<Self, IggyError> {
+ let (tx, rx) = broadcast(1000);
+
+ let proto_config = ProtocolCoreConfig {
+ auto_login: config.auto_login.clone(),
+ reestablish_after: config.reconnection.reestablish_after,
+ max_retries: config.reconnection.max_retries,
+ };
+
+ let conn = ConnectionRef::new(ProtocolCore::new(proto_config), factory);
+ let driver = ConnectionDriver(conn.clone());
+ let driver_handle = tokio::spawn(async move {
+ if let Err(e) = driver.await {
+ tracing::error!("I/O error: {e}");
+ }
+ });
+
+ Ok(Self {
+ state: conn,
+ config,
+ events: (tx, rx),
+ _driver_handle: driver_handle,
+ })
+ }
+
+ async fn send_raw(&self, code: u32, payload: Bytes) -> Result<Bytes, IggyError> {
+ let (wait_future, waker) = {
+ let mut state = self.state.0.state.lock();
+ state.enqueue_message(code, payload)
+ };
+ if let Some(waker) = waker {
+ waker.wake();
+ }
+ wait_future.await
+ }
+}
+
+#[async_trait]
+impl<S: AsyncIO + Debug + Send + Sync + 'static> Client for NewTcpClient<S> {
+ async fn connect(&self) -> Result<(), IggyError> {
+ let address = SocketAddr::from_str(&self.config.server_address).unwrap();
+ let (fut, waker) = {
+ let mut state = self.state.0.state.lock();
+ state.enqueu_command(ClientCommand::Connect(address))
+ };
+ if let Some(waker) = waker {
+ waker.wake();
+ }
+
+ match fut.await {
+ Ok(_) => {
+ self.publish_event(DiagnosticEvent::Connected).await;
+ Ok(())
+ }
+ Err(IggyError::CannotEstablishConnection) => {
+ self.publish_event(DiagnosticEvent::Disconnected).await;
+ Err(IggyError::CannotEstablishConnection)
+ }
+ Err(e) => {
+ error!("Got error: {e} on connect");
+ Err(e)
+ }
+ }
+ }
+
+ async fn disconnect(&self) -> Result<(), IggyError> {
+ let (fut, waker) = {
+ let mut state = self.state.0.state.lock();
+ state.enqueu_command(ClientCommand::Disconnect)
+ };
+ if let Some(waker) = waker {
+ waker.wake();
+ }
+ fut.await?;
+ self.publish_event(DiagnosticEvent::Disconnected).await;
+ Ok(())
+ }
+
+ async fn shutdown(&self) -> Result<(), IggyError> {
+ let (fut, waker) = {
+ let mut state = self.state.0.state.lock();
+ state.enqueu_command(ClientCommand::Shutdown)
+ };
+ if let Some(waker) = waker {
+ waker.wake();
+ }
+ fut.await?;
+ self.publish_event(DiagnosticEvent::Shutdown).await;
+ Ok(())
+ }
+
+ async fn subscribe_events(&self) -> Receiver<DiagnosticEvent> {
+ self.events.1.clone()
+ }
+}
+
+#[async_trait]
+impl<S: AsyncIO + Debug + Send + Sync + 'static> BinaryTransport for NewTcpClient<S> {
+ async fn get_state(&self) -> ClientState {
+ self.state.state()
+ }
+
+ async fn set_state(&self, state: ClientState) {
+ self.state.set_state(state);
+ }
+
+ async fn publish_event(&self, event: DiagnosticEvent) {
+ if let Err(error) = self.events.0.broadcast(event).await {
+ error!("Failed to send a TCP diagnostic event: {error}");
+ }
+ }
+
+ async fn send_with_response<T: Command>(&self, command: &T) -> Result<Bytes, IggyError> {
+ command.validate()?;
+ self.send_raw_with_response(command.code(), command.to_bytes())
+ .await
+ }
+
+ async fn send_raw_with_response(&self, code: u32, payload: Bytes) -> Result<Bytes, IggyError> {
+ self.send_raw(code, payload).await
+ }
+
+ fn get_heartbeat_interval(&self) -> IggyDuration {
+ self.config.heartbeat_interval
+ }
+}
+
+impl<S: AsyncIO + Debug + Send + Sync + 'static> BinaryClient for NewTcpClient<S> {}
+
+impl<S: AsyncIO> Drop for NewTcpClient<S> {
+ fn drop(&mut self) {
+ let mut state = self.state.0.state.lock();
+ state.inner.disconnect();
+ let waker = state.take_waker();
+ drop(state);
+ if let Some(waker) = waker {
+ waker.wake();
+ }
+ }
+}
diff --git a/core/sdk/src/lib.rs b/core/sdk/src/lib.rs
index c315fd5e..99eb68c8 100644
--- a/core/sdk/src/lib.rs
+++ b/core/sdk/src/lib.rs
@@ -26,3 +26,5 @@ pub mod prelude;
pub mod quic;
pub mod stream_builder;
pub mod tcp;
+pub mod protocol;
+pub mod connection;
\ No newline at end of file
diff --git a/core/sdk/src/protocol/mod.rs b/core/sdk/src/protocol/mod.rs
new file mode 100644
index 00000000..d6187e4f
--- /dev/null
+++ b/core/sdk/src/protocol/mod.rs
@@ -0,0 +1,289 @@
+use std::{
+ collections::VecDeque,
+ io::IoSlice,
+ net::SocketAddr,
+ sync::Arc,
+ time::{Duration, Instant},
+};
+
+use bytes::{Buf, BufMut, Bytes, BytesMut};
+use iggy_common::{AutoLogin, ClientState, Credentials, IggyDuration, IggyError, IggyTimestamp};
+use tracing::{debug, info, warn};
+
+const REQUEST_INITIAL_BYTES_LENGTH: usize = 4;
+const REQUEST_HEADER_BYTES: usize = 8;
+const RESPONSE_INITIAL_BYTES_LENGTH: usize = 8;
+#[derive(Debug)]
+pub struct ProtocolCoreConfig {
+ pub auto_login: AutoLogin,
+ pub reestablish_after: IggyDuration,
+ pub max_retries: Option<u32>,
+}
+
+#[derive(Debug)]
+pub enum ControlAction {
+ Connect(SocketAddr),
+ Wait(IggyDuration),
+ Authenticate { username: String, password: String },
+ Noop,
+ Error(IggyError),
+}
+
+pub struct TxBuf {
+ pub header: [u8; 8],
+ pub payload: Bytes,
+ pub request_id: u64,
+}
+
+impl TxBuf {
+ #[inline]
+ pub fn total_len(&self) -> usize {
+ REQUEST_HEADER_BYTES + self.payload.len()
+ }
+}
+
+#[derive(Debug)]
+pub struct Response {
+ pub status: u32,
+ pub payload: Bytes,
+}
+
+#[derive(Debug)]
+pub struct ProtocolCore {
+ pub state: ClientState,
+ config: ProtocolCoreConfig,
+ last_connect_attempt: Option<IggyTimestamp>,
+ pub retry_count: u32,
+ next_request_id: u64,
+ pending_sends: VecDeque<(u32, Bytes, u64)>,
+ sent_order: VecDeque<u64>,
+ auth_pending: bool,
+ auth_request_id: Option<u64>,
+ server_address: Option<SocketAddr>,
+ last_auth_result: Option<Result<(), IggyError>>,
+}
+
+impl ProtocolCore {
+ pub fn new(config: ProtocolCoreConfig) -> Self {
+ Self {
+ state: ClientState::Disconnected,
+ config,
+ last_connect_attempt: None,
+ retry_count: 0,
+ next_request_id: 1,
+ pending_sends: VecDeque::new(),
+ sent_order: VecDeque::new(),
+ auth_pending: false,
+ auth_request_id: None,
+ server_address: None,
+ last_auth_result: None,
+ }
+ }
+
+ pub fn poll_transmit(&mut self) -> Option<TxBuf> {
+ if let Some((code, payload, request_id)) = self.pending_sends.pop_front() {
+ let total_len = (payload.len() + REQUEST_INITIAL_BYTES_LENGTH) as u32;
+ self.sent_order.push_back(request_id);
+
+ Some(TxBuf {
+ payload,
+ header: make_header(total_len, code),
+ request_id,
+ })
+ } else {
+ None
+ }
+ }
+
+ pub fn send(&mut self, code: u32, payload: Bytes) -> Result<u64, IggyError> {
+ match self.state {
+ ClientState::Shutdown => Err(IggyError::ClientShutdown),
+ ClientState::Disconnected | ClientState::Connecting => Err(IggyError::NotConnected),
+ ClientState::Connected | ClientState::Authenticating | ClientState::Authenticated => {
+ Ok(self.queue_send(code, payload))
+ }
+ }
+ }
+
+ fn queue_send(&mut self, code: u32, payload: Bytes) -> u64 {
+ let request_id = self.next_request_id;
+ self.next_request_id += 1;
+ self.pending_sends.push_back((code, payload, request_id));
+ request_id
+ }
+
+ pub fn process_incoming_with<F: FnMut(u64, u32, Bytes)>(
+ &mut self,
+ buf: &mut BytesMut,
+ mut f: F,
+ ) {
+ loop {
+ if buf.len() < RESPONSE_INITIAL_BYTES_LENGTH {
+ break;
+ }
+ let status = u32::from_le_bytes(buf[..4].try_into().unwrap());
+ let length = u32::from_le_bytes(buf[4..8].try_into().unwrap());
+ let total = RESPONSE_INITIAL_BYTES_LENGTH + length as usize;
+ if buf.len() < total {
+ break;
+ }
+
+ buf.advance(RESPONSE_INITIAL_BYTES_LENGTH);
+ let payload = if length <= 1 {
+ Bytes::new()
+ } else {
+ buf.split_to(length as usize).freeze()
+ };
+ if let Some(id) = self.on_response(status) {
+ f(id, status, payload);
+ }
+ }
+ }
+
+ pub fn on_response(&mut self, status: u32) -> Option<u64> {
+ let request_id = self.sent_order.pop_front()?;
+
+ if Some(request_id) == self.auth_request_id {
+ if status == 0 {
+ debug!("Authentication successful");
+ self.state = ClientState::Authenticated;
+ self.auth_pending = false;
+ self.last_auth_result = Some(Ok(()));
+ } else {
+ warn!("Authentication failed with status: {}", status);
+ self.state = ClientState::Connected;
+ self.auth_pending = false;
+ self.last_auth_result = Some(Err(IggyError::Unauthenticated));
+ }
+ self.auth_request_id = None;
+ }
+
+ Some(request_id)
+ }
+
+ pub fn poll(&mut self) -> ControlAction {
+ match self.state {
+ ClientState::Shutdown => ControlAction::Error(IggyError::ClientShutdown),
+ ClientState::Disconnected => ControlAction::Noop,
+ ClientState::Authenticated | ClientState::Authenticating | ClientState::Connected => {
+ ControlAction::Noop
+ }
+ ClientState::Connecting => {
+ let server_address = match self.server_address {
+ Some(addr) => addr,
+ None => return ControlAction::Error(IggyError::ConnectionMissedSocket),
+ };
+
+ if let Some(last) = self.last_connect_attempt {
+ let now = IggyTimestamp::now();
+ let elapsed = now.as_micros().saturating_sub(last.as_micros());
+ if elapsed < self.config.reestablish_after.as_micros() {
+ let remaining =
+ IggyDuration::from(self.config.reestablish_after.as_micros() - elapsed);
+ return ControlAction::Wait(remaining);
+ }
+ }
+
+ if let Some(max_retries) = self.config.max_retries {
+ if self.retry_count >= max_retries {
+ return ControlAction::Error(IggyError::MaxRetriesExceeded);
+ }
+ }
+
+ self.retry_count += 1;
+ self.last_connect_attempt = Some(IggyTimestamp::now());
+
+ return ControlAction::Connect(server_address);
+ }
+ }
+ }
+
+ pub fn desire_connect(&mut self, server_address: SocketAddr) -> Result<(), IggyError> {
+ match self.state {
+ ClientState::Shutdown => return Err(IggyError::ClientShutdown),
+ ClientState::Connecting => return Ok(()),
+ ClientState::Connected | ClientState::Authenticating | ClientState::Authenticated => {
+ return Ok(());
+ }
+ _ => {
+ self.state = ClientState::Connecting;
+ self.server_address = Some(server_address);
+ }
+ }
+
+ Ok(())
+ }
+
+ pub fn on_connected(&mut self) -> Result<(), IggyError> {
+ debug!("Transport connected");
+ if self.state != ClientState::Connecting {
+ return Err(IggyError::IncorrectConnectionState);
+ }
+ self.state = ClientState::Connected;
+ self.retry_count = 0;
+
+ match &self.config.auto_login {
+ AutoLogin::Disabled => {
+ info!("Automatic sign-in is disabled.");
+ }
+ AutoLogin::Enabled(credentials) => {
+ if !self.auth_pending {
+ self.state = ClientState::Authenticating;
+ self.auth_pending = true;
+
+ match credentials {
+ Credentials::UsernamePassword(username, password) => {
+ let auth_payload = encode_auth(&username, &password);
+ let auth_id = self.queue_send(0x0A, auth_payload);
+ self.auth_request_id = Some(auth_id);
+ }
+ _ => {
+ todo!("add PersonalAccessToken")
+ }
+ }
+ }
+ }
+ }
+
+ Ok(())
+ }
+
+ pub fn disconnect(&mut self) {
+ debug!("Transport disconnected");
+ self.state = ClientState::Disconnected;
+ self.auth_pending = false;
+ self.auth_request_id = None;
+ self.sent_order.clear();
+ }
+
+ pub fn shutdown(&mut self) {
+ self.state = ClientState::Shutdown;
+ self.auth_pending = false;
+ self.auth_request_id = None;
+ self.sent_order.clear();
+ }
+
+ pub fn should_wait_auth(&self) -> bool {
+ matches!(self.config.auto_login, AutoLogin::Enabled(_)) && self.auth_pending
+ }
+
+ pub fn take_auth_result(&mut self) -> Option<Result<(), IggyError>> {
+ self.last_auth_result.take()
+ }
+}
+
+fn encode_auth(username: &str, password: &str) -> Bytes {
+ let mut buf = BytesMut::new();
+ buf.put_u32_le(username.len() as u32);
+ buf.put_slice(username.as_bytes());
+ buf.put_u32_le(password.len() as u32);
+ buf.put_slice(password.as_bytes());
+ buf.freeze()
+}
+
+fn make_header(total_len: u32, code: u32) -> [u8; 8] {
+ let mut h = [0u8; 8];
+ h[..4].copy_from_slice(&total_len.to_le_bytes());
+ h[4..].copy_from_slice(&code.to_le_bytes());
+ h
+}