Subsection 1.1.1: Creating a Kafka Producer

To kick off any message-passing system, it is essential to create an application that sends messages to a message queue, commonly referred to as a producer. For this article, we will utilize the rust-rdkafka library to implement both producers and consumers.

The first step involves adding the necessary dependencies, rdkafka and tokio, to the Cargo.toml file. This can be done manually or by executing the following command:

cargo add rdkafka tokio

The tokio library is a crucial component as it enables asynchronous operations, which will be leveraged in our producer implementation.

use rdkafka::config::ClientConfig;

use rdkafka::message::{Header, OwnedHeaders};

use rdkafka::producer::{FutureProducer, FutureRecord};

use std::time::Duration;

pub struct Producer {

future_producer: FutureProducer,

topic: String,

message_id: usize

}

The Producer struct will require three attributes: a FutureProducer to handle asynchronous message sending, a topic to specify where to publish messages, and a message_id to uniquely identify each sent message.

Next, we will implement two functions: a constructor to instantiate the Producer and a method to send messages to the specified Kafka topic.

impl Producer {

pub fn new(bootstrap_server: &str, message_timeout: &str, topic: &str) -> Producer {

let producer: &FutureProducer = &ClientConfig::new()

.set("bootstrap.servers", bootstrap_server)

.set("message.timeout.ms", message_timeout)

.create()

.expect("Producer creation error");

Producer { future_producer: producer.clone(), topic: String::from(topic), message_id: 0 }

}

pub async fn produce(&mut self, message_bytes: &str) -> bool {

let delivery_status = self.future_producer.send(

FutureRecord::to(self.topic.as_str())

.payload(&format!("Message: {}", message_bytes))

.key(&format!("Key: {}", self.message_id))

.headers(OwnedHeaders::new().insert(Header {

key: "header_key",

value: Some("header_value"),

})),

Duration::from_secs(0),

).await;

self.message_id += 1;

match delivery_status.err() {

None => true,

Some(err) => {

print!("Error sending message {0} on topic {1}: {2}", message_bytes, self.topic, err.0);

false

}

}

}

}

Two distinct producers can be instantiated to publish messages on different topics, enhancing code clarity.

mod producer;

use std::future::Future;

use producer::producer::Producer;

#[tokio::main]

async fn main() {

let price_updates = vec![

"AAPL 169.58", "TSLA 164.90", "IBM 189.14", "PARA 11.97", "INTC 38.71",

// Additional updates...

];

let trade_updates = vec![

"BUY AAPL 100 170.00",

"SELL PARA 453 11.23",

// Additional trades...

];

let server: &str = "127.0.0.1:9092";

let timeout: &str = "5000";

let mut price_producer = Producer::new(server, timeout, "ticker.topic");

let mut trade_producer = Producer::new(server, timeout, "trade.topic");

for update in price_updates {

match price_producer.produce(update).await {

true => println!("Price update sent successfully."),

false => println!("Failed to send message"),

}

}

for trade in trade_updates {

match trade_producer.produce(trade).await {

true => println!("Trade update sent successfully."),

false => println!("Failed to send message"),

}

}

}

This asynchronous main function demonstrates the creation of two lists of messages—trade_updates and price_updates—and utilizes dedicated producers for each topic, efficiently sending messages and logging the outcomes to the console.