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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
