diff --git a/crates/core/database/src/amqp/amqp.rs b/crates/core/database/src/amqp/amqp.rs index 5358bc34..16326045 100644 --- a/crates/core/database/src/amqp/amqp.rs +++ b/crates/core/database/src/amqp/amqp.rs @@ -29,7 +29,7 @@ impl AMQP { } } - pub async fn new_auto() -> AMQP { + pub async fn new_auto() -> revolt_result::Result { let config = revolt_config::config().await; let connection = Connection::open(&OpenConnectionArguments::new( @@ -46,21 +46,26 @@ impl AMQP { .await .expect("Failed to open RabbitMQ channel"); - channel - .exchange_declare( - ExchangeDeclareArguments::new(&config.pushd.exchange, "direct") - .durable(true) - .finish(), - ) - .await - .expect("Failed to declare exchange"); - - AMQP::new(connection, channel) + let mut resp = AMQP::new(connection, channel); + resp.configure_channels().await?; + Ok(resp) } - pub async fn configure_channels(&self) -> revolt_result::Result<()> { + pub async fn repoen_channel(&mut self) { + self.channel = self + .connection + .open_channel(None) + .await + .expect("Failed to open RabbitMQ channel"); + } + + pub async fn configure_channels(&mut self) -> revolt_result::Result<()> { let config = revolt_config::config().await; + if !self.channel.is_open() { + self.repoen_channel().await; + } + self.channel .exchange_declare( ExchangeDeclareArguments::new( diff --git a/crates/delta/src/main.rs b/crates/delta/src/main.rs index e3522de1..e3cf43a4 100644 --- a/crates/delta/src/main.rs +++ b/crates/delta/src/main.rs @@ -119,7 +119,7 @@ pub async fn web() -> Rocket { .await .expect("Failed to declare exchange"); - let amqp = AMQP::new(connection, channel); + let mut amqp = AMQP::new(connection, channel); amqp.configure_channels() .await .expect("Failed to configure channels");