Skip to content

Channel recovery #416

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Channel recovery #416

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 1 addition & 0 deletions Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,7 @@ waker-fn = "^1.1"

[dev-dependencies]
async-global-executor = "^3.1"
async-io = "^2.0"
futures-lite = "^2.0"
serde_json = "^1.0"
waker-fn = "^1.1"
Expand Down
94 changes: 94 additions & 0 deletions examples/c.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,94 @@
use futures_lite::StreamExt;
use lapin::{Connection, ConnectionProperties, options::*, types::FieldTable};
use tracing::info;

fn main() {
if std::env::var("RUST_LOG").is_err() {
unsafe { std::env::set_var("RUST_LOG", "info") };
}

tracing_subscriber::fmt::init();

let addr = std::env::var("AMQP_ADDR").unwrap_or_else(|_| "amqp://127.0.0.1:5672/%2f".into());
let recovery_config = lapin::RecoveryConfig::default().auto_recover_channels();

async_global_executor::block_on(async {
let conn = Connection::connect(
&addr,
ConnectionProperties::default().with_experimental_recovery_config(recovery_config),
)
.await
.expect("connection error");

info!("CONNECTED");

//receive channel
let channel = conn.create_channel().await.expect("create_channel");
info!(state=?conn.status().state());

let queue = channel
.queue_declare(
"hello-recover",
QueueDeclareOptions::default(),
FieldTable::default(),
)
.await
.expect("queue_declare");
info!(state=?conn.status().state());
info!(?queue, "Declared queue");

let ch = channel.clone();
async_global_executor::spawn(async move {
loop {
async_io::Timer::after(std::time::Duration::from_secs(1)).await;
info!("Trigger failure");
assert!(
ch.queue_declare(
"fake queue",
QueueDeclareOptions {
passive: true,
..QueueDeclareOptions::default()
},
FieldTable::default(),
)
.await
.is_err()
);
}
})
.detach();

info!("will consume");
let mut consumer = channel
.basic_consume(
"hello-recover",
"my_consumer",
BasicConsumeOptions::default(),
FieldTable::default(),
)
.await
.expect("basic_consume");
info!(state=?conn.status().state());

let mut count = 0;
while let Some(delivery) = consumer.next().await {
info!(message=?delivery, "received message");
if let Ok(delivery) = delivery {
let data = str::from_utf8(&delivery.data).expect("invalid utf8 data");

Check failure on line 77 in examples/c.rs

View workflow job for this annotation

GitHub Actions / build_and_test (ubuntu-latest, 1.85.0) 

no function or associated item named `from_utf8` found for type `str` in the current scope
println!(
"{}",
data
);
delivery
.ack(BasicAckOptions::default())
.await
.expect("basic_ack");
count += 1;
if data == "STOP" {
println!("Received {} msgs", count);
break;
}
}
}
})
}
111 changes: 111 additions & 0 deletions examples/p.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,111 @@
use lapin::{BasicProperties, Connection, ConnectionProperties, options::*, types::FieldTable};
use std::sync::{Arc, atomic::{AtomicUsize, Ordering}};
use tracing::info;

fn main() {
if std::env::var("RUST_LOG").is_err() {
unsafe { std::env::set_var("RUST_LOG", "info"); }
}

tracing_subscriber::fmt::init();

let addr = std::env::var("AMQP_ADDR").unwrap_or_else(|_| "amqp://127.0.0.1:5672/%2f".into());
let recovery_config = lapin::RecoveryConfig::default().auto_recover_channels();

async_global_executor::block_on(async {
let conn = Connection::connect(
&addr,
ConnectionProperties::default().with_experimental_recovery_config(recovery_config),
)
.await
.expect("connection error");

info!("CONNECTED");

let channel1 = conn.create_channel().await.expect("create_channel");
channel1
.confirm_select(ConfirmSelectOptions::default())
.await
.expect("confirm_select");
channel1
.queue_declare(
"hello-recover",
QueueDeclareOptions::default(),
FieldTable::default(),
)
.await
.expect("queue_declare");

let count = Arc::new(AtomicUsize::new(0));
let counter = count.clone();

let ch = channel1.clone();
async_global_executor::spawn(async move {
loop {
async_io::Timer::after(std::time::Duration::from_secs(1)).await;
info!("Trigger failure");
assert!(
ch.queue_declare(
"fake queue",
QueueDeclareOptions {
passive: true,
..QueueDeclareOptions::default()
},
FieldTable::default(),
)
.await
.is_err()
);
counter.fetch_add(1, Ordering::SeqCst);
}
})
.detach();

let mut published = 0;
let mut errors = 0;
info!("will publish");
loop {
let res = channel1
.basic_publish(
"",
"hello-recover",
BasicPublishOptions::default(),
b"before",
BasicProperties::default(),
)
.await;
let res = if let Ok(res) = res {
res.await.map(|_| ())
} else {
res.map(|_| ())
};
match res {
Ok(()) => {
published += 1;
}
Err(err) => {
if !err.is_amqp_soft_error() {
panic!("{}", err);
}
errors += 1;
if let Some(notifier) = err.notifier() {
notifier.await
}
}
}
if count.load(Ordering::SeqCst) > 10 {
println!("Published {} with {} errors", published, errors);
channel1
.basic_publish(
"",
"hello-recover",
BasicPublishOptions::default(),
b"STOP",
BasicProperties::default(),
)
.await.unwrap();
break;
}
}
});
}
122 changes: 122 additions & 0 deletions examples/t.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,122 @@
use lapin::{
BasicProperties, Connection, ConnectionProperties, message::DeliveryResult, options::*,
publisher_confirm::Confirmation, types::FieldTable,
};
use tracing::info;

