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
//! # Queue Stream Re-queue
//!
//! Demonstrates re-queuing messages from one channel to another. Messages are
//! polled from a source channel and individually re-queued to a target channel
//! using `msg.re_queue()`. The target channel is then verified to contain the
//! moved messages.
//!
//! ## Expected Output
//!
//! ```text
//! Received 3 messages from source
//! Re-queued message <uuid> to queues.stream.requeue.target
//! Re-queued message <uuid> to queues.stream.requeue.target
//! Re-queued message <uuid> to queues.stream.requeue.target
//! Target channel has 3 messages
//! ```
//!
//! ## Running
//!
//! Requires a running KubeMQ broker. By default connects to `localhost:50000`.
//! Override with `KUBEMQ_ADDRESS`:
//!
//! ```bash
//! KUBEMQ_ADDRESS=my-host:50000 cargo run --example queues_stream_requeue
//! ```
use kubemq::prelude::*;
use kubemq::{PollRequest, QueueMessageBuilder};
#[tokio::main]
async fn main() -> kubemq::Result<()> {
let client = KubemqClient::builder()
.host("localhost")
.port(50000)
.build()
.await?;
let source_channel = "queues.stream.requeue.source";
let target_channel = "queues.stream.requeue.target";
// Send messages to source
for i in 0..3 {
let msg = QueueMessageBuilder::new()
.channel(source_channel)
.body(format!("requeue-msg-{}", i).into_bytes())
.build();
client.send_queue_message(msg).await?;
}
let mut receiver = client.new_queue_downstream_receiver().await?;
let poll = PollRequest {
channel: source_channel.to_string(),
max_items: 3,
wait_timeout_seconds: 5,
auto_ack: false,
};
let response = receiver.poll(poll).await?;
println!("Received {} messages from source", response.messages.len());
// Re-queue each message to the target channel
for msg in &response.messages {
msg.re_queue(target_channel).await?;
println!("Re-queued message {} to {}", msg.message.id, target_channel);
}
// Verify messages are in target channel
let target_msgs = client
.receive_queue_messages(target_channel, 10, 5, false)
.await?;
println!("Target channel has {} messages", target_msgs.len());
receiver.close().await?;
client.close().await?;
Ok(())
}