feat(sdk): support sending raw requests for custom commands (#3373)
The SDK only exposed the commands it explicitly wraps, with no public
way to issue an arbitrary one. This left no room for extending a
deployment with custom server commands without modifying the SDK.
Expose two inherent methods on IggyClient: send_binary_request(code,
payload) for the binary transports (TCP, QUIC, WebSocket) and
send_http_request(method, path, body) for HTTP, each returning the
raw response bytes. Both reuse the live authenticated connection,
framing, and reconnect logic. Calling the binary method on HTTP or
the reverse returns FeatureUnavailable.
These methods stay off the Client trait to avoid burdening every
transport with calls meaningless to it. The HttpMethod enum lives in
iggy_common so the trait carries no concrete HTTP client dependency.
diff --git a/Cargo.lock b/Cargo.lock
index 4a44aa1..602eeb4 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -6441,7 +6441,7 @@
[[package]]
name = "iggy"
-version = "0.10.1-edge.1"
+version = "0.10.1-edge.2"
dependencies = [
"async-broadcast",
"async-dropper",
@@ -6651,7 +6651,7 @@
[[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 @@
[[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 1721938..506e164 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -178,10 +178,10 @@
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 a7e9635..7b28fd6 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 37cbfc1..2d5b106 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 dd504e9..5c064be 100644
--- a/core/common/src/lib.rs
+++ b/core/common/src/lib.rs
@@ -93,6 +93,7 @@
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/common/src/types/http.rs b/core/common/src/types/http.rs
new file mode 100644
index 0000000..32b6195
--- /dev/null
+++ b/core/common/src/types/http.rs
@@ -0,0 +1,40 @@
+/* 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 strum::{Display, EnumString, IntoStaticStr};
+
+/// HTTP request method, kept dependency-free so `iggy_common` carries no
+/// concrete HTTP client type. Defaults to `Get`.
+#[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 b0bace6..f1a2f37 100644
--- a/core/common/src/types/mod.rs
+++ b/core/common/src/types/mod.rs
@@ -24,6 +24,7 @@
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 09a6a86..1d5633f 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 0000000..58e6867
--- /dev/null
+++ b/core/integration/tests/sdk/raw.rs
@@ -0,0 +1,108 @@
+/* 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 bytes::Bytes;
+use iggy::prelude::*;
+use iggy_binary_protocol::WireEncode;
+use iggy_binary_protocol::codes::{GET_STATS_CODE, LOGIN_USER_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 method and returns `FeatureUnavailable`
+/// for the other's.
+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());
+
+ // Rejected before the wire (guards the VSR consensus-session panic
+ // when a login code is sent raw on a bound client).
+ let error = client
+ .send_binary_request(LOGIN_USER_CODE, Bytes::new())
+ .await
+ .expect_err("session-control codes must be rejected by the raw path");
+ assert_eq!(error, IggyError::InvalidCommand);
+
+ // VSR encoder is closed-world: unknown code rejected at encode time.
+ #[cfg(feature = "vsr")]
+ {
+ let error = client
+ .send_binary_request(60_000, Bytes::new())
+ .await
+ .expect_err("unknown code must be rejected under VSR");
+ assert_eq!(error, IggyError::InvalidCommand);
+ }
+
+ 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 3aae027..560d1f5 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 6ed641e..bc93c17 100644
--- a/core/sdk/src/clients/client.rs
+++ b/core/sdk/src/clients/client.rs
@@ -20,6 +20,7 @@
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,14 @@
use crate::websocket::websocket_client::WebSocketClient;
use async_broadcast::Receiver;
use async_trait::async_trait;
+use bytes::Bytes;
+use iggy_binary_protocol::codes::{
+ LOGIN_REGISTER_CODE, LOGIN_REGISTER_WITH_PAT_CODE, LOGIN_USER_CODE,
+ LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE, LOGOUT_USER_CODE,
+};
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;
@@ -40,6 +46,16 @@
use tracing::log::warn;
use tracing::{debug, error, info};
+/// Auth/session codes rejected by the raw binary path. Must go through the
+/// typed `login_user` / `logout_user` methods to keep session state correct.
+const SESSION_CONTROL_CODES: [u32; 5] = [
+ LOGIN_USER_CODE,
+ LOGOUT_USER_CODE,
+ LOGIN_REGISTER_CODE,
+ LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE,
+ LOGIN_REGISTER_WITH_PAT_CODE,
+];
+
/// The main client struct which implements all the `Client` traits and wraps the underlying low-level client for the specific transport.
///
/// It also provides the additional builders for the standalone consumer, consumer group, and producer.
@@ -182,6 +198,44 @@
pub async fn get_connection_info(&self) -> ConnectionInfo {
self.client.read().await.get_connection_info().await
}
+
+ /// Send a raw binary command (`code` + serialized `payload`), returning the
+ /// raw response. Binary transports only (HTTP yields `FeatureUnavailable`).
+ ///
+ /// Login and logout codes are rejected with `InvalidCommand`. Use the
+ /// `login_user` / `logout_user` methods so SDK session state stays correct.
+ ///
+ /// Custom codes only work on the classic protocol. Under `vsr` the encoder
+ /// is closed-world: an unknown code yields `InvalidCommand`, a replicated
+ /// code with no mapping yields `UnknownReplicatedCommand`.
+ pub async fn send_binary_request(&self, code: u32, payload: Bytes) -> Result<Bytes, IggyError> {
+ if SESSION_CONTROL_CODES.contains(&code) {
+ return Err(IggyError::InvalidCommand);
+ }
+ 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,
+ ClientWrapper::Tcp(_)
+ | ClientWrapper::Quic(_)
+ | ClientWrapper::WebSocket(_)
+ | ClientWrapper::Iggy(_) => Err(IggyError::FeatureUnavailable),
+ }
+ }
}
#[async_trait]
@@ -406,4 +460,33 @@
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)));
+ }
+
+ #[tokio::test]
+ async fn should_reject_session_control_codes_on_binary_request() {
+ let client = IggyClient::default();
+ for code in SESSION_CONTROL_CODES {
+ let result = client.send_binary_request(code, Bytes::new()).await;
+ assert!(
+ matches!(result, Err(IggyError::InvalidCommand)),
+ "code {code} must be rejected before reaching the transport"
+ );
+ }
+ }
}
diff --git a/core/sdk/src/http/http_client.rs b/core/sdk/src/http/http_client.rs
index 40f2d26..d7edcde 100644
--- a/core/sdk/src/http/http_client.rs
+++ b/core/sdk/src/http/http_client.rs
@@ -20,12 +20,13 @@
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 @@
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 eb49f41..b651009 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 @@
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 3a12b5e..4a3d29b 100644
--- a/core/sdk/src/http/mod.rs
+++ b/core/sdk/src/http/mod.rs
@@ -20,7 +20,7 @@
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 3cb66f0..840599b 100644
--- a/core/sdk/src/prelude.rs
+++ b/core/sdk/src/prelude.rs
@@ -53,11 +53,11 @@
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 8aa8a13..8153e11 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 <dev@iggy.apache.org>"]
license = "Apache-2.0"
@@ -37,7 +37,7 @@
[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",
diff --git a/foreign/python/pyproject.toml b/foreign/python/pyproject.toml
index 39bc64a..97e3502 100644
--- a/foreign/python/pyproject.toml
+++ b/foreign/python/pyproject.toml
@@ -22,7 +22,7 @@
[project]
name = "apache-iggy"
requires-python = ">=3.10"
-version = "0.8.1.dev1"
+version = "0.8.1.dev2"
description = "Apache 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."
readme = "README.md"
license = { file = "LICENSE" }
