| // Licensed to the Apache Software Foundation (ASF) under one |
| // or more contributor license agreements. See the NOTICE file |
| // distributed with this work for additional information |
| // regarding copyright ownership. The ASF licenses this file |
| // to you under the Apache License, Version 2.0 (the |
| // "License"); you may not use this file except in compliance |
| // with the License. You may obtain a copy of the License at |
| // |
| // http://www.apache.org/licenses/LICENSE-2.0 |
| // |
| // Unless required by applicable law or agreed to in writing, |
| // software distributed under the License is distributed on an |
| // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| // KIND, either express or implied. See the License for the |
| // specific language governing permissions and limitations |
| // under the License. |
| |
| //! Conversions between `iggy_binary_protocol` wire types and `iggy_common` domain types. |
| //! |
| //! Where the orphan rule allows it, conversions are `From`/`TryFrom` impls on domain types. |
| //! Collection conversions (`Vec<T>`) and tuple conversions use free functions instead, |
| //! since neither the container nor the wire type is local. |
| |
| use crate::{ |
| CacheMetrics, CacheMetricsKey, ClientInfo, ClientInfoDetails, ClusterMetadata, ClusterNode, |
| ClusterNodeRole, ClusterNodeStatus, CompressionAlgorithm, Consumer, ConsumerGroup, |
| ConsumerGroupDetails, ConsumerGroupInfo, ConsumerGroupMember, ConsumerOffsetInfo, |
| GlobalPermissions, HeaderKey, HeaderKind, HeaderValue, IdKind, IdentityInfo, IggyByteSize, |
| IggyError, IggyExpiry, MaxTopicSize, OptionSpec, OptionValue, OptionsProvenance, Partition, |
| Permissions, PersonalAccessTokenInfo, RawPersonalAccessToken, ResourceOptions, Stats, Stream, |
| StreamDetails, StreamPermissions, Topic, TopicDetails, TopicPermissions, TransportEndpoints, |
| UserInfo, UserInfoDetails, UserStatus, |
| }; |
| use bytes::{BufMut, BytesMut}; |
| use iggy_binary_protocol::primitives::options::{MAX_OPTIONS, MAX_OPTIONS_BYTES, WireOptions}; |
| use iggy_binary_protocol::primitives::permissions::{ |
| WireGlobalPermissions, WirePermissions, WireStreamPermissions, WireTopicPermissions, |
| }; |
| use iggy_binary_protocol::responses::clients::client_response::{ |
| ClientResponse, ConsumerGroupInfoResponse, |
| }; |
| use iggy_binary_protocol::responses::clients::get_client::ClientDetailsResponse; |
| use iggy_binary_protocol::responses::clients::get_clients::GetClientsResponse; |
| use iggy_binary_protocol::responses::consumer_groups::consumer_group_response::ConsumerGroupResponse; |
| use iggy_binary_protocol::responses::consumer_groups::get_consumer_group::{ |
| ConsumerGroupDetailsResponse, ConsumerGroupMemberResponse, |
| }; |
| use iggy_binary_protocol::responses::consumer_groups::get_consumer_groups::GetConsumerGroupsResponse; |
| use iggy_binary_protocol::responses::consumer_offsets::get_consumer_offset::ConsumerOffsetResponse; |
| use iggy_binary_protocol::responses::personal_access_tokens::create_personal_access_token::RawPersonalAccessTokenResponse; |
| use iggy_binary_protocol::responses::personal_access_tokens::get_personal_access_tokens::{ |
| GetPersonalAccessTokensResponse, PersonalAccessTokenResponse, |
| }; |
| use iggy_binary_protocol::responses::streams::StreamResponse; |
| use iggy_binary_protocol::responses::streams::get_stream::{GetStreamResponse, TopicHeader}; |
| use iggy_binary_protocol::responses::streams::get_streams::GetStreamsResponse; |
| use iggy_binary_protocol::responses::system::get_cluster_metadata::{ |
| ClusterMetadataResponse, ClusterNodeResponse, |
| }; |
| use iggy_binary_protocol::responses::system::get_stats::{CacheMetricEntry, StatsResponse}; |
| use iggy_binary_protocol::responses::topics::get_topic::{GetTopicResponse, PartitionResponse}; |
| use iggy_binary_protocol::responses::topics::get_topics::GetTopicsResponse; |
| use iggy_binary_protocol::responses::users::login_user::IdentityResponse; |
| use iggy_binary_protocol::responses::users::user_response::UserResponse; |
| use iggy_binary_protocol::responses::users::{GetUsersResponse, UserDetailsResponse}; |
| use iggy_binary_protocol::{WireConsumer, WireUserHeaderIterator}; |
| use std::collections::{BTreeMap, HashMap}; |
| |
| /// Sentinel value in the wire protocol indicating no authenticated user. |
| const WIRE_NO_USER_ID: u32 = u32::MAX; |
| |
| // --------------------------------------------------------------------------- |
| // Streams |
| // --------------------------------------------------------------------------- |
| |
| impl TryFrom<StreamResponse> for Stream { |
| type Error = IggyError; |
| |
| fn try_from(w: StreamResponse) -> Result<Self, Self::Error> { |
| Ok(Self { |
| id: w.id, |
| created_at: w.created_at.into(), |
| name: w.name.to_string(), |
| size: IggyByteSize::from(w.size_bytes), |
| messages_count: w.messages_count, |
| topics_count: w.topics_count, |
| options: resource_options_from_wire(&w.options, true)?, |
| }) |
| } |
| } |
| |
| impl TryFrom<GetStreamResponse> for StreamDetails { |
| type Error = IggyError; |
| |
| fn try_from(w: GetStreamResponse) -> Result<Self, Self::Error> { |
| let mut topics: Vec<Topic> = w |
| .topics |
| .into_iter() |
| .map(Topic::try_from) |
| .collect::<Result<_, _>>()?; |
| topics.sort_by_key(|t| t.id); |
| Ok(Self { |
| id: w.stream.id, |
| created_at: w.stream.created_at.into(), |
| name: w.stream.name.to_string(), |
| size: IggyByteSize::from(w.stream.size_bytes), |
| messages_count: w.stream.messages_count, |
| topics_count: w.stream.topics_count, |
| topics, |
| options: resource_options_from_wire(&w.stream.options, true)?, |
| }) |
| } |
| } |
| |
| pub fn streams_from_wire(w: GetStreamsResponse) -> Result<Vec<Stream>, IggyError> { |
| let mut streams: Vec<Stream> = w |
| .streams |
| .into_iter() |
| .map(Stream::try_from) |
| .collect::<Result<_, _>>()?; |
| streams.sort_by_key(|s| s.id); |
| Ok(streams) |
| } |
| |
| // --------------------------------------------------------------------------- |
| // Topics |
| // --------------------------------------------------------------------------- |
| |
| impl TryFrom<TopicHeader> for Topic { |
| type Error = IggyError; |
| |
| fn try_from(w: TopicHeader) -> Result<Self, Self::Error> { |
| let message_expiry = match w.message_expiry { |
| 0 => IggyExpiry::NeverExpire, |
| v => v.into(), |
| }; |
| let max_topic_size: MaxTopicSize = w.max_topic_size.into(); |
| let options = resource_options_from_wire_split(&w.options, &w.derived_options)?; |
| Ok(Self { |
| id: w.id, |
| created_at: w.created_at.into(), |
| name: w.name.to_string(), |
| partitions_count: w.partitions_count, |
| size: IggyByteSize::from(w.size_bytes), |
| messages_count: w.messages_count, |
| message_expiry, |
| compression_algorithm: CompressionAlgorithm::from_code(w.compression_algorithm)?, |
| max_topic_size, |
| options, |
| }) |
| } |
| } |
| |
| impl From<PartitionResponse> for Partition { |
| fn from(w: PartitionResponse) -> Self { |
| Self { |
| id: w.id, |
| created_at: w.created_at.into(), |
| segments_count: w.segments_count, |
| current_offset: w.current_offset, |
| size: IggyByteSize::from(w.size_bytes), |
| messages_count: w.messages_count, |
| } |
| } |
| } |
| |
| impl TryFrom<GetTopicResponse> for TopicDetails { |
| type Error = IggyError; |
| |
| fn try_from(w: GetTopicResponse) -> Result<Self, Self::Error> { |
| let topic = Topic::try_from(w.topic)?; |
| let mut partitions: Vec<Partition> = |
| w.partitions.into_iter().map(Partition::from).collect(); |
| partitions.sort_by_key(|p| p.id); |
| Ok(Self { |
| id: topic.id, |
| created_at: topic.created_at, |
| name: topic.name, |
| size: topic.size, |
| messages_count: topic.messages_count, |
| message_expiry: topic.message_expiry, |
| compression_algorithm: topic.compression_algorithm, |
| max_topic_size: topic.max_topic_size, |
| partitions_count: topic.partitions_count, |
| partitions, |
| options: topic.options, |
| }) |
| } |
| } |
| |
| pub fn topics_from_wire(w: GetTopicsResponse) -> Result<Vec<Topic>, IggyError> { |
| let mut topics: Vec<Topic> = w |
| .topics |
| .into_iter() |
| .map(Topic::try_from) |
| .collect::<Result<_, _>>()?; |
| topics.sort_by_key(|t| t.id); |
| Ok(topics) |
| } |
| |
| // --------------------------------------------------------------------------- |
| // Users |
| // --------------------------------------------------------------------------- |
| |
| impl TryFrom<UserResponse> for UserInfo { |
| type Error = IggyError; |
| |
| fn try_from(w: UserResponse) -> Result<Self, Self::Error> { |
| Ok(Self { |
| id: w.id, |
| created_at: w.created_at.into(), |
| status: UserStatus::from_code(w.status)?, |
| username: w.username.to_string(), |
| options: resource_options_from_wire(&w.options, true)?, |
| }) |
| } |
| } |
| |
| impl TryFrom<UserDetailsResponse> for UserInfoDetails { |
| type Error = IggyError; |
| |
| fn try_from(w: UserDetailsResponse) -> Result<Self, Self::Error> { |
| let user = UserInfo::try_from(w.user)?; |
| let permissions = w.permissions.map(Permissions::from); |
| Ok(Self { |
| id: user.id, |
| created_at: user.created_at, |
| status: user.status, |
| username: user.username, |
| permissions, |
| options: user.options, |
| }) |
| } |
| } |
| |
| pub fn users_from_wire(w: GetUsersResponse) -> Result<Vec<UserInfo>, IggyError> { |
| let mut users: Vec<UserInfo> = w |
| .users |
| .into_iter() |
| .map(UserInfo::try_from) |
| .collect::<Result<_, _>>()?; |
| users.sort_by_key(|u| u.id); |
| Ok(users) |
| } |
| |
| impl From<IdentityResponse> for IdentityInfo { |
| fn from(w: IdentityResponse) -> Self { |
| Self { |
| user_id: w.user_id, |
| access_token: None, |
| } |
| } |
| } |
| |
| // --------------------------------------------------------------------------- |
| // Clients |
| // --------------------------------------------------------------------------- |
| |
| impl From<ConsumerGroupInfoResponse> for ConsumerGroupInfo { |
| fn from(w: ConsumerGroupInfoResponse) -> Self { |
| Self { |
| stream_id: w.stream_id, |
| topic_id: w.topic_id, |
| group_id: w.group_id, |
| } |
| } |
| } |
| |
| impl From<ClientResponse> for ClientInfo { |
| fn from(w: ClientResponse) -> Self { |
| let user_id = match w.user_id { |
| WIRE_NO_USER_ID => None, |
| id => Some(id), |
| }; |
| let transport = match w.transport { |
| 1 => "TCP", |
| 2 => "QUIC", |
| 3 => "HTTP", |
| 4 => "WebSocket", |
| _ => "Unknown", |
| } |
| .to_string(); |
| Self { |
| client_id: w.client_id, |
| user_id, |
| address: w.address, |
| transport, |
| consumer_groups_count: w.consumer_groups_count, |
| } |
| } |
| } |
| |
| impl From<ClientDetailsResponse> for ClientInfoDetails { |
| fn from(w: ClientDetailsResponse) -> Self { |
| let client = ClientInfo::from(w.client); |
| let mut consumer_groups: Vec<ConsumerGroupInfo> = w |
| .consumer_groups |
| .into_iter() |
| .map(ConsumerGroupInfo::from) |
| .collect(); |
| consumer_groups.sort_by_key(|cg| cg.group_id); |
| Self { |
| client_id: client.client_id, |
| user_id: client.user_id, |
| address: client.address, |
| transport: client.transport, |
| consumer_groups_count: client.consumer_groups_count, |
| consumer_groups, |
| } |
| } |
| } |
| |
| pub fn clients_from_wire(w: GetClientsResponse) -> Vec<ClientInfo> { |
| let mut clients: Vec<ClientInfo> = w.clients.into_iter().map(ClientInfo::from).collect(); |
| clients.sort_by_key(|c| c.client_id); |
| clients |
| } |
| |
| // --------------------------------------------------------------------------- |
| // Consumer Groups |
| // --------------------------------------------------------------------------- |
| |
| impl From<ConsumerGroupResponse> for ConsumerGroup { |
| fn from(w: ConsumerGroupResponse) -> Self { |
| Self { |
| id: w.id, |
| partitions_count: w.partitions_count, |
| members_count: w.members_count, |
| name: w.name.to_string(), |
| } |
| } |
| } |
| |
| impl From<ConsumerGroupMemberResponse> for ConsumerGroupMember { |
| fn from(w: ConsumerGroupMemberResponse) -> Self { |
| Self { |
| id: w.id, |
| partitions_count: w.partitions_count, |
| partitions: w.partitions, |
| } |
| } |
| } |
| |
| impl From<ConsumerGroupDetailsResponse> for ConsumerGroupDetails { |
| fn from(w: ConsumerGroupDetailsResponse) -> Self { |
| let group = ConsumerGroup::from(w.group); |
| let mut members: Vec<ConsumerGroupMember> = w |
| .members |
| .into_iter() |
| .map(ConsumerGroupMember::from) |
| .collect(); |
| members.sort_by_key(|m| m.id); |
| Self { |
| id: group.id, |
| name: group.name, |
| partitions_count: group.partitions_count, |
| members_count: group.members_count, |
| members, |
| } |
| } |
| } |
| |
| pub fn consumer_groups_from_wire(w: GetConsumerGroupsResponse) -> Vec<ConsumerGroup> { |
| let mut groups: Vec<ConsumerGroup> = w.groups.into_iter().map(ConsumerGroup::from).collect(); |
| groups.sort_by_key(|g| g.id); |
| groups |
| } |
| |
| // --------------------------------------------------------------------------- |
| // Consumer Offsets |
| // --------------------------------------------------------------------------- |
| |
| impl From<ConsumerOffsetResponse> for ConsumerOffsetInfo { |
| fn from(w: ConsumerOffsetResponse) -> Self { |
| Self { |
| partition_id: w.partition_id, |
| current_offset: w.current_offset, |
| stored_offset: w.stored_offset, |
| } |
| } |
| } |
| |
| // --------------------------------------------------------------------------- |
| // Personal Access Tokens |
| // --------------------------------------------------------------------------- |
| |
| impl From<PersonalAccessTokenResponse> for PersonalAccessTokenInfo { |
| fn from(w: PersonalAccessTokenResponse) -> Self { |
| let expiry_at = match w.expiry_at { |
| 0 => None, |
| v => Some(v.into()), |
| }; |
| Self { |
| name: w.name.to_string(), |
| expiry_at, |
| } |
| } |
| } |
| |
| pub fn personal_access_tokens_from_wire( |
| w: GetPersonalAccessTokensResponse, |
| ) -> Vec<PersonalAccessTokenInfo> { |
| let mut tokens: Vec<PersonalAccessTokenInfo> = w |
| .tokens |
| .into_iter() |
| .map(PersonalAccessTokenInfo::from) |
| .collect(); |
| tokens.sort_by(|a, b| a.name.cmp(&b.name)); |
| tokens |
| } |
| |
| impl From<RawPersonalAccessTokenResponse> for RawPersonalAccessToken { |
| fn from(w: RawPersonalAccessTokenResponse) -> Self { |
| Self { |
| token: w.token.to_string(), |
| } |
| } |
| } |
| |
| // --------------------------------------------------------------------------- |
| // System - Stats |
| // --------------------------------------------------------------------------- |
| |
| fn cache_metric_from_wire(w: CacheMetricEntry) -> (CacheMetricsKey, CacheMetrics) { |
| ( |
| CacheMetricsKey { |
| stream_id: w.stream_id, |
| topic_id: w.topic_id, |
| partition_id: w.partition_id, |
| }, |
| CacheMetrics { |
| hits: w.hits, |
| misses: w.misses, |
| hit_ratio: w.hit_ratio, |
| }, |
| ) |
| } |
| |
| impl From<StatsResponse> for Stats { |
| fn from(w: StatsResponse) -> Self { |
| let cache_metrics: HashMap<CacheMetricsKey, CacheMetrics> = w |
| .cache_metrics |
| .into_iter() |
| .map(cache_metric_from_wire) |
| .collect(); |
| Self { |
| process_id: w.process_id, |
| cpu_usage: w.cpu_usage, |
| total_cpu_usage: w.total_cpu_usage, |
| memory_usage: IggyByteSize::from(w.memory_usage), |
| total_memory: IggyByteSize::from(w.total_memory), |
| available_memory: IggyByteSize::from(w.available_memory), |
| run_time: w.run_time.into(), |
| start_time: w.start_time.into(), |
| read_bytes: IggyByteSize::from(w.read_bytes), |
| written_bytes: IggyByteSize::from(w.written_bytes), |
| messages_size_bytes: IggyByteSize::from(w.messages_size_bytes), |
| streams_count: w.streams_count, |
| topics_count: w.topics_count, |
| partitions_count: w.partitions_count, |
| segments_count: w.segments_count, |
| messages_count: w.messages_count, |
| clients_count: w.clients_count, |
| consumer_groups_count: w.consumer_groups_count, |
| hostname: w.hostname, |
| os_name: w.os_name, |
| os_version: w.os_version, |
| kernel_version: w.kernel_version, |
| iggy_server_version: w.iggy_server_version, |
| iggy_server_semver: w.iggy_server_semver, |
| cache_metrics, |
| threads_count: w.threads_count, |
| free_disk_space: IggyByteSize::from(w.free_disk_space), |
| total_disk_space: IggyByteSize::from(w.total_disk_space), |
| } |
| } |
| } |
| |
| // --------------------------------------------------------------------------- |
| // System - Cluster Metadata |
| // --------------------------------------------------------------------------- |
| |
| impl TryFrom<ClusterNodeResponse> for ClusterNode { |
| type Error = IggyError; |
| |
| fn try_from(w: ClusterNodeResponse) -> Result<Self, Self::Error> { |
| Ok(Self { |
| name: w.name, |
| ip: w.ip, |
| endpoints: TransportEndpoints::new( |
| w.tcp_port, |
| w.quic_port, |
| w.http_port, |
| w.websocket_port, |
| ), |
| role: ClusterNodeRole::try_from(w.role)?, |
| status: ClusterNodeStatus::try_from(w.status)?, |
| }) |
| } |
| } |
| |
| impl TryFrom<ClusterMetadataResponse> for ClusterMetadata { |
| type Error = IggyError; |
| |
| fn try_from(w: ClusterMetadataResponse) -> Result<Self, Self::Error> { |
| let nodes: Vec<ClusterNode> = w |
| .nodes |
| .into_iter() |
| .map(ClusterNode::try_from) |
| .collect::<Result<_, _>>()?; |
| Ok(Self { |
| name: w.name, |
| nodes, |
| }) |
| } |
| } |
| |
| /// Helper to convert a `ClusterMetadataResponse` into domain `ClusterMetadata`. |
| /// Errors if node role or status codes are unrecognized. |
| pub fn cluster_metadata_from_wire( |
| w: ClusterMetadataResponse, |
| ) -> Result<ClusterMetadata, IggyError> { |
| ClusterMetadata::try_from(w) |
| } |
| |
| // --------------------------------------------------------------------------- |
| // Identifier (domain -> wire) |
| // --------------------------------------------------------------------------- |
| |
| /// Convert a domain `Identifier` to `WireIdentifier`. |
| pub fn identifier_to_wire( |
| id: &crate::Identifier, |
| ) -> Result<iggy_binary_protocol::WireIdentifier, IggyError> { |
| match id.kind { |
| IdKind::Numeric => { |
| let value = id.get_u32_value()?; |
| Ok(iggy_binary_protocol::WireIdentifier::numeric(value)) |
| } |
| IdKind::String => { |
| let name = id.get_string_value()?; |
| iggy_binary_protocol::WireIdentifier::named(name) |
| .map_err(|_| IggyError::InvalidIdentifier) |
| } |
| } |
| } |
| |
| /// Convert a domain `Consumer` to `WireConsumer`. |
| pub fn consumer_to_wire(consumer: &Consumer) -> Result<WireConsumer, IggyError> { |
| let wire_id = identifier_to_wire(&consumer.id)?; |
| Ok(WireConsumer { |
| kind: consumer.kind.as_code(), |
| id: wire_id, |
| }) |
| } |
| |
| /// Convert a domain `PollingStrategy` to `WirePollingStrategy`. |
| pub fn polling_strategy_to_wire( |
| strategy: &crate::PollingStrategy, |
| ) -> iggy_binary_protocol::primitives::polling_strategy::WirePollingStrategy { |
| iggy_binary_protocol::primitives::polling_strategy::WirePollingStrategy { |
| kind: strategy.kind.as_code(), |
| value: strategy.value, |
| } |
| } |
| |
| /// Convert a domain `Partitioning` to `WirePartitioning`. |
| pub fn partitioning_to_wire( |
| partitioning: &crate::Partitioning, |
| ) -> Result<iggy_binary_protocol::primitives::partitioning::WirePartitioning, IggyError> { |
| use iggy_binary_protocol::primitives::partitioning::WirePartitioning; |
| match partitioning.kind { |
| crate::PartitioningKind::Balanced => Ok(WirePartitioning::Balanced), |
| crate::PartitioningKind::PartitionId => { |
| let bytes: [u8; 4] = partitioning |
| .value |
| .get(..4) |
| .and_then(|s| s.try_into().ok()) |
| .ok_or(IggyError::InvalidCommand)?; |
| Ok(WirePartitioning::PartitionId(u32::from_le_bytes(bytes))) |
| } |
| crate::PartitioningKind::MessagesKey => { |
| Ok(WirePartitioning::MessagesKey(partitioning.value.clone())) |
| } |
| } |
| } |
| |
| // --------------------------------------------------------------------------- |
| // Permissions (wire -> domain) |
| // --------------------------------------------------------------------------- |
| |
| impl From<WireGlobalPermissions> for GlobalPermissions { |
| fn from(w: WireGlobalPermissions) -> Self { |
| Self { |
| manage_servers: w.manage_servers, |
| read_servers: w.read_servers, |
| manage_users: w.manage_users, |
| read_users: w.read_users, |
| manage_streams: w.manage_streams, |
| read_streams: w.read_streams, |
| manage_topics: w.manage_topics, |
| read_topics: w.read_topics, |
| poll_messages: w.poll_messages, |
| send_messages: w.send_messages, |
| } |
| } |
| } |
| |
| fn wire_topic_permissions_to_domain(w: WireTopicPermissions) -> (usize, TopicPermissions) { |
| ( |
| w.topic_id as usize, |
| TopicPermissions { |
| manage_topic: w.manage_topic, |
| read_topic: w.read_topic, |
| poll_messages: w.poll_messages, |
| send_messages: w.send_messages, |
| }, |
| ) |
| } |
| |
| fn wire_stream_permissions_to_domain(w: WireStreamPermissions) -> (usize, StreamPermissions) { |
| let topics: Option<BTreeMap<usize, TopicPermissions>> = if w.topics.is_empty() { |
| None |
| } else { |
| Some( |
| w.topics |
| .into_iter() |
| .map(wire_topic_permissions_to_domain) |
| .collect(), |
| ) |
| }; |
| ( |
| w.stream_id as usize, |
| StreamPermissions { |
| manage_stream: w.manage_stream, |
| read_stream: w.read_stream, |
| manage_topics: w.manage_topics, |
| read_topics: w.read_topics, |
| poll_messages: w.poll_messages, |
| send_messages: w.send_messages, |
| topics, |
| }, |
| ) |
| } |
| |
| impl From<WirePermissions> for Permissions { |
| fn from(w: WirePermissions) -> Self { |
| let streams: Option<BTreeMap<usize, StreamPermissions>> = if w.streams.is_empty() { |
| None |
| } else { |
| Some( |
| w.streams |
| .into_iter() |
| .map(wire_stream_permissions_to_domain) |
| .collect(), |
| ) |
| }; |
| Self { |
| global: GlobalPermissions::from(w.global), |
| streams, |
| } |
| } |
| } |
| |
| /// Convert `&WirePermissions` to domain `Permissions` without consuming the input. |
| pub fn wire_permissions_to_permissions(wp: &WirePermissions) -> Permissions { |
| Permissions::from(wp.clone()) |
| } |
| |
| // --------------------------------------------------------------------------- |
| // Permissions (domain -> wire) |
| // --------------------------------------------------------------------------- |
| |
| /// Convert domain `Permissions` to `WirePermissions`. |
| pub fn permissions_to_wire(perms: &Permissions) -> WirePermissions { |
| let streams: Vec<WireStreamPermissions> = perms |
| .streams |
| .as_ref() |
| .map(|map| { |
| map.iter() |
| .map(|(&sid, sp)| stream_permissions_to_wire(sid, sp)) |
| .collect() |
| }) |
| .unwrap_or_default(); |
| WirePermissions { |
| global: WireGlobalPermissions { |
| manage_servers: perms.global.manage_servers, |
| read_servers: perms.global.read_servers, |
| manage_users: perms.global.manage_users, |
| read_users: perms.global.read_users, |
| manage_streams: perms.global.manage_streams, |
| read_streams: perms.global.read_streams, |
| manage_topics: perms.global.manage_topics, |
| read_topics: perms.global.read_topics, |
| poll_messages: perms.global.poll_messages, |
| send_messages: perms.global.send_messages, |
| }, |
| streams, |
| } |
| } |
| |
| fn stream_permissions_to_wire(stream_id: usize, sp: &StreamPermissions) -> WireStreamPermissions { |
| let topics: Vec<WireTopicPermissions> = sp |
| .topics |
| .as_ref() |
| .map(|map| { |
| map.iter() |
| .map(|(&tid, tp)| topic_permissions_to_wire(tid, tp)) |
| .collect() |
| }) |
| .unwrap_or_default(); |
| WireStreamPermissions { |
| stream_id: stream_id as u32, |
| manage_stream: sp.manage_stream, |
| read_stream: sp.read_stream, |
| manage_topics: sp.manage_topics, |
| read_topics: sp.read_topics, |
| poll_messages: sp.poll_messages, |
| send_messages: sp.send_messages, |
| topics, |
| } |
| } |
| |
| fn topic_permissions_to_wire(topic_id: usize, tp: &TopicPermissions) -> WireTopicPermissions { |
| WireTopicPermissions { |
| topic_id: topic_id as u32, |
| manage_topic: tp.manage_topic, |
| read_topic: tp.read_topic, |
| poll_messages: tp.poll_messages, |
| send_messages: tp.send_messages, |
| } |
| } |
| |
| // -- User Headers conversions -- |
| |
| /// Encode domain user headers into a [`WireUserHeaders`](iggy_binary_protocol::WireUserHeaders) wrapper. |
| pub fn user_headers_to_wire( |
| headers: &BTreeMap<HeaderKey, HeaderValue>, |
| ) -> iggy_binary_protocol::WireUserHeaders { |
| use bytes::{BufMut, BytesMut}; |
| use iggy_binary_protocol::WireUserHeaders; |
| |
| if headers.is_empty() { |
| return WireUserHeaders::empty(); |
| } |
| let size: usize = headers |
| .iter() |
| .map(|(k, v)| 1 + 4 + k.as_bytes().len() + 1 + 4 + v.as_bytes().len()) |
| .sum(); |
| let mut buf = BytesMut::with_capacity(size); |
| for (key, value) in headers { |
| buf.put_u8(key.kind().as_code()); |
| #[allow(clippy::cast_possible_truncation)] |
| buf.put_u32_le(key.as_bytes().len() as u32); |
| buf.put_slice(key.as_bytes()); |
| buf.put_u8(value.kind().as_code()); |
| #[allow(clippy::cast_possible_truncation)] |
| buf.put_u32_le(value.as_bytes().len() as u32); |
| buf.put_slice(value.as_bytes()); |
| } |
| // Buffer was just encoded from valid HeaderKey/HeaderValue entries, |
| // so structural TLV validity is guaranteed by construction. |
| WireUserHeaders::from_validated(buf.freeze()) |
| } |
| |
| /// Decode a [`WireUserHeaders`](iggy_binary_protocol::WireUserHeaders) wrapper into domain user headers. |
| /// |
| /// Wire-level validation accepts unknown kind codes for forward compatibility |
| /// (VSR rolling upgrades). Domain-level `from_code()` rejects them - the wire |
| /// layer stores structurally valid data, the domain layer requires known semantics. |
| pub fn user_headers_from_wire( |
| wire: &iggy_binary_protocol::WireUserHeaders, |
| ) -> Result<BTreeMap<HeaderKey, HeaderValue>, IggyError> { |
| user_headers_from_validated_slice(wire.as_bytes()) |
| } |
| |
| /// Decode user headers from a slice that has already passed structural TLV validation. |
| /// |
| /// Callers holding only a borrowed buffer can use this to skip the copy that |
| /// [`iggy_binary_protocol::WireUserHeaders::from_slice`] performs. |
| /// |
| /// # Panics |
| /// |
| /// Panics or yields garbage if `buf` has not been validated by |
| /// [`iggy_binary_protocol::validate_user_headers`]. The underlying iterator |
| /// slices each TLV field without bounds checks, relying on that validation. |
| pub(crate) fn user_headers_from_validated_slice( |
| buf: &[u8], |
| ) -> Result<BTreeMap<HeaderKey, HeaderValue>, IggyError> { |
| headers_from_validated_slice(buf, UnknownKinds::Reject) |
| } |
| |
| /// What a decode does with an entry whose kind code has no domain meaning. |
| #[derive(Debug, Clone, Copy, PartialEq, Eq)] |
| enum UnknownKinds { |
| /// Fail the whole decode. A message header the reader cannot interpret is |
| /// the caller's problem, not something to hide from it. |
| Reject, |
| /// Drop the entry and keep the rest. |
| Skip, |
| } |
| |
| fn headers_from_validated_slice( |
| buf: &[u8], |
| unknown: UnknownKinds, |
| ) -> Result<BTreeMap<HeaderKey, HeaderValue>, IggyError> { |
| if buf.is_empty() { |
| return Ok(BTreeMap::new()); |
| } |
| let mut headers = BTreeMap::new(); |
| for entry in WireUserHeaderIterator::new(buf) { |
| let (key_kind, value_kind) = match ( |
| HeaderKind::from_code(entry.key_kind.0), |
| HeaderKind::from_code(entry.value_kind.0), |
| ) { |
| (Ok(key_kind), Ok(value_kind)) => (key_kind, value_kind), |
| (Err(error), _) | (Ok(_), Err(error)) => { |
| if unknown == UnknownKinds::Skip { |
| continue; |
| } |
| return Err(error); |
| } |
| }; |
| if let Some(expected) = key_kind.expected_size() |
| && entry.key.len() != expected |
| { |
| return Err(IggyError::InvalidHeaderKey); |
| } |
| if let Some(expected) = value_kind.expected_size() |
| && entry.value.len() != expected |
| { |
| return Err(IggyError::InvalidHeaderValue); |
| } |
| |
| headers.insert( |
| HeaderKey::new_unchecked(key_kind, entry.key), |
| HeaderValue::new_unchecked(value_kind, entry.value), |
| ); |
| } |
| Ok(headers) |
| } |
| |
| // -- Options conversions -- |
| |
| /// Decode a `DescribeOptions` response into domain option specs. |
| /// |
| /// # Errors |
| /// |
| /// Returns `IggyError::InvalidHeaderKind` when an entry carries an unknown |
| /// canonical kind code. |
| pub fn option_specs_from_wire( |
| wire: iggy_binary_protocol::responses::system::DescribeOptionsResponse, |
| ) -> Result<Vec<OptionSpec>, IggyError> { |
| wire.entries |
| .into_iter() |
| .map(|entry| { |
| Ok(OptionSpec { |
| key: entry.key.to_string(), |
| kind: HeaderKind::from_code(entry.kind)?, |
| default_value: entry.default_value.to_vec(), |
| description: entry.description, |
| }) |
| }) |
| .collect() |
| } |
| |
| /// Decode a [`WireOptions`] block into domain resource options. |
| /// |
| /// Every decoded entry is marked with the given `explicit` flag; the wire |
| /// block carries no per-entry provenance. Admission calls this once for the |
| /// client block (`explicit == true`) and fills defaults separately. |
| /// |
| /// Entries whose kind code has no domain meaning are skipped, not rejected. |
| /// The wire layer forwards unknown kinds verbatim so a mixed-version cluster |
| /// can round-trip them, and this decode sits on the apply path, on the SDK's |
| /// list-decode and on the server's own reads: refusing the entry would mean an |
| /// old replica rejecting a commit a new one accepted, one unreadable entry |
| /// failing a whole `get_topics()`, and a topic that cannot be read back. |
| /// |
| /// # Errors |
| /// |
| /// Returns `IggyError::InvalidHeaderKey` / `InvalidHeaderValue` when a known |
| /// fixed-size kind carries a mismatched payload. |
| pub fn resource_options_from_wire( |
| wire: &iggy_binary_protocol::WireOptions, |
| explicit: bool, |
| ) -> Result<ResourceOptions, IggyError> { |
| let headers = headers_from_validated_slice(wire.as_bytes(), UnknownKinds::Skip)?; |
| Ok(headers |
| .into_iter() |
| .map(|(key, value)| (key, OptionValue { value, explicit })) |
| .collect()) |
| } |
| |
| /// Merge response-side `(explicit, derived)` wire blocks back into domain |
| /// resource options, restoring per-key provenance. Explicit wins on a key |
| /// collision (which a well-formed response never produces). |
| /// |
| /// # Errors |
| /// |
| /// Same contract as [`resource_options_from_wire`]. |
| pub fn resource_options_from_wire_split( |
| explicit: &iggy_binary_protocol::WireOptions, |
| derived: &iggy_binary_protocol::WireOptions, |
| ) -> Result<ResourceOptions, IggyError> { |
| let mut options = resource_options_from_wire(derived, false)?; |
| options.extend(resource_options_from_wire(explicit, true)?); |
| Ok(options) |
| } |
| |
| /// Split domain resource options into `(explicit, derived)` wire blocks, |
| /// the response-side layout that preserves per-key provenance. |
| /// |
| /// # Errors |
| /// |
| /// Same contract as [`resource_options_to_wire`]. |
| pub fn resource_options_to_wire_split( |
| options: &ResourceOptions, |
| ) -> Result<(WireOptions, WireOptions), IggyError> { |
| Ok(( |
| resource_options_to_wire(options, OptionsProvenance::Explicit)?, |
| resource_options_to_wire(options, OptionsProvenance::Derived)?, |
| )) |
| } |
| |
| /// Fixed per-entry cost of the TLV encoding: a kind byte and a `u32` length |
| /// for the key and for the value. |
| const OPTION_ENTRY_OVERHEAD: usize = 2 * (1 + 4); |
| |
| /// Encode domain resource options into a [`WireOptions`] block, keeping only |
| /// the entries matching `provenance`. Encoding just the explicit ones is what |
| /// lets a client round-trip a create without pinning server defaults. |
| /// |
| /// # Errors |
| /// |
| /// Returns `IggyError::OptionsBlockTooLarge` when the selected entries exceed |
| /// `MAX_OPTIONS` or `MAX_OPTIONS_BYTES`. [`WireOptions::from_validated`] skips |
| /// revalidation, so the two caps have to hold here: a block written past them |
| /// is one the receiving peer's `decode_options_prefixed` refuses, which would |
| /// make the resource permanently unreadable rather than merely oversized. |
| pub fn resource_options_to_wire( |
| options: &ResourceOptions, |
| provenance: OptionsProvenance, |
| ) -> Result<WireOptions, IggyError> { |
| let entries: Vec<(&HeaderKey, &OptionValue)> = options |
| .iter() |
| .filter(|(_, option)| option.matches(provenance)) |
| .collect(); |
| if entries.is_empty() { |
| return Ok(WireOptions::empty()); |
| } |
| if entries.len() > MAX_OPTIONS as usize { |
| return Err(IggyError::OptionsBlockTooLarge(format!( |
| "{} entries, maximum {MAX_OPTIONS}", |
| entries.len() |
| ))); |
| } |
| let size: usize = entries |
| .iter() |
| .map(|(key, option)| { |
| OPTION_ENTRY_OVERHEAD + key.as_bytes().len() + option.value.as_bytes().len() |
| }) |
| .sum(); |
| if size > MAX_OPTIONS_BYTES { |
| return Err(IggyError::OptionsBlockTooLarge(format!( |
| "{size} bytes, maximum {MAX_OPTIONS_BYTES}" |
| ))); |
| } |
| let mut buf = BytesMut::with_capacity(size); |
| for (key, option) in entries { |
| // A non-string key would encode a block the receiving peer refuses, |
| // making the resource permanently unreadable. It holds because every |
| // producer went through `validate_options`, not because `HeaderKey` |
| // guarantees it, so the invariant is asserted where it is relied on. |
| debug_assert_eq!( |
| key.kind(), |
| HeaderKind::String, |
| "option key must be a string" |
| ); |
| buf.put_u8(key.kind().as_code()); |
| #[allow(clippy::cast_possible_truncation)] |
| buf.put_u32_le(key.as_bytes().len() as u32); |
| buf.put_slice(key.as_bytes()); |
| buf.put_u8(option.value.kind().as_code()); |
| #[allow(clippy::cast_possible_truncation)] |
| buf.put_u32_le(option.value.as_bytes().len() as u32); |
| buf.put_slice(option.value.as_bytes()); |
| } |
| // Structural validity holds by construction: entries come from valid |
| // `HeaderKey`s walked in `BTreeMap` order, so keys are sorted and unique, |
| // and the loop above asserts they are string-kinded. The caps cover what |
| // construction cannot. |
| Ok(WireOptions::from_validated(buf.freeze())) |
| } |
| |
| #[cfg(test)] |
| mod tests { |
| use super::*; |
| use iggy_binary_protocol::WireUserHeaders; |
| use std::str::FromStr; |
| |
| fn sample_headers() -> BTreeMap<HeaderKey, HeaderValue> { |
| BTreeMap::from([ |
| ( |
| HeaderKey::from_str("content-type").unwrap(), |
| HeaderValue::from_str("text/plain").unwrap(), |
| ), |
| (HeaderKey::from_str("retries").unwrap(), 7u32.into()), |
| ]) |
| } |
| |
| #[test] |
| fn given_encoded_headers_when_decoded_from_slice_should_recover_the_originals() { |
| let wire = user_headers_to_wire(&sample_headers()); |
| |
| // The originals are the oracle here. Comparing against `user_headers_from_wire` |
| // would be vacuous, since it delegates to the function under test. |
| let decoded = user_headers_from_validated_slice(wire.as_bytes()).unwrap(); |
| |
| assert_eq!(decoded, sample_headers()); |
| for (key, value) in &decoded { |
| let (expected_key, expected_value) = sample_headers() |
| .into_iter() |
| .find(|(k, _)| k == key) |
| .unwrap(); |
| assert_eq!(key.kind(), expected_key.kind()); |
| assert_eq!(value.kind(), expected_value.kind()); |
| assert_eq!(value.as_bytes(), expected_value.as_bytes()); |
| } |
| } |
| |
| #[test] |
| fn given_headers_copied_into_owned_wire_when_decoded_should_match_borrowed_decode() { |
| // Guards the assumption that decoding a borrowed slice is equivalent to |
| // decoding an independently copied buffer. |
| let wire = user_headers_to_wire(&sample_headers()); |
| let copied = WireUserHeaders::from_slice(wire.as_bytes()).unwrap(); |
| |
| assert_eq!( |
| user_headers_from_validated_slice(wire.as_bytes()).unwrap(), |
| user_headers_from_validated_slice(copied.as_bytes()).unwrap() |
| ); |
| } |
| |
| #[test] |
| fn given_empty_buffer_when_decoded_from_slice_should_return_empty_map() { |
| assert!(user_headers_from_validated_slice(&[]).unwrap().is_empty()); |
| } |
| |
| #[test] |
| fn given_unknown_kind_code_when_decoded_from_slice_should_return_error() { |
| // Structurally valid TLV whose key kind code has no domain meaning. |
| let mut buf = Vec::new(); |
| buf.push(0xFF); |
| buf.extend_from_slice(&3u32.to_le_bytes()); |
| buf.extend_from_slice(b"key"); |
| buf.push(0xFF); |
| buf.extend_from_slice(&3u32.to_le_bytes()); |
| buf.extend_from_slice(b"val"); |
| |
| assert!(iggy_binary_protocol::validate_user_headers(&buf).is_ok()); |
| assert!(user_headers_from_validated_slice(&buf).is_err()); |
| assert!(user_headers_from_wire(&WireUserHeaders::from_slice(&buf).unwrap()).is_err()); |
| } |
| |
| fn put_option_entry(buf: &mut Vec<u8>, key: &str, value_kind: u8, value: &[u8]) { |
| buf.push(HeaderKind::String.as_code()); |
| buf.extend_from_slice(&u32::try_from(key.len()).unwrap().to_le_bytes()); |
| buf.extend_from_slice(key.as_bytes()); |
| buf.push(value_kind); |
| buf.extend_from_slice(&u32::try_from(value.len()).unwrap().to_le_bytes()); |
| buf.extend_from_slice(value); |
| } |
| |
| #[test] |
| fn given_option_with_unknown_value_kind_when_decoded_should_skip_only_that_entry() { |
| // A newer peer's option value kind. The wire layer forwards it, so the |
| // domain decode has to drop the entry rather than fail the block: this |
| // decode runs on the apply path, where an error means one replica |
| // rejecting a commit another accepted. |
| let mut buf = Vec::new(); |
| put_option_entry(&mut buf, "from_the_future", 200, b"opaque"); |
| put_option_entry( |
| &mut buf, |
| "segment_size", |
| HeaderKind::String.as_code(), |
| b"1MB", |
| ); |
| let wire = WireOptions::from_slice(&buf).expect("unknown value kinds stay wire-valid"); |
| |
| let options = resource_options_from_wire(&wire, true).unwrap(); |
| |
| assert_eq!(options.len(), 1); |
| assert!(options.contains_key(&HeaderKey::from_str("segment_size").unwrap())); |
| } |
| } |