fn main() {
if std::env::var("RUST_LOG").is_err() {
std::env::set_var("RUST_LOG", "info");

Check failure on line 9 in examples/t.rs

View workflow job for this annotation

GitHub Actions / build_and_test (ubuntu-latest, beta) 

call to unsafe function `set_var` is unsafe and requires unsafe block

Check failure on line 9 in examples/t.rs

View workflow job for this annotation

GitHub Actions / build_and_test (ubuntu-latest, stable) 

call to unsafe function `set_var` is unsafe and requires unsafe block

Check failure on line 9 in examples/t.rs

View workflow job for this annotation

GitHub Actions / build_and_test (ubuntu-latest, nightly) 

call to unsafe function `set_var` is unsafe and requires unsafe block

Check failure on line 9 in examples/t.rs

View workflow job for this annotation

GitHub Actions / build_and_test (ubuntu-latest, beta) 

call to unsafe function `set_var` is unsafe and requires unsafe block

Check failure on line 9 in examples/t.rs

View workflow job for this annotation

GitHub Actions / build_and_test (ubuntu-latest, stable) 

call to unsafe function `set_var` is unsafe and requires unsafe block

Check failure on line 9 in examples/t.rs

View workflow job for this annotation

GitHub Actions / build_and_test (ubuntu-latest, 1.85.0) 

call to unsafe function `set_var` is unsafe and requires unsafe block

Check failure on line 9 in examples/t.rs

View workflow job for this annotation

GitHub Actions / build_and_test (ubuntu-latest, nightly) 

call to unsafe function `set_var` is unsafe and requires unsafe block

Check failure on line 9 in examples/t.rs

View workflow job for this annotation

GitHub Actions / build_and_test (ubuntu-latest, 1.85.0) 

call to unsafe function `set_var` is unsafe and requires unsafe block
}

tracing_subscriber::fmt::init();

let addr = std::env::var("AMQP_ADDR").unwrap_or_else(|_| "amqp://127.0.0.1:5672/%2f".into());
let recovery_config = lapin::RecoveryConfig::default().auto_recover_channels();

async_global_executor::block_on(async {
let conn = Connection::connect(
&addr,
ConnectionProperties::default().with_experimental_recovery_config(recovery_config),
)
.await
.expect("connection error");

info!("CONNECTED");

{
let channel1 = conn.create_channel().await.expect("create_channel");
let channel2 = conn.create_channel().await.expect("create_channel");
channel1
.confirm_select(ConfirmSelectOptions::default())
.await
.expect("confirm_select");
channel1
.queue_declare(
"recover-test",
QueueDeclareOptions::default(),
FieldTable::default(),
)
.await
.expect("queue_declare");

info!("will consume");
let channel = channel2.clone();
channel2
.basic_consume(
"recover-test",
"my_consumer",
BasicConsumeOptions::default(),
FieldTable::default(),
)
.await
.expect("basic_consume")
.set_delegate(move |delivery: DeliveryResult| {
let channel = channel.clone();
async move {
info!(message=?delivery, "received message");
if let Ok(Some(delivery)) = delivery {
delivery
.ack(BasicAckOptions::default())
.await
.expect("basic_ack");
if &delivery.data[..] == b"after" {
channel
.basic_cancel("my_consumer", BasicCancelOptions::default())
.await
.expect("basic_cancel");
}
}
}
});

info!("will publish");
let confirm = channel1
.basic_publish(
"",
"recover-test",
BasicPublishOptions::default(),
b"before",
BasicProperties::default(),
)
.await
.expect("basic_publish")
.await
.expect("publisher-confirms");
assert_eq!(confirm, Confirmation::Ack(None));

info!("before fail");
assert!(
channel1
.queue_declare(
"fake queue",
QueueDeclareOptions {
passive: true,
..QueueDeclareOptions::default()
},
FieldTable::default(),
)
.await
.is_err()
);
info!("after fail");

info!("publish after");
let confirm = channel1
.basic_publish(
"",
"recover-test",
BasicPublishOptions::default(),
b"after",
BasicProperties::default(),
)
.await
.expect("basic_publish")
.await
.expect("publisher-confirms");
assert_eq!(confirm, Confirmation::Ack(None));
}

conn.run().expect("conn.run");
});
}
Loading