You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by pi...@apache.org on 2026/05/29 10:30:24 UTC
(iggy) branch iggy_sdk_raw_request created (now 3f153a743)
This is an automated email from the ASF dual-hosted git repository.
spetz pushed a change to branch iggy_sdk_raw_request
in repository https://gitbox.apache.org/repos/asf/iggy.git
at 3f153a743 feat(sdk): support sending raw requests for custom commands
This branch includes the following new commits:
new 3f153a743 feat(sdk): support sending raw requests for custom commands
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.
(iggy) 01/01: feat(sdk): support sending raw requests for custom commands
Posted by pi...@apache.org.
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 <pi...@gmail.com>
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:secret@127.0.0.1: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 <de...@iggy.apache.org>"]
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",