Skip to content

Commit 03d4165

Browse files
KeruspeKeruspe
committed
experiments
Signed-off-by: Marc-Antoine Perennou <Marc-Antoine@Perennou.com>
1 parent 1131078 commit 03d4165

5 files changed

Lines changed: 330 additions & 0 deletions

File tree

‎Cargo.lock‎

Lines changed: 1 addition & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎Cargo.toml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,7 @@ waker-fn = "^1.1"
7373

7474
[dev-dependencies]
7575
async-global-executor = "^3.1"
76+
async-io = "^2.0"
7677
futures-lite = "^2.0"
7778
serde_json = "^1.0"
7879
waker-fn = "^1.1"

‎examples/c.rs‎

Lines changed: 91 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,91 @@
1+
use futures_lite::StreamExt;
2+
use lapin::{Connection, ConnectionProperties, options::*, types::FieldTable};
3+
use tracing::info;
4+
5+
fn main() {
6+
if std::env::var("RUST_LOG").is_err() {
7+
unsafe { std::env::set_var("RUST_LOG", "info") };
8+
}
9+
10+
tracing_subscriber::fmt::init();
11+
12+
let addr = std::env::var("AMQP_ADDR").unwrap_or_else(|_| "amqp://127.0.0.1:5672/%2f".into());
13+
let recovery_config = lapin::RecoveryConfig::default().auto_recover_channels();
14+
15+
async_global_executor::block_on(async {
16+
let conn = Connection::connect(
17+
&addr,
18+
ConnectionProperties::default().with_experimental_recovery_config(recovery_config),
19+
)
20+
.await
21+
.expect("connection error");
22+
23+
info!("CONNECTED");
24+
25+
//receive channel
26+
let channel = conn.create_channel().await.expect("create_channel");
27+
info!(state=?conn.status().state());
28+
29+
let queue = channel
30+
.queue_declare(
31+
"hello-recover",
32+
QueueDeclareOptions::default(),
33+
FieldTable::default(),
34+
)
35+
.await
36+
.expect("queue_declare");
37+
info!(state=?conn.status().state());
38+
info!(?queue, "Declared queue");
39+
40+
let ch = channel.clone();
41+
async_global_executor::spawn(async move {
42+
loop {
43+
async_io::Timer::after(std::time::Duration::from_secs(1)).await;
44+
info!("Trigger failure");
45+
assert!(
46+
ch.queue_declare(
47+
"fake queue",
48+
QueueDeclareOptions {
49+
passive: true,
50+
..QueueDeclareOptions::default()
51+
},
52+
FieldTable::default(),
53+
)
54+
.await
55+
.is_err()
56+
);
57+
}
58+
})
59+
.detach();
60+
61+
info!("will consume");
62+
let mut consumer = channel
63+
.basic_consume(
64+
"hello-recover",
65+
"my_consumer",
66+
BasicConsumeOptions::default(),
67+
FieldTable::default(),
68+
)
69+
.await
70+
.expect("basic_consume");
71+
info!(state=?conn.status().state());
72+
73+
let mut count = 0;
74+
while let Some(delivery) = consumer.next().await {
75+
info!(message=?delivery, "received message");
76+
if let Ok(delivery) = delivery {
77+
let data = str::from_utf8(&delivery.data).expect("invalid utf8 data");
78+
println!("{}", data);
79+
delivery
80+
.ack(BasicAckOptions::default())
81+
.await
82+
.expect("basic_ack");
83+
count += 1;
84+
if data == "STOP" {
85+
println!("Received {} msgs", count);
86+
break;
87+
}
88+
}
89+
}
90+
})
91+
}

‎examples/p.rs‎

Lines changed: 115 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,115 @@
1+
use lapin::{BasicProperties, Connection, ConnectionProperties, options::*, types::FieldTable};
2+
use std::sync::{
3+
Arc,
4+
atomic::{AtomicUsize, Ordering},
5+
};
6+
use tracing::info;
7+
8+
fn main() {
9+
if std::env::var("RUST_LOG").is_err() {
10+
unsafe {
11+
std::env::set_var("RUST_LOG", "info");
12+
}
13+
}
14+
15+
tracing_subscriber::fmt::init();
16+
17+
let addr = std::env::var("AMQP_ADDR").unwrap_or_else(|_| "amqp://127.0.0.1:5672/%2f".into());
18+
let recovery_config = lapin::RecoveryConfig::default().auto_recover_channels();
19+
20+
async_global_executor::block_on(async {
21+
let conn = Connection::connect(
22+
&addr,
23+
ConnectionProperties::default().with_experimental_recovery_config(recovery_config),
24+
)
25+
.await
26+
.expect("connection error");
27+
28+
info!("CONNECTED");
29+
30+
let channel1 = conn.create_channel().await.expect("create_channel");
31+
channel1
32+
.confirm_select(ConfirmSelectOptions::default())
33+
.await
34+
.expect("confirm_select");
35+
channel1
36+
.queue_declare(
37+
"hello-recover",
38+
QueueDeclareOptions::default(),
39+
FieldTable::default(),
40+
)
41+
.await
42+
.expect("queue_declare");
43+
44+
let count = Arc::new(AtomicUsize::new(0));
45+
let counter = count.clone();
46+
47+
let ch = channel1.clone();
48+
async_global_executor::spawn(async move {
49+
loop {
50+
async_io::Timer::after(std::time::Duration::from_secs(1)).await;
51+
info!("Trigger failure");
52+
assert!(
53+
ch.queue_declare(
54+
"fake queue",
55+
QueueDeclareOptions {
56+
passive: true,
57+
..QueueDeclareOptions::default()
58+
},
59+
FieldTable::default(),
60+
)
61+
.await
62+
.is_err()
63+
);
64+
counter.fetch_add(1, Ordering::SeqCst);
65+
}
66+
})
67+
.detach();
68+
69+
let mut published = 0;
70+
let mut errors = 0;
71+
info!("will publish");
72+
loop {
73+
let res = channel1
74+
.basic_publish(
75+
"",
76+
"hello-recover",
77+
BasicPublishOptions::default(),
78+
b"before",
79+
BasicProperties::default(),
80+
)
81+
.await;
82+
let res = if let Ok(res) = res {
83+
res.await.map(|_| ())
84+
} else {
85+
res.map(|_| ())
86+
};
87+
match res {
88+
Ok(()) => {
89+
published += 1;
90+
}
91+
Err(err) => {
92+
errors += 1;
93+
if let Err(err) = channel1.wait_for_recovery(err).await {
94+
panic!("{}", err);
95+
}
96+
info!("notifier done");
97+
}
98+
}
99+
if count.load(Ordering::SeqCst) > 10 {
100+
println!("Published {} with {} errors", published, errors);
101+
channel1
102+
.basic_publish(
103+
"",
104+
"hello-recover",
105+
BasicPublishOptions::default(),
106+
b"STOP",
107+
BasicProperties::default(),
108+
)
109+
.await
110+
.unwrap();
111+
break;
112+
}
113+
}
114+
});
115+
}

