You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by "slbotbm (via GitHub)" <gi...@apache.org> on 2026/07/05 13:57:30 UTC
[PR] feat(cpp): add functions related to client lifecycle, and message headers [iggy]
slbotbm opened a new pull request, #3610:
URL: https://github.com/apache/iggy/pull/3610
## Which issue does this PR address?
<!--
We generally require a GitHub issue for all bug fixes and enhancements. Link it with GitHub syntax, keep the line that applies and delete the other:
- `Closes #123` auto-closes the issue when this PR merges (full fix).
- `Relates to #123` links without closing (partial or related work).
-->
Closes #
Relates to #2100
## Rationale
Contributes to completion of cpp sdk.
## What changed?
added user header handling, and the following functions:
- disconnect
- shutdown
- delete_segments
- get_cluster_metadata
- update_permissions
- change_password
## Local Execution
- Passed
- Pre-commit hooks ran
## AI Usage
If AI tools were used, please answer:
1. Which tools? codex
2. Scope of usage? code generation, code investigation
3. How did you verify the generated code works correctly? read through it, tested it.
4. Can you explain every line of the code if asked? yes
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(cpp): add functions related to client lifecycle, and message headers [iggy]
Posted by "slbotbm (via GitHub)" <gi...@apache.org>.
slbotbm commented on code in PR #3610:
URL: https://github.com/apache/iggy/pull/3610#discussion_r3600730300
##########
foreign/cpp/src/type_conversion.rs:
##########
@@ -314,12 +433,36 @@ impl From<RustConsumerGroupDetails> for ffi::ConsumerGroupDetails {
}
}
-impl From<RustIggyMessage> for ffi::IggyMessagePolled {
- fn from(message: RustIggyMessage) -> Self {
+impl TryFrom<RustIggyMessage> for ffi::IggyMessagePolled {
+ type Error = String;
+
+ fn try_from(message: RustIggyMessage) -> Result<Self, Self::Error> {
let id_bytes = message.header.id.to_le_bytes();
let id_lo = u64::from_le_bytes(id_bytes[0..8].try_into().unwrap());
let id_hi = u64::from_le_bytes(id_bytes[8..16].try_into().unwrap());
- ffi::IggyMessagePolled {
+ let user_headers = match message.user_headers {
+ Some(raw_headers) => {
+ let wire_headers = WireUserHeaders::from_bytes(raw_headers).map_err(|error| {
Review Comment:
I changed the code to emit empty headers for malformed user-headers TLV while preserving the message and the batch, though I cannot issue a warning from the rust side to the cpp side without introducing some field like `user_header_warning` that will get populated on errors. Though this is a pretty awkward design.
##########
foreign/cpp/src/lib.rs:
##########
@@ -88,11 +88,40 @@ mod ffi {
topics_count: u32,
}
+ #[repr(u8)]
+ enum HeaderKind {
+ Raw = 1,
+ String = 2,
+ Bool = 3,
+ Int8 = 4,
+ Int16 = 5,
+ Int32 = 6,
+ Int64 = 7,
+ Int128 = 8,
+ UInt8 = 9,
Review Comment:
Done
##########
foreign/cpp/src/type_conversion.rs:
##########
@@ -188,6 +194,119 @@ impl From<RustStats> for ffi::Stats {
}
}
+impl From<RustTransportEndpoints> for ffi::TransportEndpoints {
+ fn from(endpoints: RustTransportEndpoints) -> Self {
+ ffi::TransportEndpoints {
+ tcp: endpoints.tcp,
+ quic: endpoints.quic,
+ http: endpoints.http,
+ websocket: endpoints.websocket,
+ }
+ }
+}
+
+impl From<RustClusterNode> for ffi::ClusterNode {
+ fn from(node: RustClusterNode) -> Self {
+ ffi::ClusterNode {
+ name: node.name,
+ ip: node.ip,
+ endpoints: ffi::TransportEndpoints::from(node.endpoints),
+ role: node.role.to_string(),
+ status: node.status.to_string(),
+ }
+ }
+}
+
+impl From<RustClusterMetadata> for ffi::ClusterMetadata {
+ fn from(metadata: RustClusterMetadata) -> Self {
+ ffi::ClusterMetadata {
+ name: metadata.name,
+ nodes: metadata
+ .nodes
+ .into_iter()
+ .map(ffi::ClusterNode::from)
+ .collect(),
+ }
+ }
+}
+
+impl From<ffi::GlobalPermissions> for RustGlobalPermissions {
+ fn from(permissions: ffi::GlobalPermissions) -> Self {
+ RustGlobalPermissions {
+ manage_servers: permissions.manage_servers,
+ read_servers: permissions.read_servers,
+ manage_users: permissions.manage_users,
+ read_users: permissions.read_users,
+ manage_streams: permissions.manage_streams,
+ read_streams: permissions.read_streams,
+ manage_topics: permissions.manage_topics,
+ read_topics: permissions.read_topics,
+ poll_messages: permissions.poll_messages,
+ send_messages: permissions.send_messages,
+ }
+ }
+}
+
+impl From<ffi::TopicPermissions> for RustTopicPermissions {
+ fn from(permissions: ffi::TopicPermissions) -> Self {
+ RustTopicPermissions {
+ manage_topic: permissions.manage_topic,
+ read_topic: permissions.read_topic,
+ poll_messages: permissions.poll_messages,
+ send_messages: permissions.send_messages,
+ }
+ }
+}
+
+impl TryFrom<ffi::StreamPermissions> for RustStreamPermissions {
+ type Error = String;
+
+ fn try_from(permissions: ffi::StreamPermissions) -> Result<Self, Self::Error> {
+ let mut topics = BTreeMap::new();
+ for entry in permissions.topics {
+ let topic_id = entry.topic_id as usize;
+ if topics
+ .insert(topic_id, RustTopicPermissions::from(entry.permissions))
+ .is_some()
+ {
+ return Err(format!("duplicate topic permission ID: {topic_id}"));
+ }
+ }
+ let topics = (!topics.is_empty()).then_some(topics);
+
+ Ok(RustStreamPermissions {
+ manage_stream: permissions.manage_stream,
+ read_stream: permissions.read_stream,
+ manage_topics: permissions.manage_topics,
+ read_topics: permissions.read_topics,
+ poll_messages: permissions.poll_messages,
+ send_messages: permissions.send_messages,
+ topics,
+ })
+ }
+}
+
+impl TryFrom<ffi::Permissions> for RustPermissions {
Review Comment:
This will be difficult to do -- the current tests only test the final cpp library, and there is no wiring in bazel or in the CI to actually run rust tests. We'll need to explicitly call cargo to run the tests from bazel to do this
##########
foreign/cpp/src/type_conversion.rs:
##########
@@ -314,12 +433,36 @@ impl From<RustConsumerGroupDetails> for ffi::ConsumerGroupDetails {
}
}
-impl From<RustIggyMessage> for ffi::IggyMessagePolled {
- fn from(message: RustIggyMessage) -> Self {
+impl TryFrom<RustIggyMessage> for ffi::IggyMessagePolled {
+ type Error = String;
+
+ fn try_from(message: RustIggyMessage) -> Result<Self, Self::Error> {
let id_bytes = message.header.id.to_le_bytes();
let id_lo = u64::from_le_bytes(id_bytes[0..8].try_into().unwrap());
let id_hi = u64::from_le_bytes(id_bytes[8..16].try_into().unwrap());
- ffi::IggyMessagePolled {
+ let user_headers = match message.user_headers {
Review Comment:
added comment
##########
foreign/cpp/src/type_conversion.rs:
##########
@@ -330,45 +473,352 @@ impl From<RustIggyMessage> for ffi::IggyMessagePolled {
payload_length: message.header.payload_length,
reserved: message.header.reserved,
payload: message.payload.to_vec(),
- user_headers: message
- .user_headers
- .map(|headers| headers.to_vec())
- .unwrap_or_default(),
+ user_headers,
+ })
+ }
+}
+
+impl From<RustHeaderEntry> for ffi::HeaderEntry {
+ fn from(entry: RustHeaderEntry) -> Self {
+ ffi::HeaderEntry {
+ key: ffi::HeaderField {
+ kind: entry.key.kind().as_code(),
+ value: entry.key.as_bytes().to_vec(),
+ },
+ value: ffi::HeaderField {
+ kind: entry.value.kind().as_code(),
+ value: entry.value.as_bytes().to_vec(),
+ },
}
}
}
+impl TryFrom<ffi::HeaderEntry> for RustHeaderEntry {
Review Comment:
done
##########
foreign/cpp/src/type_conversion.rs:
##########
@@ -330,45 +473,352 @@ impl From<RustIggyMessage> for ffi::IggyMessagePolled {
payload_length: message.header.payload_length,
reserved: message.header.reserved,
payload: message.payload.to_vec(),
- user_headers: message
- .user_headers
- .map(|headers| headers.to_vec())
- .unwrap_or_default(),
+ user_headers,
+ })
+ }
+}
+
+impl From<RustHeaderEntry> for ffi::HeaderEntry {
Review Comment:
done
##########
foreign/cpp/src/type_conversion.rs:
##########
@@ -330,45 +473,352 @@ impl From<RustIggyMessage> for ffi::IggyMessagePolled {
payload_length: message.header.payload_length,
reserved: message.header.reserved,
payload: message.payload.to_vec(),
- user_headers: message
- .user_headers
- .map(|headers| headers.to_vec())
- .unwrap_or_default(),
+ user_headers,
+ })
+ }
+}
+
+impl From<RustHeaderEntry> for ffi::HeaderEntry {
+ fn from(entry: RustHeaderEntry) -> Self {
+ ffi::HeaderEntry {
+ key: ffi::HeaderField {
+ kind: entry.key.kind().as_code(),
+ value: entry.key.as_bytes().to_vec(),
+ },
+ value: ffi::HeaderField {
+ kind: entry.value.kind().as_code(),
+ value: entry.value.as_bytes().to_vec(),
+ },
}
}
}
+impl TryFrom<ffi::HeaderEntry> for RustHeaderEntry {
+ type Error = String;
+
+ fn try_from(entry: ffi::HeaderEntry) -> Result<Self, Self::Error> {
+ Ok(RustHeaderEntry {
+ key: {
+ let kind = RustHeaderKind::from_code(entry.key.kind)
+ .map_err(|error| format!("Could not convert header field: {error}"))?;
+
+ match kind {
+ RustHeaderKind::Raw => RustHeaderKey::try_from(entry.key.value)
+ .map_err(|error| format!("Could not convert header field: {error}"))?,
+ RustHeaderKind::String => {
+ let value = String::from_utf8(entry.key.value).map_err(|_| {
+ "Could not convert header field: invalid UTF-8 string".to_string()
+ })?;
+ RustHeaderKey::try_from(value)
+ .map_err(|error| format!("Could not convert header field: {error}"))?
+ }
+ RustHeaderKind::Bool => match entry.key.value.as_slice() {
+ [0] => RustHeaderKey::from(false),
+ [1] => RustHeaderKey::from(true),
+ _ => {
+ return Err(
+ "Could not convert header field: bool values must encode as 0 or 1"
+ .to_string(),
+ );
+ }
+ },
+ RustHeaderKind::Int8 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(i8::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int8 values require exactly 1 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Int16 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(i16::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int16 values require exactly 2 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Int32 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(i32::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int32 values require exactly 4 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Int64 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(i64::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int64 values require exactly 8 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Int128 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(i128::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int128 values require exactly 16 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint8 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(u8::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint8 values require exactly 1 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint16 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(u16::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint16 values require exactly 2 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint32 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(u32::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint32 values require exactly 4 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint64 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(u64::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint64 values require exactly 8 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint128 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(u128::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint128 values require exactly 16 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Float32 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(f32::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: float32 values require exactly 4 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Float64 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(f64::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: float64 values require exactly 8 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ }
+ },
+ value: {
+ let kind = RustHeaderKind::from_code(entry.value.kind)
+ .map_err(|error| format!("Could not convert header field: {error}"))?;
+
+ match kind {
+ RustHeaderKind::Raw => RustHeaderValue::try_from(entry.value.value)
+ .map_err(|error| format!("Could not convert header field: {error}"))?,
+ RustHeaderKind::String => {
+ let value = String::from_utf8(entry.value.value).map_err(|_| {
+ "Could not convert header field: invalid UTF-8 string".to_string()
+ })?;
+ RustHeaderValue::try_from(value)
+ .map_err(|error| format!("Could not convert header field: {error}"))?
+ }
+ RustHeaderKind::Bool => match entry.value.value.as_slice() {
+ [0] => RustHeaderValue::from(false),
+ [1] => RustHeaderValue::from(true),
+ _ => {
+ return Err(
+ "Could not convert header field: bool values must encode as 0 or 1"
+ .to_string(),
+ );
+ }
+ },
+ RustHeaderKind::Int8 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(i8::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int8 values require exactly 1 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Int16 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(i16::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int16 values require exactly 2 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Int32 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(i32::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int32 values require exactly 4 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Int64 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(i64::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int64 values require exactly 8 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Int128 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(i128::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int128 values require exactly 16 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint8 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(u8::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint8 values require exactly 1 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint16 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(u16::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint16 values require exactly 2 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint32 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(u32::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint32 values require exactly 4 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint64 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(u64::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint64 values require exactly 8 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint128 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(u128::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint128 values require exactly 16 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Float32 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(f32::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: float32 values require exactly 4 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Float64 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(f64::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: float64 values require exactly 8 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ }
+ },
+ })
+ }
+}
+
impl TryFrom<ffi::IggyMessageToSend> for RustIggyMessage {
type Error = String;
fn try_from(message: ffi::IggyMessageToSend) -> Result<Self, Self::Error> {
- if !message.user_headers.is_empty() {
- return Err(
- "Could not convert message: user_headers are not yet supported in the C++ SDK"
- .to_string(),
- );
+ let mut user_headers = BTreeMap::new();
+ for entry in message.user_headers {
+ let header_entry = RustHeaderEntry::try_from(entry)
+ .map_err(|error| format!("Could not convert message user headers: {error}"))?;
+ if user_headers
+ .insert(header_entry.key, header_entry.value)
+ .is_some()
+ {
+ return Err(
+ "Could not convert message user headers: duplicate header key".to_string(),
+ );
+ }
}
let id = ((message.id_hi as u128) << 64) | (message.id_lo as u128);
let payload = Bytes::from(message.payload);
- RustIggyMessage::builder()
- .id(id)
- .payload(payload)
- .build()
- .map_err(|error| format!("Could not convert message: {error}"))
+ if user_headers.is_empty() {
Review Comment:
done
##########
foreign/cpp/src/type_conversion.rs:
##########
@@ -330,45 +473,352 @@ impl From<RustIggyMessage> for ffi::IggyMessagePolled {
payload_length: message.header.payload_length,
reserved: message.header.reserved,
payload: message.payload.to_vec(),
- user_headers: message
- .user_headers
- .map(|headers| headers.to_vec())
- .unwrap_or_default(),
+ user_headers,
+ })
+ }
+}
+
+impl From<RustHeaderEntry> for ffi::HeaderEntry {
+ fn from(entry: RustHeaderEntry) -> Self {
+ ffi::HeaderEntry {
+ key: ffi::HeaderField {
+ kind: entry.key.kind().as_code(),
+ value: entry.key.as_bytes().to_vec(),
+ },
+ value: ffi::HeaderField {
+ kind: entry.value.kind().as_code(),
+ value: entry.value.as_bytes().to_vec(),
+ },
}
}
}
+impl TryFrom<ffi::HeaderEntry> for RustHeaderEntry {
+ type Error = String;
+
+ fn try_from(entry: ffi::HeaderEntry) -> Result<Self, Self::Error> {
+ Ok(RustHeaderEntry {
+ key: {
+ let kind = RustHeaderKind::from_code(entry.key.kind)
+ .map_err(|error| format!("Could not convert header field: {error}"))?;
+
+ match kind {
+ RustHeaderKind::Raw => RustHeaderKey::try_from(entry.key.value)
+ .map_err(|error| format!("Could not convert header field: {error}"))?,
+ RustHeaderKind::String => {
+ let value = String::from_utf8(entry.key.value).map_err(|_| {
+ "Could not convert header field: invalid UTF-8 string".to_string()
+ })?;
+ RustHeaderKey::try_from(value)
+ .map_err(|error| format!("Could not convert header field: {error}"))?
+ }
+ RustHeaderKind::Bool => match entry.key.value.as_slice() {
+ [0] => RustHeaderKey::from(false),
+ [1] => RustHeaderKey::from(true),
+ _ => {
+ return Err(
+ "Could not convert header field: bool values must encode as 0 or 1"
+ .to_string(),
+ );
+ }
+ },
+ RustHeaderKind::Int8 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(i8::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int8 values require exactly 1 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Int16 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(i16::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int16 values require exactly 2 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Int32 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(i32::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int32 values require exactly 4 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Int64 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(i64::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int64 values require exactly 8 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Int128 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(i128::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int128 values require exactly 16 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint8 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(u8::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint8 values require exactly 1 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint16 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(u16::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint16 values require exactly 2 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint32 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(u32::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint32 values require exactly 4 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint64 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(u64::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint64 values require exactly 8 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint128 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(u128::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint128 values require exactly 16 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Float32 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(f32::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: float32 values require exactly 4 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Float64 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(f64::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: float64 values require exactly 8 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ }
+ },
+ value: {
+ let kind = RustHeaderKind::from_code(entry.value.kind)
+ .map_err(|error| format!("Could not convert header field: {error}"))?;
+
+ match kind {
+ RustHeaderKind::Raw => RustHeaderValue::try_from(entry.value.value)
+ .map_err(|error| format!("Could not convert header field: {error}"))?,
+ RustHeaderKind::String => {
+ let value = String::from_utf8(entry.value.value).map_err(|_| {
+ "Could not convert header field: invalid UTF-8 string".to_string()
+ })?;
+ RustHeaderValue::try_from(value)
+ .map_err(|error| format!("Could not convert header field: {error}"))?
+ }
+ RustHeaderKind::Bool => match entry.value.value.as_slice() {
+ [0] => RustHeaderValue::from(false),
+ [1] => RustHeaderValue::from(true),
+ _ => {
+ return Err(
+ "Could not convert header field: bool values must encode as 0 or 1"
+ .to_string(),
+ );
+ }
+ },
+ RustHeaderKind::Int8 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(i8::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int8 values require exactly 1 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Int16 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(i16::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int16 values require exactly 2 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Int32 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(i32::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int32 values require exactly 4 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Int64 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(i64::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int64 values require exactly 8 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Int128 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(i128::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int128 values require exactly 16 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint8 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(u8::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint8 values require exactly 1 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint16 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(u16::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint16 values require exactly 2 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint32 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(u32::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint32 values require exactly 4 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint64 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(u64::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint64 values require exactly 8 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint128 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(u128::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint128 values require exactly 16 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Float32 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(f32::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: float32 values require exactly 4 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Float64 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(f64::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: float64 values require exactly 8 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ }
+ },
+ })
+ }
+}
+
impl TryFrom<ffi::IggyMessageToSend> for RustIggyMessage {
type Error = String;
fn try_from(message: ffi::IggyMessageToSend) -> Result<Self, Self::Error> {
- if !message.user_headers.is_empty() {
- return Err(
- "Could not convert message: user_headers are not yet supported in the C++ SDK"
- .to_string(),
- );
+ let mut user_headers = BTreeMap::new();
Review Comment:
added todo to mention this in high level sdk doc
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(cpp): add functions related to client lifecycle, and message headers [iggy]
Posted by "slbotbm (via GitHub)" <gi...@apache.org>.
slbotbm commented on PR #3610:
URL: https://github.com/apache/iggy/pull/3610#issuecomment-4999627278
/ready
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(cpp): add functions related to client lifecycle, and message headers [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio merged PR #3610:
URL: https://github.com/apache/iggy/pull/3610
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(cpp): add functions related to client lifecycle, and message headers [iggy]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #3610:
URL: https://github.com/apache/iggy/pull/3610#issuecomment-4886294268
<!-- iggy-pr-triage-welcome -->
Thanks for the PR. It is labeled `S-waiting-on-review` and queued for review.
Slash commands (own line, regular comment) move it around the queue:
- `/ready` - back to `S-waiting-on-review` after addressing feedback
- `/author` - flip to `S-waiting-on-author` while you finish changes
- `/request-review @user-or-team` - request a reviewer
See [CONTRIBUTING.md](https://github.com/apache/iggy/blob/master/CONTRIBUTING.md#pr-triage-commands) for details.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(cpp): add functions related to client lifecycle, and message headers [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on code in PR #3610:
URL: https://github.com/apache/iggy/pull/3610#discussion_r3601732261
##########
foreign/cpp/src/type_conversion.rs:
##########
@@ -188,6 +194,119 @@ impl From<RustStats> for ffi::Stats {
}
}
+impl From<RustTransportEndpoints> for ffi::TransportEndpoints {
+ fn from(endpoints: RustTransportEndpoints) -> Self {
+ ffi::TransportEndpoints {
+ tcp: endpoints.tcp,
+ quic: endpoints.quic,
+ http: endpoints.http,
+ websocket: endpoints.websocket,
+ }
+ }
+}
+
+impl From<RustClusterNode> for ffi::ClusterNode {
+ fn from(node: RustClusterNode) -> Self {
+ ffi::ClusterNode {
+ name: node.name,
+ ip: node.ip,
+ endpoints: ffi::TransportEndpoints::from(node.endpoints),
+ role: node.role.to_string(),
+ status: node.status.to_string(),
+ }
+ }
+}
+
+impl From<RustClusterMetadata> for ffi::ClusterMetadata {
+ fn from(metadata: RustClusterMetadata) -> Self {
+ ffi::ClusterMetadata {
+ name: metadata.name,
+ nodes: metadata
+ .nodes
+ .into_iter()
+ .map(ffi::ClusterNode::from)
+ .collect(),
+ }
+ }
+}
+
+impl From<ffi::GlobalPermissions> for RustGlobalPermissions {
+ fn from(permissions: ffi::GlobalPermissions) -> Self {
+ RustGlobalPermissions {
+ manage_servers: permissions.manage_servers,
+ read_servers: permissions.read_servers,
+ manage_users: permissions.manage_users,
+ read_users: permissions.read_users,
+ manage_streams: permissions.manage_streams,
+ read_streams: permissions.read_streams,
+ manage_topics: permissions.manage_topics,
+ read_topics: permissions.read_topics,
+ poll_messages: permissions.poll_messages,
+ send_messages: permissions.send_messages,
+ }
+ }
+}
+
+impl From<ffi::TopicPermissions> for RustTopicPermissions {
+ fn from(permissions: ffi::TopicPermissions) -> Self {
+ RustTopicPermissions {
+ manage_topic: permissions.manage_topic,
+ read_topic: permissions.read_topic,
+ poll_messages: permissions.poll_messages,
+ send_messages: permissions.send_messages,
+ }
+ }
+}
+
+impl TryFrom<ffi::StreamPermissions> for RustStreamPermissions {
+ type Error = String;
+
+ fn try_from(permissions: ffi::StreamPermissions) -> Result<Self, Self::Error> {
+ let mut topics = BTreeMap::new();
+ for entry in permissions.topics {
+ let topic_id = entry.topic_id as usize;
+ if topics
+ .insert(topic_id, RustTopicPermissions::from(entry.permissions))
+ .is_some()
+ {
+ return Err(format!("duplicate topic permission ID: {topic_id}"));
+ }
+ }
+ let topics = (!topics.is_empty()).then_some(topics);
+
+ Ok(RustStreamPermissions {
+ manage_stream: permissions.manage_stream,
+ read_stream: permissions.read_stream,
+ manage_topics: permissions.manage_topics,
+ read_topics: permissions.read_topics,
+ poll_messages: permissions.poll_messages,
+ send_messages: permissions.send_messages,
+ topics,
+ })
+ }
+}
+
+impl TryFrom<ffi::Permissions> for RustPermissions {
Review Comment:
alright. lets leave it as is.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(cpp): add functions related to client lifecycle, and message headers [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on code in PR #3610:
URL: https://github.com/apache/iggy/pull/3610#discussion_r3598487630
##########
foreign/cpp/src/type_conversion.rs:
##########
@@ -188,6 +194,119 @@ impl From<RustStats> for ffi::Stats {
}
}
+impl From<RustTransportEndpoints> for ffi::TransportEndpoints {
+ fn from(endpoints: RustTransportEndpoints) -> Self {
+ ffi::TransportEndpoints {
+ tcp: endpoints.tcp,
+ quic: endpoints.quic,
+ http: endpoints.http,
+ websocket: endpoints.websocket,
+ }
+ }
+}
+
+impl From<RustClusterNode> for ffi::ClusterNode {
+ fn from(node: RustClusterNode) -> Self {
+ ffi::ClusterNode {
+ name: node.name,
+ ip: node.ip,
+ endpoints: ffi::TransportEndpoints::from(node.endpoints),
+ role: node.role.to_string(),
+ status: node.status.to_string(),
+ }
+ }
+}
+
+impl From<RustClusterMetadata> for ffi::ClusterMetadata {
+ fn from(metadata: RustClusterMetadata) -> Self {
+ ffi::ClusterMetadata {
+ name: metadata.name,
+ nodes: metadata
+ .nodes
+ .into_iter()
+ .map(ffi::ClusterNode::from)
+ .collect(),
+ }
+ }
+}
+
+impl From<ffi::GlobalPermissions> for RustGlobalPermissions {
+ fn from(permissions: ffi::GlobalPermissions) -> Self {
+ RustGlobalPermissions {
+ manage_servers: permissions.manage_servers,
+ read_servers: permissions.read_servers,
+ manage_users: permissions.manage_users,
+ read_users: permissions.read_users,
+ manage_streams: permissions.manage_streams,
+ read_streams: permissions.read_streams,
+ manage_topics: permissions.manage_topics,
+ read_topics: permissions.read_topics,
+ poll_messages: permissions.poll_messages,
+ send_messages: permissions.send_messages,
+ }
+ }
+}
+
+impl From<ffi::TopicPermissions> for RustTopicPermissions {
+ fn from(permissions: ffi::TopicPermissions) -> Self {
+ RustTopicPermissions {
+ manage_topic: permissions.manage_topic,
+ read_topic: permissions.read_topic,
+ poll_messages: permissions.poll_messages,
+ send_messages: permissions.send_messages,
+ }
+ }
+}
+
+impl TryFrom<ffi::StreamPermissions> for RustStreamPermissions {
+ type Error = String;
+
+ fn try_from(permissions: ffi::StreamPermissions) -> Result<Self, Self::Error> {
+ let mut topics = BTreeMap::new();
+ for entry in permissions.topics {
+ let topic_id = entry.topic_id as usize;
+ if topics
+ .insert(topic_id, RustTopicPermissions::from(entry.permissions))
+ .is_some()
+ {
+ return Err(format!("duplicate topic permission ID: {topic_id}"));
+ }
+ }
+ let topics = (!topics.is_empty()).then_some(topics);
+
+ Ok(RustStreamPermissions {
+ manage_stream: permissions.manage_stream,
+ read_stream: permissions.read_stream,
+ manage_topics: permissions.manage_topics,
+ read_topics: permissions.read_topics,
+ poll_messages: permissions.poll_messages,
+ send_messages: permissions.send_messages,
+ topics,
+ })
+ }
+}
+
+impl TryFrom<ffi::Permissions> for RustPermissions {
Review Comment:
this whole permissions conversion chain (`TryFrom<ffi::Permissions>`, `TryFrom<ffi::StreamPermissions>`, and the `From` impls above) plus `update_permissions` ship with no test. it's an authz path and the `has_permissions` false -> `None` vs true + all-false -> `Some` distinction is load-bearing, so it's worth a direct unit test on `TryFrom<ffi::Permissions>` - duplicate-id rejection, empty -> `None`, and the None/Some cases. doesn't need `create_user` to exist yet since the conversion is pure.
##########
foreign/cpp/src/lib.rs:
##########
@@ -88,11 +88,40 @@ mod ffi {
topics_count: u32,
}
+ #[repr(u8)]
+ enum HeaderKind {
+ Raw = 1,
+ String = 2,
+ Bool = 3,
+ Int8 = 4,
+ Int16 = 5,
+ Int32 = 6,
+ Int64 = 7,
+ Int128 = 8,
+ UInt8 = 9,
Review Comment:
the header kind variants are named `UInt8..UInt128` here, but every other SDK and the Rust core use `Uint8..Uint128` (lowercase i) - see `HeaderKind` in iggy_common, plus the C# and Java SDKs. the numeric codes match so there's no wire impact, but this is the name C++ users type, and renaming a public enum after release is breaking. cheap to align to `Uint8` now.
##########
foreign/cpp/src/type_conversion.rs:
##########
@@ -314,12 +433,36 @@ impl From<RustConsumerGroupDetails> for ffi::ConsumerGroupDetails {
}
}
-impl From<RustIggyMessage> for ffi::IggyMessagePolled {
- fn from(message: RustIggyMessage) -> Self {
+impl TryFrom<RustIggyMessage> for ffi::IggyMessagePolled {
+ type Error = String;
+
+ fn try_from(message: RustIggyMessage) -> Result<Self, Self::Error> {
let id_bytes = message.header.id.to_le_bytes();
let id_lo = u64::from_le_bytes(id_bytes[0..8].try_into().unwrap());
let id_hi = u64::from_le_bytes(id_bytes[8..16].try_into().unwrap());
- ffi::IggyMessagePolled {
+ let user_headers = match message.user_headers {
Review Comment:
poll decode here is structural-only (via `WireUserHeaders`), so it accepts unknown kind codes 16-255 and wrong-width fixed types, whereas the Rust SDK poll is strict (`user_headers_from_wire` checks kind + width). same stored bytes read differently across SDKs. looks intentional for forward-compat - worth a comment saying so.
##########
foreign/cpp/src/type_conversion.rs:
##########
@@ -330,45 +473,352 @@ impl From<RustIggyMessage> for ffi::IggyMessagePolled {
payload_length: message.header.payload_length,
reserved: message.header.reserved,
payload: message.payload.to_vec(),
- user_headers: message
- .user_headers
- .map(|headers| headers.to_vec())
- .unwrap_or_default(),
+ user_headers,
+ })
+ }
+}
+
+impl From<RustHeaderEntry> for ffi::HeaderEntry {
+ fn from(entry: RustHeaderEntry) -> Self {
+ ffi::HeaderEntry {
+ key: ffi::HeaderField {
+ kind: entry.key.kind().as_code(),
+ value: entry.key.as_bytes().to_vec(),
+ },
+ value: ffi::HeaderField {
+ kind: entry.value.kind().as_code(),
+ value: entry.value.as_bytes().to_vec(),
+ },
}
}
}
+impl TryFrom<ffi::HeaderEntry> for RustHeaderEntry {
+ type Error = String;
+
+ fn try_from(entry: ffi::HeaderEntry) -> Result<Self, Self::Error> {
+ Ok(RustHeaderEntry {
+ key: {
+ let kind = RustHeaderKind::from_code(entry.key.kind)
+ .map_err(|error| format!("Could not convert header field: {error}"))?;
+
+ match kind {
+ RustHeaderKind::Raw => RustHeaderKey::try_from(entry.key.value)
+ .map_err(|error| format!("Could not convert header field: {error}"))?,
+ RustHeaderKind::String => {
+ let value = String::from_utf8(entry.key.value).map_err(|_| {
+ "Could not convert header field: invalid UTF-8 string".to_string()
+ })?;
+ RustHeaderKey::try_from(value)
+ .map_err(|error| format!("Could not convert header field: {error}"))?
+ }
+ RustHeaderKind::Bool => match entry.key.value.as_slice() {
+ [0] => RustHeaderKey::from(false),
+ [1] => RustHeaderKey::from(true),
+ _ => {
+ return Err(
+ "Could not convert header field: bool values must encode as 0 or 1"
+ .to_string(),
+ );
+ }
+ },
+ RustHeaderKind::Int8 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(i8::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int8 values require exactly 1 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Int16 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(i16::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int16 values require exactly 2 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Int32 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(i32::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int32 values require exactly 4 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Int64 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(i64::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int64 values require exactly 8 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Int128 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(i128::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int128 values require exactly 16 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint8 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(u8::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint8 values require exactly 1 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint16 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(u16::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint16 values require exactly 2 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint32 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(u32::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint32 values require exactly 4 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint64 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(u64::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint64 values require exactly 8 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint128 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(u128::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint128 values require exactly 16 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Float32 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(f32::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: float32 values require exactly 4 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Float64 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(f64::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: float64 values require exactly 8 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ }
+ },
+ value: {
+ let kind = RustHeaderKind::from_code(entry.value.kind)
+ .map_err(|error| format!("Could not convert header field: {error}"))?;
+
+ match kind {
+ RustHeaderKind::Raw => RustHeaderValue::try_from(entry.value.value)
+ .map_err(|error| format!("Could not convert header field: {error}"))?,
+ RustHeaderKind::String => {
+ let value = String::from_utf8(entry.value.value).map_err(|_| {
+ "Could not convert header field: invalid UTF-8 string".to_string()
+ })?;
+ RustHeaderValue::try_from(value)
+ .map_err(|error| format!("Could not convert header field: {error}"))?
+ }
+ RustHeaderKind::Bool => match entry.value.value.as_slice() {
+ [0] => RustHeaderValue::from(false),
+ [1] => RustHeaderValue::from(true),
+ _ => {
+ return Err(
+ "Could not convert header field: bool values must encode as 0 or 1"
+ .to_string(),
+ );
+ }
+ },
+ RustHeaderKind::Int8 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(i8::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int8 values require exactly 1 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Int16 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(i16::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int16 values require exactly 2 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Int32 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(i32::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int32 values require exactly 4 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Int64 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(i64::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int64 values require exactly 8 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Int128 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(i128::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int128 values require exactly 16 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint8 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(u8::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint8 values require exactly 1 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint16 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(u16::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint16 values require exactly 2 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint32 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(u32::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint32 values require exactly 4 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint64 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(u64::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint64 values require exactly 8 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint128 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(u128::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint128 values require exactly 16 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Float32 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(f32::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: float32 values require exactly 4 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Float64 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(f64::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: float64 values require exactly 8 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ }
+ },
+ })
+ }
+}
+
impl TryFrom<ffi::IggyMessageToSend> for RustIggyMessage {
type Error = String;
fn try_from(message: ffi::IggyMessageToSend) -> Result<Self, Self::Error> {
- if !message.user_headers.is_empty() {
- return Err(
- "Could not convert message: user_headers are not yet supported in the C++ SDK"
- .to_string(),
- );
+ let mut user_headers = BTreeMap::new();
+ for entry in message.user_headers {
+ let header_entry = RustHeaderEntry::try_from(entry)
+ .map_err(|error| format!("Could not convert message user headers: {error}"))?;
+ if user_headers
+ .insert(header_entry.key, header_entry.value)
+ .is_some()
+ {
+ return Err(
+ "Could not convert message user headers: duplicate header key".to_string(),
+ );
+ }
}
let id = ((message.id_hi as u128) << 64) | (message.id_lo as u128);
let payload = Bytes::from(message.payload);
- RustIggyMessage::builder()
- .id(id)
- .payload(payload)
- .build()
- .map_err(|error| format!("Could not convert message: {error}"))
+ if user_headers.is_empty() {
Review Comment:
the empty/non-empty branch here duplicates the whole builder chain just to conditionally add headers. the builder is bon and `user_headers` is an `Option`, so it generates `maybe_user_headers` - this collapses to one chain: `.maybe_user_headers((!user_headers.is_empty()).then_some(user_headers))`. same result, empty map stays `None`.
##########
foreign/cpp/src/type_conversion.rs:
##########
@@ -314,12 +433,36 @@ impl From<RustConsumerGroupDetails> for ffi::ConsumerGroupDetails {
}
}
-impl From<RustIggyMessage> for ffi::IggyMessagePolled {
- fn from(message: RustIggyMessage) -> Self {
+impl TryFrom<RustIggyMessage> for ffi::IggyMessagePolled {
+ type Error = String;
+
+ fn try_from(message: RustIggyMessage) -> Result<Self, Self::Error> {
let id_bytes = message.header.id.to_le_bytes();
let id_lo = u64::from_le_bytes(id_bytes[0..8].try_into().unwrap());
let id_hi = u64::from_le_bytes(id_bytes[8..16].try_into().unwrap());
- ffi::IggyMessagePolled {
+ let user_headers = match message.user_headers {
+ Some(raw_headers) => {
+ let wire_headers = WireUserHeaders::from_bytes(raw_headers).map_err(|error| {
Review Comment:
poll decode propagates a `WireUserHeaders::from_bytes` error here, and `TryFrom<RustPolledMessages>` collects with `collect::<Result<Vec,_>>()?` (line 821), so one message with a malformed user-header TLV fails the entire poll batch - every good message in that batch is dropped with it. the server does not validate user-header TLV on the produce path (no `validate_user_headers` / `WireUserHeaders` call anywhere in core/server), so a non-first-party or buggy producer can store a header that then breaks C++ polls of that range. with offset-based polling the consumer re-fetches the same range and keeps failing (stuck); with auto_commit the batch gets skipped and its good messages are lost. the reference Rust SDK handles this gracefully - `user_headers_map()` warns and returns `Ok(None)`, keeping the message readable. suggest mirroring that: match `from_bytes`, on error warn and emit empty headers, keep the message.
##########
foreign/cpp/src/type_conversion.rs:
##########
@@ -330,45 +473,352 @@ impl From<RustIggyMessage> for ffi::IggyMessagePolled {
payload_length: message.header.payload_length,
reserved: message.header.reserved,
payload: message.payload.to_vec(),
- user_headers: message
- .user_headers
- .map(|headers| headers.to_vec())
- .unwrap_or_default(),
+ user_headers,
+ })
+ }
+}
+
+impl From<RustHeaderEntry> for ffi::HeaderEntry {
+ fn from(entry: RustHeaderEntry) -> Self {
+ ffi::HeaderEntry {
+ key: ffi::HeaderField {
+ kind: entry.key.kind().as_code(),
+ value: entry.key.as_bytes().to_vec(),
+ },
+ value: ffi::HeaderField {
+ kind: entry.value.kind().as_code(),
+ value: entry.value.as_bytes().to_vec(),
+ },
}
}
}
+impl TryFrom<ffi::HeaderEntry> for RustHeaderEntry {
Review Comment:
the key block and value block here are two nearly identical 15-arm matches that differ only in producing `HeaderKey` vs `HeaderValue` - and both are `HeaderField<T>` with all the `From` / `TryFrom` impls generic over the marker. this collapses to one generic helper `fn decode_field<T>(kind: u8, value: Vec<u8>) -> Result<HeaderField<T>, String>` called for key and value, cutting around 135 lines and folding the duplicated error strings into one spot.
##########
foreign/cpp/src/type_conversion.rs:
##########
@@ -330,45 +473,352 @@ impl From<RustIggyMessage> for ffi::IggyMessagePolled {
payload_length: message.header.payload_length,
reserved: message.header.reserved,
payload: message.payload.to_vec(),
- user_headers: message
- .user_headers
- .map(|headers| headers.to_vec())
- .unwrap_or_default(),
+ user_headers,
+ })
+ }
+}
+
+impl From<RustHeaderEntry> for ffi::HeaderEntry {
Review Comment:
this `From<RustHeaderEntry> for ffi::HeaderEntry` has no callers - the poll path builds `ffi::HeaderEntry` inline from the `WireUserHeaders` iterator, and `RustIggyMessage` stores headers as `Option<Bytes>`, never `Vec<RustHeaderEntry>`. dead, can be deleted.
##########
foreign/cpp/src/type_conversion.rs:
##########
@@ -330,45 +473,352 @@ impl From<RustIggyMessage> for ffi::IggyMessagePolled {
payload_length: message.header.payload_length,
reserved: message.header.reserved,
payload: message.payload.to_vec(),
- user_headers: message
- .user_headers
- .map(|headers| headers.to_vec())
- .unwrap_or_default(),
+ user_headers,
+ })
+ }
+}
+
+impl From<RustHeaderEntry> for ffi::HeaderEntry {
+ fn from(entry: RustHeaderEntry) -> Self {
+ ffi::HeaderEntry {
+ key: ffi::HeaderField {
+ kind: entry.key.kind().as_code(),
+ value: entry.key.as_bytes().to_vec(),
+ },
+ value: ffi::HeaderField {
+ kind: entry.value.kind().as_code(),
+ value: entry.value.as_bytes().to_vec(),
+ },
}
}
}
+impl TryFrom<ffi::HeaderEntry> for RustHeaderEntry {
+ type Error = String;
+
+ fn try_from(entry: ffi::HeaderEntry) -> Result<Self, Self::Error> {
+ Ok(RustHeaderEntry {
+ key: {
+ let kind = RustHeaderKind::from_code(entry.key.kind)
+ .map_err(|error| format!("Could not convert header field: {error}"))?;
+
+ match kind {
+ RustHeaderKind::Raw => RustHeaderKey::try_from(entry.key.value)
+ .map_err(|error| format!("Could not convert header field: {error}"))?,
+ RustHeaderKind::String => {
+ let value = String::from_utf8(entry.key.value).map_err(|_| {
+ "Could not convert header field: invalid UTF-8 string".to_string()
+ })?;
+ RustHeaderKey::try_from(value)
+ .map_err(|error| format!("Could not convert header field: {error}"))?
+ }
+ RustHeaderKind::Bool => match entry.key.value.as_slice() {
+ [0] => RustHeaderKey::from(false),
+ [1] => RustHeaderKey::from(true),
+ _ => {
+ return Err(
+ "Could not convert header field: bool values must encode as 0 or 1"
+ .to_string(),
+ );
+ }
+ },
+ RustHeaderKind::Int8 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(i8::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int8 values require exactly 1 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Int16 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(i16::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int16 values require exactly 2 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Int32 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(i32::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int32 values require exactly 4 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Int64 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(i64::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int64 values require exactly 8 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Int128 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(i128::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int128 values require exactly 16 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint8 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(u8::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint8 values require exactly 1 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint16 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(u16::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint16 values require exactly 2 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint32 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(u32::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint32 values require exactly 4 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint64 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(u64::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint64 values require exactly 8 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint128 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(u128::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint128 values require exactly 16 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Float32 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(f32::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: float32 values require exactly 4 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Float64 => match entry.key.value.try_into() {
+ Ok(bytes) => RustHeaderKey::from(f64::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: float64 values require exactly 8 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ }
+ },
+ value: {
+ let kind = RustHeaderKind::from_code(entry.value.kind)
+ .map_err(|error| format!("Could not convert header field: {error}"))?;
+
+ match kind {
+ RustHeaderKind::Raw => RustHeaderValue::try_from(entry.value.value)
+ .map_err(|error| format!("Could not convert header field: {error}"))?,
+ RustHeaderKind::String => {
+ let value = String::from_utf8(entry.value.value).map_err(|_| {
+ "Could not convert header field: invalid UTF-8 string".to_string()
+ })?;
+ RustHeaderValue::try_from(value)
+ .map_err(|error| format!("Could not convert header field: {error}"))?
+ }
+ RustHeaderKind::Bool => match entry.value.value.as_slice() {
+ [0] => RustHeaderValue::from(false),
+ [1] => RustHeaderValue::from(true),
+ _ => {
+ return Err(
+ "Could not convert header field: bool values must encode as 0 or 1"
+ .to_string(),
+ );
+ }
+ },
+ RustHeaderKind::Int8 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(i8::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int8 values require exactly 1 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Int16 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(i16::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int16 values require exactly 2 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Int32 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(i32::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int32 values require exactly 4 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Int64 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(i64::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int64 values require exactly 8 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Int128 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(i128::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: int128 values require exactly 16 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint8 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(u8::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint8 values require exactly 1 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint16 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(u16::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint16 values require exactly 2 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint32 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(u32::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint32 values require exactly 4 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint64 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(u64::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint64 values require exactly 8 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Uint128 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(u128::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: uint128 values require exactly 16 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Float32 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(f32::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: float32 values require exactly 4 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ RustHeaderKind::Float64 => match entry.value.value.try_into() {
+ Ok(bytes) => RustHeaderValue::from(f64::from_le_bytes(bytes)),
+ Err(value) => {
+ return Err(format!(
+ "Could not convert header field: float64 values require exactly 8 bytes, got {}",
+ value.len()
+ ));
+ }
+ },
+ }
+ },
+ })
+ }
+}
+
impl TryFrom<ffi::IggyMessageToSend> for RustIggyMessage {
type Error = String;
fn try_from(message: ffi::IggyMessageToSend) -> Result<Self, Self::Error> {
- if !message.user_headers.is_empty() {
- return Err(
- "Could not convert message: user_headers are not yet supported in the C++ SDK"
- .to_string(),
- );
+ let mut user_headers = BTreeMap::new();
Review Comment:
worth a note that user headers are unordered and keys unique: they go into a `BTreeMap`, so the C++ `Vec<HeaderEntry>` insertion order is not preserved on send or poll (sorted by kind+value), and duplicate keys are rejected. the `Vec` shape reads like it keeps order.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org