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/07/05 04:19:31 UTC
(iggy) branch draft-sans updated (178f9c19 -> bcb70f0d)
This is an automated email from the ASF dual-hosted git repository.
bashirbekov pushed a change to branch draft-sans
in repository https://gitbox.apache.org/repos/asf/iggy.git
discard 178f9c19 del
new bcb70f0d del
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (178f9c19)
\
N -- N -- N refs/heads/draft-sans (bcb70f0d)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
core/common/src/error/iggy_error.rs | 2 ++
core/sdk/Cargo.toml | 3 ++-
core/sdk/src/proto/runtime.rs | 16 +++++++++++++++-
core/sdk/src/transport_adapter/mod.rs | 21 ++++++++++++++++++---
core/sdk/src/transport_adapter/quic.rs | 27 ++++++++++++++-------------
5 files changed, 51 insertions(+), 18 deletions(-)
(iggy) 01/01: del
Posted by ba...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
bashirbekov pushed a commit to branch draft-sans
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit bcb70f0ded1f59b2eab4a266d01af219aeab3639
Author: haze518 <as...@gmail.com>
AuthorDate: Sat Jul 5 10:19:23 2025 +0600
del
---
core/common/src/error/iggy_error.rs | 2 +
core/sdk/Cargo.toml | 3 +-
core/sdk/src/connection/quic/mod.rs | 146 +++++++++++++++++++++++++++++----
core/sdk/src/connection/tcp/mod.rs | 4 +-
core/sdk/src/connection/tcp/tcp.rs | 18 +++-
core/sdk/src/connection/tcp/tls.rs | 102 +++++++++++------------
core/sdk/src/lib.rs | 1 +
core/sdk/src/proto/mod.rs | 18 ----
core/sdk/src/proto/runtime.rs | 16 +++-
core/sdk/src/proto/tcp_adapter.rs | 31 -------
core/sdk/src/quic/mod.rs | 2 +-
core/sdk/src/transport_adapter/mod.rs | 25 ++++++
core/sdk/src/transport_adapter/quic.rs | 43 ++++++++++
13 files changed, 287 insertions(+), 124 deletions(-)
diff --git a/core/common/src/error/iggy_error.rs b/core/common/src/error/iggy_error.rs
index cc8a284b..f3836726 100644
--- a/core/common/src/error/iggy_error.rs
+++ b/core/common/src/error/iggy_error.rs
@@ -459,6 +459,8 @@ pub enum IggyError {
CannotReadIndexPosition = 10011,
#[error("Cannot read index timestamp")]
CannotReadIndexTimestamp = 10012,
+ #[error("Receive error")]
+ ReceiveError = 10050,
}
impl IggyError {
diff --git a/core/sdk/Cargo.toml b/core/sdk/Cargo.toml
index 44cf8d2d..844a5f41 100644
--- a/core/sdk/Cargo.toml
+++ b/core/sdk/Cargo.toml
@@ -29,8 +29,9 @@ repository = "https://github.com/apache/iggy"
readme = "../../README.md"
[features]
-default = ["tokio_lock"]
+default = ["tokio_lock", "runtime_tokio"]
tokio_lock = []
+runtime_tokio = []
# Moved to common crate. Not sure if this is even needed anymore
#fast_async_lock = ["dep:fast-async-mutex"]
diff --git a/core/sdk/src/connection/quic/mod.rs b/core/sdk/src/connection/quic/mod.rs
index 804f0b71..c5d9feed 100644
--- a/core/sdk/src/connection/quic/mod.rs
+++ b/core/sdk/src/connection/quic/mod.rs
@@ -1,27 +1,141 @@
use std::sync::Arc;
+use std::{io, net::SocketAddr, pin::Pin, time::Duration};
+use iggy_common::{IggyError, QuicClientConfig};
+use rustls::crypto::CryptoProvider;
+use tracing::{error, warn};
+use crate::quic::skip_server_verification::SkipServerVerification;
-use iggy_common::QuicClientConfig;
-use quinn::{ClientConfig, Endpoint};
+use quinn::crypto::rustls::QuicClientConfig as QuinnQuicClientConfig;
+use quinn::{ClientConfig, Connection, Endpoint, IdleTimeout, VarInt};
-use crate::connection::SocketFactory;
-
-pub struct QuicFactory {
- config: ClientConfig,
- quic_config: Arc<QuicClientConfig>, // rename to QuicFabricConfig
+pub trait QuicFactory {
+ type Conn;
+ fn connect(&self) -> Pin<Box<dyn Future<Output = Result<Self::Conn, IggyError>> + Send>>;
}
-impl SocketFactory for QuicFactory {
- type Stream = Arc<quinn::Connection>;
+pub struct QuinnFactory {
+ config: Arc<QuicClientConfig>,
+ ep: Arc<Endpoint>,
+ server_address: SocketAddr,
+}
- fn connect(&self, addr: std::net::SocketAddr) -> std::pin::Pin<Box<dyn Future<Output = std::io::Result<Self::Stream>> + Send>> {
- let endpoint = Endpoint::client(addr)?;
- let config = self.config.clone();
- endpoint.set_default_client_config(config);
+impl QuicFactory for QuinnFactory {
+ type Conn = Connection;
- let connecting = endpoint.connect(addr, &self.quic_config.server_name)?;
+ fn connect(&self) -> Pin<Box<dyn Future<Output = Result<Self::Conn, IggyError>> + Send>> {
+ let ep = self.ep.clone();
+ let sn = self.config.server_name.clone();
+ let sa = self.server_address.clone();
Box::pin(async move {
- let connection = connecting.await?;
- Ok(stream)
+ let connecting = ep
+ .connect(sa, &sn)
+ .map_err(|_| IggyError::CannotEstablishConnection)?;
+
+ connecting
+ .await
+ .map_err(|_| IggyError::CannotEstablishConnection)
})
}
}
+
+impl QuinnFactory {
+ pub fn new(config: QuicClientConfig) -> Result<Self, IggyError> {
+ let cfg = Arc::new(config);
+
+ let server_address = cfg
+ .server_address
+ .parse::<SocketAddr>()
+ .map_err(|error| {
+ error!("Invalid server address: {error}");
+ IggyError::InvalidServerAddress
+ })?;
+ let client_address = if server_address.is_ipv6()
+ && cfg.client_address == QuicClientConfig::default().client_address
+ {
+ "[::1]:0"
+ } else {
+ &cfg.client_address
+ }
+ .parse::<SocketAddr>()
+ .map_err(|error| {
+ error!("Invalid client address: {error}");
+ IggyError::InvalidClientAddress
+ })?;
+
+ let quic_config = configure(&cfg)?;
+ let endpoint = Endpoint::client(client_address);
+ if endpoint.is_err() {
+ error!("Cannot create client endpoint");
+ return Err(IggyError::CannotCreateEndpoint);
+ }
+
+ let mut endpoint = endpoint.unwrap();
+ endpoint.set_default_client_config(quic_config);
+
+ Ok(Self { config: cfg, ep: Arc::new(endpoint), server_address })
+ }
+}
+
+fn configure(config: &QuicClientConfig) -> Result<ClientConfig, IggyError> {
+ let max_concurrent_bidi_streams = VarInt::try_from(config.max_concurrent_bidi_streams);
+ if max_concurrent_bidi_streams.is_err() {
+ error!(
+ "Invalid 'max_concurrent_bidi_streams': {}",
+ config.max_concurrent_bidi_streams
+ );
+ return Err(IggyError::InvalidConfiguration);
+ }
+
+ let receive_window = VarInt::try_from(config.receive_window);
+ if receive_window.is_err() {
+ error!("Invalid 'receive_window': {}", config.receive_window);
+ return Err(IggyError::InvalidConfiguration);
+ }
+
+ let mut transport = quinn::TransportConfig::default();
+ transport.initial_mtu(config.initial_mtu);
+ transport.send_window(config.send_window);
+ transport.receive_window(receive_window.unwrap());
+ transport.datagram_send_buffer_size(config.datagram_send_buffer_size as usize);
+ transport.max_concurrent_bidi_streams(max_concurrent_bidi_streams.unwrap());
+ if config.keep_alive_interval > 0 {
+ transport.keep_alive_interval(Some(Duration::from_millis(config.keep_alive_interval)));
+ }
+ if config.max_idle_timeout > 0 {
+ let max_idle_timeout =
+ IdleTimeout::try_from(Duration::from_millis(config.max_idle_timeout));
+ if max_idle_timeout.is_err() {
+ error!("Invalid 'max_idle_timeout': {}", config.max_idle_timeout);
+ return Err(IggyError::InvalidConfiguration);
+ }
+ transport.max_idle_timeout(Some(max_idle_timeout.unwrap()));
+ }
+
+ if CryptoProvider::get_default().is_none() {
+ if let Err(e) = rustls::crypto::ring::default_provider().install_default() {
+ warn!(
+ "Failed to install rustls crypto provider. Error: {:?}. This may be normal if another thread installed it first.",
+ e
+ );
+ }
+ }
+ let mut client_config = match config.validate_certificate {
+ true => ClientConfig::with_platform_verifier(),
+ false => {
+ match QuinnQuicClientConfig::try_from(
+ rustls::ClientConfig::builder()
+ .dangerous()
+ .with_custom_certificate_verifier(SkipServerVerification::new())
+ .with_no_client_auth(),
+ ) {
+ Ok(config) => ClientConfig::new(Arc::new(config)),
+ Err(error) => {
+ error!("Failed to create QUIC client configuration: {error}");
+ return Err(IggyError::InvalidConfiguration);
+ }
+ }
+ }
+ };
+ client_config.transport_config(Arc::new(transport));
+ Ok(client_config)
+}
diff --git a/core/sdk/src/connection/tcp/mod.rs b/core/sdk/src/connection/tcp/mod.rs
index ec1d2deb..3d133a45 100644
--- a/core/sdk/src/connection/tcp/mod.rs
+++ b/core/sdk/src/connection/tcp/mod.rs
@@ -7,8 +7,8 @@ use futures::{AsyncRead, AsyncWrite};
pub trait AsyncStream: AsyncRead + AsyncWrite + Unpin + Send + 'static {}
impl<T> AsyncStream for T where T: AsyncRead + AsyncWrite + Unpin + Send + 'static {}
-pub trait SocketFactory { // TODO rename to something common
+pub trait SocketFactory {
type Stream: AsyncStream;
- fn connect(&self, addr: SocketAddr) -> Pin<Box<dyn Future<Output = io::Result<Self::Stream>> + Send>>;
+ fn connect(&self) -> Pin<Box<dyn Future<Output = io::Result<Self::Stream>> + Send>>;
}
diff --git a/core/sdk/src/connection/tcp/tcp.rs b/core/sdk/src/connection/tcp/tcp.rs
index 561a81d7..d511867a 100644
--- a/core/sdk/src/connection/tcp/tcp.rs
+++ b/core/sdk/src/connection/tcp/tcp.rs
@@ -1,18 +1,30 @@
+use std::sync::Arc;
+
use crate::connection::tcp::SocketFactory;
+use iggy_common::TcpClientConfig;
use tokio_util::compat::TokioAsyncReadCompatExt;
pub type TokioCompatStream = tokio_util::compat::Compat<tokio::net::TcpStream>;
-pub struct TokioTcpFactory;
+pub struct TokioTcpFactory {
+ config: Arc<TcpClientConfig>,
+}
impl SocketFactory for TokioTcpFactory {
type Stream = TokioCompatStream;
- fn connect(&self, addr: std::net::SocketAddr) -> std::pin::Pin<Box<dyn Future<Output = std::io::Result<Self::Stream>> + Send>> {
+ fn connect(&self) -> std::pin::Pin<Box<dyn Future<Output = std::io::Result<Self::Stream>> + Send>> {
+ let sa = self.config.server_address.clone();
Box::pin(async move {
- let stream = tokio::net::TcpStream::connect(addr).await?;
+ let stream = tokio::net::TcpStream::connect(sa).await?;
stream.set_nodelay(true)?;
Ok(stream.compat())
})
}
}
+
+impl TokioTcpFactory {
+ pub fn new(config: TcpClientConfig) -> Self {
+ Self { config: Arc::new(config) }
+ }
+}
diff --git a/core/sdk/src/connection/tcp/tls.rs b/core/sdk/src/connection/tcp/tls.rs
index 624b09e2..a6949067 100644
--- a/core/sdk/src/connection/tcp/tls.rs
+++ b/core/sdk/src/connection/tcp/tls.rs
@@ -5,54 +5,54 @@ use tokio_util::compat::{Compat, TokioAsyncReadCompatExt};
use crate::connection::tcp::{AsyncStream, SocketFactory};
-pub trait TlsConnector<Plain: AsyncStream> {
- type TlsStream: AsyncStream;
-
- fn handshake(
- &self,
- plain: Plain,
- domain: &str,
- ) -> Pin<Box<dyn Future<Output = io::Result<Self::TlsStream>> + Send>>;
-}
-
-pub struct TokioRustls {
- inner: tokio_rustls::TlsConnector,
-}
-impl TlsConnector<Compat<tokio::net::TcpStream>> for TokioRustls {
- type TlsStream = Compat<tokio_rustls::client::TlsStream<tokio::net::TcpStream>>;
-
- fn handshake(
- &self,
- plain: Compat<tokio::net::TcpStream>,
- domain: &str,
- ) -> Pin<Box<dyn Future<Output = std::io::Result<Self::TlsStream>> + Send>> {
- let dom = ServerName::try_from(domain.to_owned()).unwrap();
- let tcp = plain.into_inner();
- let fut = self.inner.connect(dom, tcp);
- Box::pin(async move { Ok(fut.await?.compat()) })
- }
-}
-
-pub struct TlsFactory<F, C> {
- base: F,
- tls: C,
- domain: String,
-}
-
-impl<F, C> SocketFactory for TlsFactory<F, C>
-where
- F: SocketFactory,
- C: TlsConnector<F::Stream> + Clone + Send + Sync + 'static,
-{
- type Stream = C::TlsStream;
-
- fn connect(&self, addr: SocketAddr) -> Pin<Box<dyn Future<Output = io::Result<Self::Stream>> + Send>> {
- let base = self.base.connect(addr);
- let tls = self.tls.clone();
- let dom = self.domain.clone();
- Box::pin(async move {
- let plain = base.await?;
- tls.handshake(plain, &dom).await
- })
- }
-}
+// pub trait TlsConnector<Plain: AsyncStream> {
+// type TlsStream: AsyncStream;
+
+// fn handshake(
+// &self,
+// plain: Plain,
+// domain: &str,
+// ) -> Pin<Box<dyn Future<Output = io::Result<Self::TlsStream>> + Send>>;
+// }
+
+// pub struct TokioRustls {
+// inner: tokio_rustls::TlsConnector,
+// }
+// impl TlsConnector<Compat<tokio::net::TcpStream>> for TokioRustls {
+// type TlsStream = Compat<tokio_rustls::client::TlsStream<tokio::net::TcpStream>>;
+
+// fn handshake(
+// &self,
+// plain: Compat<tokio::net::TcpStream>,
+// domain: &str,
+// ) -> Pin<Box<dyn Future<Output = std::io::Result<Self::TlsStream>> + Send>> {
+// let dom = ServerName::try_from(domain.to_owned()).unwrap();
+// let tcp = plain.into_inner();
+// let fut = self.inner.connect(dom, tcp);
+// Box::pin(async move { Ok(fut.await?.compat()) })
+// }
+// }
+
+// pub struct TlsFactory<F, C> {
+// base: F,
+// tls: C,
+// domain: String,
+// }
+
+// impl<F, C> SocketFactory for TlsFactory<F, C>
+// where
+// F: SocketFactory,
+// C: TlsConnector<F::Stream> + Clone + Send + Sync + 'static,
+// {
+// type Stream = C::TlsStream;
+
+// fn connect(&self) -> Pin<Box<dyn Future<Output = io::Result<Self::Stream>> + Send>> {
+// let base = self.base.connect(addr);
+// let tls = self.tls.clone();
+// let dom = self.domain.clone();
+// Box::pin(async move {
+// let plain = base.await?;
+// tls.handshake(plain, &dom).await
+// })
+// }
+// }
diff --git a/core/sdk/src/lib.rs b/core/sdk/src/lib.rs
index 87efe42c..e27a3554 100644
--- a/core/sdk/src/lib.rs
+++ b/core/sdk/src/lib.rs
@@ -29,3 +29,4 @@ pub mod tcp;
pub mod proto;
pub mod transport_factory;
pub mod connection;
+pub mod transport_adapter;
diff --git a/core/sdk/src/proto/mod.rs b/core/sdk/src/proto/mod.rs
index 478f6fac..62d0aca6 100644
--- a/core/sdk/src/proto/mod.rs
+++ b/core/sdk/src/proto/mod.rs
@@ -1,20 +1,2 @@
pub mod connection;
-pub mod tcp_adapter;
pub mod runtime;
-
-use std::collections::VecDeque;
-
-use bytes::Bytes;
-use iggy_common::ClientState;
-
-pub struct IggyCore {
- buffer: VecDeque<Bytes>,
- current_state: ClientState,
-}
-
-impl IggyCore {
- pub fn write(&mut self, payload: Bytes) {
- self.buffer.push_back(payload)
- }
-
-}
diff --git a/core/sdk/src/proto/runtime.rs b/core/sdk/src/proto/runtime.rs
index 7adb1b19..43a382c2 100644
--- a/core/sdk/src/proto/runtime.rs
+++ b/core/sdk/src/proto/runtime.rs
@@ -1,7 +1,16 @@
use std::{ops::{Deref, DerefMut}, pin::Pin};
+use iggy_common::IggyError;
+
+#[cfg(feature = "runtime_tokio")]
+type Mutex<T> = tokio::sync::Mutex<T>;
+#[cfg(feature = "runtime_tokio")]
+pub type OneShotSender<T> = tokio::sync::oneshot::Sender<T>;
+#[cfg(feature = "runtime_tokio")]
+pub type OneShotReceiver<T> = tokio::sync::oneshot::Receiver<T>;
+
pub trait Runtime: Sync + Send + 'static {
- type Mutex<T>: Lockable<T> + Send + Sync + 'static
+ type Mutex<T>: Lockable<T> + Send + Sync + 'static // TODO remove
where
T: Send + 'static;
@@ -16,3 +25,8 @@ pub trait Lockable<T>: Send + Sync + 'static {
fn lock(&self) -> Pin<Box<dyn Future<Output = Self::Guard<'_>> + Send + '_>>;
}
+
+#[cfg(feature = "runtime_tokio")]
+pub fn oneshot<T>() -> (OneShotSender<T>, OneShotReceiver<T>) {
+ tokio::sync::oneshot::channel()
+}
diff --git a/core/sdk/src/proto/tcp_adapter.rs b/core/sdk/src/proto/tcp_adapter.rs
deleted file mode 100644
index 41edd05d..00000000
--- a/core/sdk/src/proto/tcp_adapter.rs
+++ /dev/null
@@ -1,31 +0,0 @@
-use std::{sync::Mutex, task::Context};
-
-use bytes::Bytes;
-use futures::future::poll_fn;
-use iggy_common::{Command, IggyError};
-
-use crate::proto::{connection::IggyCore, runtime::{Lockable, Runtime}};
-use crate::tcp::tcp_connection_stream_kind::ConnectionStreamKind;
-
-pub struct TCPAdapter<R: Runtime> {
- rt: R,
- core: Mutex<IggyCore>,
- pub(crate) stream: R::Mutex<Option<ConnectionStreamKind>>,
-}
-
-impl<R: Runtime> TCPAdapter<R> {
- pub async fn send_with_response(&self, command: &impl Command) -> Result<Bytes, IggyError> {
- // poll_fn(|cx| )
- Ok(Bytes::new())
- }
-
- fn write(&self, command: &impl Command) -> Result<(), IggyError> {
- command.validate()?;
- self.core.lock().unwrap().write(command)
- }
-
- fn execute_poll(&mut self, cx: &mut Context, command: &impl Command) -> Result<Bytes, IggyError> {
- self.write(command)?;
-
- }
-}
diff --git a/core/sdk/src/quic/mod.rs b/core/sdk/src/quic/mod.rs
index 0ffce248..c67398e0 100644
--- a/core/sdk/src/quic/mod.rs
+++ b/core/sdk/src/quic/mod.rs
@@ -17,4 +17,4 @@
*/
pub mod quick_client;
-mod skip_server_verification;
+pub mod skip_server_verification;
diff --git a/core/sdk/src/transport_adapter/mod.rs b/core/sdk/src/transport_adapter/mod.rs
new file mode 100644
index 00000000..c7d6a932
--- /dev/null
+++ b/core/sdk/src/transport_adapter/mod.rs
@@ -0,0 +1,25 @@
+pub mod quic;
+
+use std::{pin::Pin, sync::Arc};
+
+use bytes::Bytes;
+use futures::FutureExt;
+use iggy_common::{Command, IggyError};
+
+use crate::proto::runtime::OneShotReceiver;
+
+pub struct RespFut {
+ rx: OneShotReceiver<Bytes>
+}
+
+impl Future for RespFut {
+ type Output = Result<Bytes, IggyError>;
+
+ fn poll(self: Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> std::task::Poll<Self::Output> {
+ self.get_mut().rx.poll_unpin(cx).map_err(|_| IggyError::ReceiveError)
+ }
+}
+
+pub trait TransportAdapter {
+ fn send_with_response<'a, T: Command>(&'a self, command: &'a T) -> Pin<Box<dyn Future<Output = Result<RespFut, IggyError>> + Send + 'a>>;
+}
diff --git a/core/sdk/src/transport_adapter/quic.rs b/core/sdk/src/transport_adapter/quic.rs
new file mode 100644
index 00000000..8900df5f
--- /dev/null
+++ b/core/sdk/src/transport_adapter/quic.rs
@@ -0,0 +1,43 @@
+use std::{collections::VecDeque, pin::Pin, sync::{Arc, Mutex}, task::Waker};
+
+use bytes::Bytes;
+use futures::{channel::oneshot, FutureExt};
+use iggy_common::{Command, IggyError};
+
+use crate::{connection::quic::QuicFactory, proto::{connection::IggyCore, runtime::{self, Lockable, Runtime}}, transport_adapter::{RespFut, TransportAdapter}};
+
+pub struct QuicAdapter<F: QuicFactory, R: Runtime> {
+ factory: Arc<F>,
+ rt: Arc<R>,
+ core: R::Mutex<IggyCore>,
+ waiters: VecDeque<Waker>
+}
+
+impl<F, R> TransportAdapter for QuicAdapter<F, R>
+where
+ F: QuicFactory + Send + Sync + 'static,
+ R: Runtime,
+{
+ fn send_with_response<'a, T: Command>(&'a self, command: &'a T) -> Pin<Box<dyn Future<Output = Result<RespFut, IggyError>> + Send + 'a >> {
+ Box::pin(async move {
+ let (tx, rx) = runtime::oneshot::<Bytes>();
+ self.core.lock().await.write(command)?;
+ self.waiters.push_back(value);
+ Ok(RespFut { rx: rx })
+ })
+ }
+}
+
+
+
+/*
+// tood для transprot
+ fn send_with_response<'a, T: iggy_common::Command>(&'a self, command: &'a T) -> Pin<Box<dyn Future<Output = Result<bytes::Bytes, iggy_common::IggyError>> + Send + 'a>> {
+ Box::pin(async move {
+ self.core.lock().await.write(command)?;
+
+ Ok(Bytes::new())
+ })
+ }
+
+*/
\ No newline at end of file