bgustcrawler 0.2.0

Distributed web crawler and scheduler integrated with Kafka, SurrealDB, and S3 storage
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

use rdkafka::producer::{FutureProducer, FutureRecord};
use rdkafka::config::ClientConfig;
use anyhow::Result;
use std::time::Duration;

pub struct Producer {
    producer: FutureProducer,
}

impl Producer {
    pub fn new(brokers: &str) -> Result<Self> {
        let producer: FutureProducer = ClientConfig::new()
            .set("bootstrap.servers", brokers)
            .set("message.timeout.ms", "5000")
            .create()?;
        Ok(Self { producer })
    }

    pub async fn send_message(&self, topic: &str, key: &str, payload: &str) -> Result<()> {
        let record = FutureRecord::to(topic)
            .key(key)
            .payload(payload);
        
        self.producer.send(record, Duration::from_secs(5))
            .await
            .map_err(|(e, _)| anyhow::anyhow!("Kafka send failed: {:?}", e))?;
        Ok(())
    }
}