This is an automated email from the ASF dual-hosted git repository. spetz pushed a commit to branch iggy_sdk_raw_request in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 3f153a743ad877826da269c59fdbc5680e4a1ae7 Author: spetz <[email protected]> AuthorDate: Fri May 29 12:30:14 2026 +0200 feat(sdk): support sending raw requests for custom commands --- Cargo.lock | 6 +- Cargo.toml | 6 +- core/binary_protocol/Cargo.toml | 2 +- core/common/Cargo.toml | 2 +- core/common/src/lib.rs | 1 + .../src/http/mod.rs => common/src/types/http.rs} | 35 +++++---- core/common/src/types/mod.rs | 1 + core/integration/tests/sdk/mod.rs | 1 + core/integration/tests/sdk/raw.rs | 90 ++++++++++++++++++++++ core/sdk/Cargo.toml | 2 +- core/sdk/src/clients/client.rs | 46 ++++++++++- core/sdk/src/http/http_client.rs | 30 +++++++- core/sdk/src/http/http_transport.rs | 12 ++- core/sdk/src/http/mod.rs | 2 +- core/sdk/src/prelude.rs | 10 +-- foreign/python/Cargo.toml | 4 +- 16 files changed, 216 insertions(+), 34 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 4a44aa177..602eeb4ab 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6441,7 +6441,7 @@ checksum = "cd62e6b5e86ea8eeeb8db1de02880a6abc01a397b2ebb64b5d74ac255318f5cb" [[package]] name = "iggy" -version = "0.10.1-edge.1" +version = "0.10.1-edge.2" dependencies = [ "async-broadcast", "async-dropper", @@ -6651,7 +6651,7 @@ dependencies = [ [[package]] name = "iggy_binary_protocol" -version = "0.10.1-edge.1" +version = "0.10.1-edge.2" dependencies = [ "aligned-vec", "bytemuck", @@ -6663,7 +6663,7 @@ dependencies = [ [[package]] name = "iggy_common" -version = "0.10.1-edge.1" +version = "0.10.1-edge.2" dependencies = [ "aes-gcm", "async-broadcast", diff --git a/Cargo.toml b/Cargo.toml index 172193865..506e164d3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -178,10 +178,10 @@ hwlocality = "1.0.0-alpha.12" iceberg = "0.9.1" iceberg-catalog-rest = "0.9.1" iceberg-storage-opendal = "0.9.1" -iggy = { path = "core/sdk", version = "0.10.1-edge.1" } +iggy = { path = "core/sdk", version = "0.10.1-edge.2" } iggy-cli = { path = "core/cli", version = "0.13.1-edge.1" } -iggy_binary_protocol = { path = "core/binary_protocol", version = "0.10.1-edge.1" } -iggy_common = { path = "core/common", version = "0.10.1-edge.1" } +iggy_binary_protocol = { path = "core/binary_protocol", version = "0.10.1-edge.2" } +iggy_common = { path = "core/common", version = "0.10.1-edge.2" } iggy_connector_sdk = { path = "core/connectors/sdk", version = "0.3.1-edge.1" } indexmap = "2.14.0" integration = { path = "core/integration" } diff --git a/core/binary_protocol/Cargo.toml b/core/binary_protocol/Cargo.toml index a7e963534..7b28fd618 100644 --- a/core/binary_protocol/Cargo.toml +++ b/core/binary_protocol/Cargo.toml @@ -17,7 +17,7 @@ [package] name = "iggy_binary_protocol" -version = "0.10.1-edge.1" +version = "0.10.1-edge.2" description = "Wire protocol types and codec for the Iggy binary protocol. Shared between server and SDK." edition = "2024" license = "Apache-2.0" diff --git a/core/common/Cargo.toml b/core/common/Cargo.toml index 37cbfc1c6..2d5b106c2 100644 --- a/core/common/Cargo.toml +++ b/core/common/Cargo.toml @@ -16,7 +16,7 @@ # under the License. [package] name = "iggy_common" -version = "0.10.1-edge.1" +version = "0.10.1-edge.2" description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second." edition = "2024" license = "Apache-2.0" diff --git a/core/common/src/lib.rs b/core/common/src/lib.rs index dd504e9f5..5c064be44 100644 --- a/core/common/src/lib.rs +++ b/core/common/src/lib.rs @@ -93,6 +93,7 @@ pub use types::consumer::consumer_offset_info::*; pub use types::consumer::consumer_offsets::*; pub use types::diagnostic::diagnostic_event::DiagnosticEvent; pub use types::either::Either; +pub use types::http::HttpMethod; pub use types::identifier::*; pub use types::message::*; pub use types::partition::*; diff --git a/core/sdk/src/http/mod.rs b/core/common/src/types/http.rs similarity index 57% copy from core/sdk/src/http/mod.rs copy to core/common/src/types/http.rs index 3a12b5e84..54f3339d5 100644 --- a/core/sdk/src/http/mod.rs +++ b/core/common/src/types/http.rs @@ -16,16 +16,25 @@ * under the License. */ -pub mod cluster; -pub mod consumer_groups; -pub mod consumer_offsets; -pub mod http_client; -mod http_transport; -pub mod messages; -pub mod partitions; -pub mod personal_access_tokens; -pub mod segments; -pub mod streams; -pub mod system; -pub mod topics; -pub mod users; +use strum::{Display, EnumString, IntoStaticStr}; + +/// HTTP request method, kept dependency-free so `iggy_common` carries no +/// concrete HTTP client type. +#[derive(Clone, Copy, Debug, Default, Display, PartialEq, Eq, EnumString, IntoStaticStr)] +pub enum HttpMethod { + #[default] + #[strum(serialize = "GET")] + Get, + #[strum(serialize = "POST")] + Post, + #[strum(serialize = "PUT")] + Put, + #[strum(serialize = "PATCH")] + Patch, + #[strum(serialize = "DELETE")] + Delete, + #[strum(serialize = "HEAD")] + Head, + #[strum(serialize = "OPTIONS")] + Options, +} diff --git a/core/common/src/types/mod.rs b/core/common/src/types/mod.rs index b0bace663..f1a2f37af 100644 --- a/core/common/src/types/mod.rs +++ b/core/common/src/types/mod.rs @@ -24,6 +24,7 @@ pub(crate) mod configuration; pub(crate) mod consumer; pub(crate) mod diagnostic; pub(crate) mod either; +pub(crate) mod http; pub(crate) mod identifier; pub(crate) mod message; pub(crate) mod partition; diff --git a/core/integration/tests/sdk/mod.rs b/core/integration/tests/sdk/mod.rs index 09a6a8697..1d5633f67 100644 --- a/core/integration/tests/sdk/mod.rs +++ b/core/integration/tests/sdk/mod.rs @@ -19,3 +19,4 @@ mod hello_world; #[cfg(not(feature = "vsr"))] mod producer; +mod raw; diff --git a/core/integration/tests/sdk/raw.rs b/core/integration/tests/sdk/raw.rs new file mode 100644 index 000000000..d533b865f --- /dev/null +++ b/core/integration/tests/sdk/raw.rs @@ -0,0 +1,90 @@ +/* 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 iggy::prelude::*; +use iggy_binary_protocol::WireEncode; +use iggy_binary_protocol::codes::{GET_STATS_CODE, PING_CODE}; +use iggy_binary_protocol::requests::system::{GetStatsRequest, PingRequest}; +use integration::iggy_harness; + +#[cfg(not(feature = "vsr"))] +#[iggy_harness(test_client_transport = [Tcp, Quic, Http, WebSocket])] +async fn given_authenticated_client_when_sending_raw_request_should_round_trip( + harness: &TestHarness, +) { + let client = harness.root_client().await.unwrap(); + assert_raw_round_trip(&client).await; +} + +#[cfg(feature = "vsr")] +#[iggy_harness(test_client_transport = [Tcp, WebSocket])] +async fn given_authenticated_client_when_sending_raw_request_should_round_trip( + harness: &TestHarness, +) { + let client = harness.new_client().await.unwrap(); + client + .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD) + .await + .unwrap(); + assert_raw_round_trip(&client).await; +} + +/// Each transport answers its own raw escape hatch and rejects the other's +/// with [`IggyError::FeatureUnavailable`]: the binary `(code, payload)` frame +/// has no HTTP analogue and the `(method, path)` request has no binary one. +async fn assert_raw_round_trip(client: &IggyClient) { + match client.get_connection_info().await.protocol { + TransportProtocol::Http => { + client + .send_http_request(HttpMethod::Get, "/ping", None) + .await + .expect("HTTP ping request should succeed"); + + let stats = client + .send_http_request(HttpMethod::Get, "/stats", None) + .await + .expect("authenticated HTTP request should return a body"); + assert!(!stats.is_empty()); + + let error = client + .send_binary_request(PING_CODE, PingRequest.to_bytes()) + .await + .expect_err("binary command must be unavailable on HTTP"); + assert_eq!(error, IggyError::FeatureUnavailable); + } + _ => { + let response = client + .send_binary_request(PING_CODE, PingRequest.to_bytes()) + .await + .expect("binary ping request should succeed"); + assert!(response.is_empty()); + + let stats = client + .send_binary_request(GET_STATS_CODE, GetStatsRequest.to_bytes()) + .await + .expect("authenticated binary command should return a body"); + assert!(!stats.is_empty()); + + let error = client + .send_http_request(HttpMethod::Get, "/ping", None) + .await + .expect_err("HTTP request must be unavailable on binary transports"); + assert_eq!(error, IggyError::FeatureUnavailable); + } + } +} diff --git a/core/sdk/Cargo.toml b/core/sdk/Cargo.toml index 3aae027cb..560d1f5b0 100644 --- a/core/sdk/Cargo.toml +++ b/core/sdk/Cargo.toml @@ -17,7 +17,7 @@ [package] name = "iggy" -version = "0.10.1-edge.1" +version = "0.10.1-edge.2" description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second." edition = "2024" license = "Apache-2.0" diff --git a/core/sdk/src/clients/client.rs b/core/sdk/src/clients/client.rs index 6ed641e69..03f520ef2 100644 --- a/core/sdk/src/clients/client.rs +++ b/core/sdk/src/clients/client.rs @@ -20,6 +20,7 @@ use crate::client_wrappers::client_wrapper::ClientWrapper; use crate::client_wrappers::connection_info::ConnectionInfo; use crate::clients::client_builder::IggyClientBuilder; use crate::http::http_client::HttpClient; +use crate::http::http_transport::HttpTransport; use crate::prelude::EncryptorKind; use crate::prelude::IggyConsumerBuilder; use crate::prelude::IggyError; @@ -29,9 +30,10 @@ use crate::tcp::tcp_client::TcpClient; use crate::websocket::websocket_client::WebSocketClient; use async_broadcast::Receiver; use async_trait::async_trait; +use bytes::Bytes; use iggy_common::Consumer; use iggy_common::locking::{IggyRwLock, IggyRwLockFn}; -use iggy_common::{Client, SystemClient}; +use iggy_common::{BinaryTransport, Client, HttpMethod, SystemClient}; use iggy_common::{ConnectionStringUtils, DiagnosticEvent, Partitioner, TransportProtocol}; use std::fmt::Debug; use std::sync::Arc; @@ -182,6 +184,31 @@ impl IggyClient { pub async fn get_connection_info(&self) -> ConnectionInfo { self.client.read().await.get_connection_info().await } + + /// Send a raw binary command (`code` + serialized `payload`) and return the + /// raw response. Binary transports only; HTTP yields `FeatureUnavailable`. + pub async fn send_binary_request(&self, code: u32, payload: Bytes) -> Result<Bytes, IggyError> { + match &*self.client.read().await { + ClientWrapper::Tcp(client) => client.send_raw_with_response(code, payload).await, + ClientWrapper::Quic(client) => client.send_raw_with_response(code, payload).await, + ClientWrapper::WebSocket(client) => client.send_raw_with_response(code, payload).await, + ClientWrapper::Http(_) | ClientWrapper::Iggy(_) => Err(IggyError::FeatureUnavailable), + } + } + + /// Invoke an arbitrary HTTP endpoint and return the raw response body. HTTP + /// transport only; binary transports yield `FeatureUnavailable`. + pub async fn send_http_request( + &self, + method: HttpMethod, + path: &str, + body: Option<Bytes>, + ) -> Result<Bytes, IggyError> { + match &*self.client.read().await { + ClientWrapper::Http(client) => client.send_http_request(method, path, body).await, + _ => Err(IggyError::FeatureUnavailable), + } + } } #[async_trait] @@ -406,4 +433,21 @@ mod tests { let client = IggyClient::from_connection_string(&value); assert!(client.is_ok()); } + + #[tokio::test] + async fn should_reject_http_request_on_binary_transport() { + let client = IggyClient::default(); + let result = client + .send_http_request(HttpMethod::Get, "/ping", None) + .await; + assert!(matches!(result, Err(IggyError::FeatureUnavailable))); + } + + #[tokio::test] + async fn should_reject_binary_request_on_http_transport() { + let client = + IggyClient::from_connection_string("iggy+http://user:[email protected]:1234";).unwrap(); + let result = client.send_binary_request(0, Bytes::new()).await; + assert!(matches!(result, Err(IggyError::FeatureUnavailable))); + } } diff --git a/core/sdk/src/http/http_client.rs b/core/sdk/src/http/http_client.rs index 40f2d26d7..d7edcdee7 100644 --- a/core/sdk/src/http/http_client.rs +++ b/core/sdk/src/http/http_client.rs @@ -20,12 +20,13 @@ use crate::http::http_transport::HttpTransport; use crate::prelude::{Client, HttpClientConfig, IggyDuration, IggyError}; use async_broadcast::{Receiver, Sender, broadcast}; use async_trait::async_trait; +use bytes::Bytes; use iggy_common::locking::{IggyRwLock, IggyRwLockFn}; use iggy_common::{ ConnectionString, ConnectionStringUtils, DiagnosticEvent, HttpConnectionStringOptions, - IdentityInfo, TransportProtocol, validate_api_url, + HttpMethod, IdentityInfo, TransportProtocol, validate_api_url, }; -use reqwest::{Response, StatusCode, Url}; +use reqwest::{Method, Response, StatusCode, Url}; use reqwest_middleware::{ClientBuilder, ClientWithMiddleware}; use reqwest_retry::{RetryTransientMiddleware, policies::ExponentialBackoff}; use reqwest_tracing::{SpanBackendWithUrl, TracingMiddleware}; @@ -202,6 +203,31 @@ impl HttpTransport for HttpClient { Self::handle_response(response).await } + async fn send_http_request( + &self, + method: HttpMethod, + path: &str, + body: Option<Bytes>, + ) -> Result<Bytes, IggyError> { + let method = Method::from_bytes(<&str>::from(method).as_bytes()) + .map_err(|_| IggyError::InvalidHttpRequest)?; + let url = self.get_url(path)?; + let token = self.access_token.read().await; + let mut request = self.client.request(method, url).bearer_auth(token.deref()); + if let Some(body) = body { + request = request.body(body); + } + let response = request + .send() + .await + .map_err(|_| IggyError::InvalidHttpRequest)?; + let response = Self::handle_response(response).await?; + response + .bytes() + .await + .map_err(|_| IggyError::InvalidHttpRequest) + } + /// Returns true if the client is authenticated. async fn is_authenticated(&self) -> bool { let token = self.access_token.read().await; diff --git a/core/sdk/src/http/http_transport.rs b/core/sdk/src/http/http_transport.rs index eb49f41af..b651009d6 100644 --- a/core/sdk/src/http/http_transport.rs +++ b/core/sdk/src/http/http_transport.rs @@ -16,7 +16,8 @@ // under the License. use async_trait::async_trait; -use iggy_common::{IdentityInfo, IggyError}; +use bytes::Bytes; +use iggy_common::{HttpMethod, IdentityInfo, IggyError}; use reqwest::{Response, Url}; use serde::Serialize; @@ -59,6 +60,15 @@ pub trait HttpTransport { query: &T, ) -> Result<Response, IggyError>; + /// Invoke an arbitrary HTTP endpoint with the client's bearer auth and + /// return the raw response bytes. + async fn send_http_request( + &self, + method: HttpMethod, + path: &str, + body: Option<Bytes>, + ) -> Result<Bytes, IggyError>; + /// Returns true if the client is authenticated. async fn is_authenticated(&self) -> bool; diff --git a/core/sdk/src/http/mod.rs b/core/sdk/src/http/mod.rs index 3a12b5e84..4a3d29bd8 100644 --- a/core/sdk/src/http/mod.rs +++ b/core/sdk/src/http/mod.rs @@ -20,7 +20,7 @@ pub mod cluster; pub mod consumer_groups; pub mod consumer_offsets; pub mod http_client; -mod http_transport; +pub(crate) mod http_transport; pub mod messages; pub mod partitions; pub mod personal_access_tokens; diff --git a/core/sdk/src/prelude.rs b/core/sdk/src/prelude.rs index 3cb66f07a..840599be0 100644 --- a/core/sdk/src/prelude.rs +++ b/core/sdk/src/prelude.rs @@ -53,11 +53,11 @@ pub use iggy_common::{ ClientInfoDetails, ClusterMetadata, ClusterNode, ClusterNodeRole, ClusterNodeStatus, CompressionAlgorithm, Consumer, ConsumerGroupDetails, ConsumerKind, EncryptorKind, GlobalPermissions, HeaderKey, HeaderKind, 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, + HttpClientConfigBuilder, HttpMethod, 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, StreamDetails, StreamPermissions, SystemSnapshotType, TcpClientConfig, TcpClientConfigBuilder, TcpClientReconnectionConfig, Topic, TopicDetails, TopicPermissions, TransportEndpoints, diff --git a/foreign/python/Cargo.toml b/foreign/python/Cargo.toml index 8aa8a1354..8153e114b 100644 --- a/foreign/python/Cargo.toml +++ b/foreign/python/Cargo.toml @@ -17,7 +17,7 @@ [package] name = "apache-iggy" -version = "0.8.1-dev1" +version = "0.8.1-dev2" edition = "2024" authors = ["Iggy Committers <[email protected]>"] license = "Apache-2.0" @@ -37,7 +37,7 @@ doc = false [dependencies] bytes = "1.11.1" futures = "0.3.32" -iggy = { path = "../../core/sdk", version = "0.10.1-edge.1" } +iggy = { path = "../../core/sdk", version = "0.10.1-edge.2" } pyo3 = "0.28.3" pyo3-async-runtimes = { version = "0.28.0", features = [ "attributes",
