hubcio commented on code in PR #2613: URL: https://github.com/apache/iggy/pull/2613#discussion_r2733071715
########## examples/rust/src/message-headers/message-compression/README.md: ########## @@ -0,0 +1,269 @@ +# An Example on Message Compression in Transit + +This example illustrates how (with some minor additional implementation) the Iggy SDK can be used to compress and decompress messages in transit. + +## Running the Example + +Details on how to run the examples for the Rust Iggy SDK can be found in the parent folder [README.md](https://github.com/apache/iggy/tree/master/examples/rust#readme). + +Run the following commands + +1. Start the server + + ```bash + cargo run --bin iggy-server -- --with-default-root-credentials + ``` + + **NOTE**: In case the server was running before, make sure to run `rm -rf local_data/` to delete server state data from prior runs. + +2. Run the producer to write compressed messages to the server + + ```bash + cargo run --example message-headers-compression-producer + ``` + +3. Run the consumer to read and decompress messages from the server + + ```bash + cargo run --example message-headers-compression-consumer + ``` + +## The Codec + +The **co**mpression and **dec**compression utilities are implemented in `examples/rust/src/shared/codec.rs` and used when sending messages to the server and reading them from the server. + +First, define a stream and a topic name. +The producer will first initiate the stream and the topic on that stream and then write the example messages to that topic within that stream. +The consumer will use the names as identifier to read messages from that topic on that stream. + +```rust +pub const STREAM_NAME: &str = "compression-stream"; +pub const TOPIC_NAME: &str = "compression-topic"; +``` + +Additionally, set a constant that defines the number of messages to be send to the server via the producer. + +```rust +pub const NUM_MESSAGES: u32 = 1000; +``` + +### Spotlight: IggyMessage's + +In order to add functionality to compress and decompress messages during transit, we need to know what a message actually is and how it is implemented. +Iggy implements two important types, that we need to know. + +* [IggyMessage](https://github.com/apache/iggy/blob/e46f294b7af4f86b0d7e26d984205a019a8885f8/core/common/src/types/message/iggy_message.rs#L108) +* [ReceivedMessage](https://github.com/apache/iggy/blob/b26246252502ba6f5d6cad2895e7c468d9f959e4/core/sdk/src/clients/consumer.rs#L905) + +A message send to the server needs to be of type `IggyMessage`. + +```Rust +pub struct IggyMessage { + /// Message metadata + pub header: IggyMessageHeader, + /// Message content + pub payload: Bytes, + /// Optional user-defined headers + pub user_headers: Option<Bytes>, +} +``` + +The important bits in context of this example are the *payload* and the *user_headers*. +Payload is of type Bytes and corresponds to the actual message that we want to send to the server. + +Let's suppose our example is an abstraction over a real world scenario, where some application sends it's application logs to the iggy-server. This application is therefore the producer. +We also have a monitoring service, that inspects the logs of our application to check for any service disruptions. So this monitoring service needs to read the logs from the iggy-server and is therefore the consumer. + +Further suppose, the application logs are quite large and repetitive, since they follow a structured pattern (as logs usually do). +It may be a good idea to reduce bandwidth by trading of some idle CPU time to compress the logs before sending them to the server. +We go straight ahead, implement some compression functionalities and send the compressed messages to the server. +If the monitoring service now consumes these messages we have a problem. The logs are still compressed. +Even if we know that the messages are compressed we do not know how to decompress them because the algorithm that was used for compression is unknown. + +This is where `user_headers` become handy. +The definition above tells us, that user_headers are (optional) Bytes. But thats because finally everything is serialized before sending to the server. +Looking at the implementation of `IggyMessage` we see that user_headers are a serialized `HashMap` with Iggy specific types `HeaderKey` and `HeaderValue`. +So the user_headers are basically a set of metadata defined by us, the user, using a key and a value. + +Thus, for the compression scenario the user_headers can be used to signal to a consumer that a message was compressed before it was sent to the server. +The key to highlight message compression in this example is defined as: + +```rust +pub const COMPRESSION_HEADER_KEY: &str = "iggy-compression"; +``` + +By reading the user_headers and finding the "iggy-compression" key a consumer now knows, that the message was compressed. But it's still not transparent how it can be decompressed. +We can use the `HeaderValue` to store the information on how to decompress the message, e.g. using a **Codec**. + +---- + +So the HeaderKey is used to indicate that a message was compressed before transit and the HeaderValue indicates how it was compressed. +Using this idea we can implement a Codec that is shared between the consumer and producer. + +The Codec is an enum listing all the available algorithms to compress and decompress. + +```rust +pub enum Codec { + None, + Lz4, +} +``` + +Further, we implement three methods + +* `header_key`: Returns the HeaderKey that defines if a message was compressed or not. +The consumer uses it to look for the "iggy-compression" HeaderKey when inspecting the user_headers of a message. +* `to_header_value`: Generates a HeaderValue from the specific Codec instance. +* `from_header_value`: Resolves a HeaderValue into a Codec. +This is used in the consumer. After the "iggy-compression" HeaderKey was found in the user_headers we can obtain the HeaderValue, from which +we obtain the Codec type using this method and thereby gain access to the decompress method. + +```rust +impl Codec { + pub fn header_key() -> HeaderKey { + HeaderKey::new(COMPRESSION_HEADER_KEY) + .expect("COMPRESSION_HEADER_KEY is an InvalidHeaderKey.") + } + + pub fn to_header_value(&self) -> HeaderValue { + HeaderValue::from_str(&self.to_string()).expect("failed generating HeaderValue.") + } + + pub fn from_header_value(value: &HeaderValue) -> Self { + let name = value + .as_str() + .expect("could not convert HeaderValue into str."); + Self::from_str(name).expect("compression algorithm not available.") + } +``` + +The other two methods implement the compression and decompression logic, which is specifc to the actual Codec instance, dependent on the enum's variant. +The example Codec implements two. *None*, where data is not compressed and *Lz4* (using the lz4_flex crate). +Note, that this can be easily extended to more algorithms. + +```rust +impl Codec { + pub fn compress(&self, data: &[u8]) -> Result<Vec<u8>, IggyError> { + match self { + Codec::None => Ok(data.to_vec()), + Codec::Lz4 => { + let mut compressed_data = Vec::new(); + let mut encoder = FrameEncoder::new(&mut compressed_data); + encoder + .write_all(data) + .expect("Cannot write into buffer using Lz4 compression."); + encoder.finish().expect("Cannot finish Lz4 compression."); + Ok(compressed_data) + } + } + } + + pub fn decompress(&self, data: &[u8]) -> Result<Vec<u8>, IggyError> { Review Comment: this is incorrect, in code it actually returning Vec<u8> -- 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]