‎examples/t.rs‎

Lines changed: 122 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,122 @@
1+
use lapin::{
2+
BasicProperties, Connection, ConnectionProperties, message::DeliveryResult, options::*,
3+
publisher_confirm::Confirmation, types::FieldTable,
4+
};
5+
use tracing::info;
6+
7+
fn main() {
8+
if std::env::var("RUST_LOG").is_err() {
9+
std::env::set_var("RUST_LOG", "info");
10+
}
11+
12+
tracing_subscriber::fmt::init();
13+
14+
let addr = std::env::var("AMQP_ADDR").unwrap_or_else(|_| "amqp://127.0.0.1:5672/%2f".into());
15+
let recovery_config = lapin::RecoveryConfig::default().auto_recover_channels();
16+
17+
async_global_executor::block_on(async {
18+
let conn = Connection::connect(
19+
&addr,
20+
ConnectionProperties::default().with_experimental_recovery_config(recovery_config),
21+
)
22+
.await
23+
.expect("connection error");
24+
25+
info!("CONNECTED");
26+
27+
{
28+
let channel1 = conn.create_channel().await.expect("create_channel");
29+
let channel2 = conn.create_channel().await.expect("create_channel");
30+
channel1
31+
.confirm_select(ConfirmSelectOptions::default())
32+
.await
33+
.expect("confirm_select");
34+
channel1
35+
.queue_declare(
36+
"recover-test",
37+
QueueDeclareOptions::default(),
38+
FieldTable::default(),
39+
)
40+
.await
41+
.expect("queue_declare");
42+
43+
info!("will consume");
44+
let channel = channel2.clone();
45+
channel2
46+
.basic_consume(
47+
"recover-test",
48+
"my_consumer",
49+
BasicConsumeOptions::default(),
50+
FieldTable::default(),
51+
)
52+
.await
53+
.expect("basic_consume")
54+
.set_delegate(move |delivery: DeliveryResult| {
55+
let channel = channel.clone();
56+
async move {
57+
info!(message=?delivery, "received message");
58+
if let Ok(Some(delivery)) = delivery {
59+
delivery
60+
.ack(BasicAckOptions::default())
61+
.await
62+
.expect("basic_ack");
63+
if &delivery.data[..] == b"after" {
64+
channel
65+
.basic_cancel("my_consumer", BasicCancelOptions::default())
66+
.await
67+
.expect("basic_cancel");
68+
}
69+
}
70+
}
71+
});
72+
73+
info!("will publish");
74+
let confirm = channel1
75+
.basic_publish(
76+
"",
77+
"recover-test",
78+
BasicPublishOptions::default(),
79+
b"before",
80+
BasicProperties::default(),
81+
)
82+
.await
83+
.expect("basic_publish")
84+
.await
85+
.expect("publisher-confirms");
86+
assert_eq!(confirm, Confirmation::Ack(None));
87+
88+
info!("before fail");
89+
assert!(
90+
channel1
91+
.queue_declare(
92+
"fake queue",
93+
QueueDeclareOptions {
94+
passive: true,
95+
..QueueDeclareOptions::default()
96+
},
97+
FieldTable::default(),
98+
)
99+
.await
100+
.is_err()
101+
);
102+
info!("after fail");
103+
104+
info!("publish after");
105+
let confirm = channel1
106+
.basic_publish(
107+
"",
108+
"recover-test",
109+
BasicPublishOptions::default(),
110+
b"after",
111+
BasicProperties::default(),
112+
)
113+
.await
114+
.expect("basic_publish")
115+
.await
116+
.expect("publisher-confirms");
117+
assert_eq!(confirm, Confirmation::Ack(None));
118+
}
119+
120+
conn.run().expect("conn.run");
121+
});
122+
}

0 commit comments

Comments
 (0)