pulsar 6.7.2

Rust client for Apache Pulsar
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
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98

#![recursion_limit = "256"]
#[macro_use]
extern crate serde;
use futures::TryStreamExt;
use pulsar::{
    message::{proto, proto::command_subscribe::SubType, Payload},
    producer, Consumer, DeserializeMessage, Error as PulsarError, Pulsar, SerializeMessage,
    TokioExecutor,
};

#[derive(Serialize, Deserialize)]
struct TestData {
    data: String,
}

impl SerializeMessage for TestData {
    fn serialize_message(input: Self) -> Result<producer::Message, PulsarError> {
        let payload = serde_json::to_vec(&input).map_err(|e| PulsarError::Custom(e.to_string()))?;
        Ok(producer::Message {
            payload,
            ..Default::default()
        })
    }
}

impl DeserializeMessage for TestData {
    type Output = Result<TestData, serde_json::Error>;

    fn deserialize_message(payload: &Payload) -> Self::Output {
        serde_json::from_slice(&payload.data)
    }
}

#[tokio::main]
async fn main() -> Result<(), pulsar::Error> {
    env_logger::init();

    let addr = "pulsar://127.0.0.1:6650";
    let pulsar: Pulsar<_> = Pulsar::builder(addr, TokioExecutor).build().await?;
    let mut producer = pulsar
        .producer()
        .with_topic("test")
        .with_name("my-producer")
        .with_options(producer::ProducerOptions {
            schema: Some(proto::Schema {
                r#type: proto::schema::Type::String as i32,
                ..Default::default()
            }),
            ..Default::default()
        })
        .build()
        .await?;

    tokio::task::spawn(async move {
        let mut counter = 0usize;
        loop {
            producer
                .send_non_blocking(TestData {
                    data: "data".to_string(),
                })
                .await
                .unwrap()
                .await
                .unwrap();
            counter += 1;
            if counter % 1000 == 0 {
                println!("sent {counter} messages");
            }
        }
    });

    let pulsar2: Pulsar<_> = Pulsar::builder(addr, TokioExecutor).build().await?;

    let mut consumer: Consumer<TestData, _> = pulsar2
        .consumer()
        .with_topic("test")
        .with_consumer_name("test_consumer")
        .with_subscription_type(SubType::Exclusive)
        .with_subscription("test_subscription")
        .build()
        .await?;

    let mut counter = 0usize;
    while let Some(msg) = consumer.try_next().await? {
        log::info!("id: {:?}", msg.message_id());
        consumer.ack(&msg).await?;
        let data = msg.deserialize().unwrap();
        if data.data.as_str() != "data" {
            panic!("Unexpected payload: {}", &data.data);
        }
        counter += 1;
        if counter % 1000 == 0 {
            println!("received {counter} messages");
        }
    }

    Ok(())
}