starred/minne
1
0
Fork 0
mirror of https://github.com/perstarkse/minne.git synced 2026-03-30 06:12:00 +02:00
Code Issues Packages Projects Releases 10 Wiki Activity

init

Browse Source 
new file:   .aider.chat.history.md
This commit is contained in:
Per Stark
2024-09-18 12:29:27 +02:00
parent 303ddc5811
commit a3a1cdd9e5
43 changed files with 5419 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
src/rabbitmq/mod.rs
View File

@@ -7,7 +7,6 @@ use amqprs::{
consumer::DefaultConsumer,
BasicProperties,
};
use tokio::sync::mpsc;
pub struct RabbitMQ {
pub connection: Connection,
@@ -15,69 +14,43 @@ pub struct RabbitMQ {
}
impl RabbitMQ {
pub async fn new() -> Result<Self, Box<dyn std::error::Error>> {
pub async fn new() -> Self {
let connection = Connection::open(&OpenConnectionArguments::new(
"localhost",
5672,
"user",
"bitnami",
))
.await?;
.await
.unwrap();
connection
.register_callback(DefaultConnectionCallback)
.await?;
.await
.unwrap();
let channel = connection.open_channel(None).await?;
let channel = connection.open_channel(None).await.unwrap();
channel
.register_callback(DefaultChannelCallback)
.await?;
.await
.unwrap();
Ok(RabbitMQ { connection, channel })
RabbitMQ { connection, channel }
}
pub async fn declare_queue(&self, queue_name: &str) -> Result<(String, u32, u32), Box<dyn std::error::Error>> {
pub async fn declare_queue(&self, queue_name: &str) -> (String, u32, u32) {
self.channel
.queue_declare(QueueDeclareArguments::durable_client_named(queue_name))
.await?
.ok_or_else(|| Box::new(std::io::Error::new(std::io::ErrorKind::Other, "Failed to declare queue")) as Box<dyn std::error::Error>)
.await
.unwrap()
.unwrap()
}
pub async fn bind_queue(&self, queue_name: &str, exchange_name: &str, routing_key: &str) -> Result<(), Box<dyn std::error::Error>> {
pub async fn bind_queue(&self, queue_name: &str, exchange_name: &str, routing_key: &str) {
self.channel
.queue_bind(QueueBindArguments::new(queue_name, exchange_name, routing_key))
.await?;
Ok(())
}
pub async fn publish_message(&self, exchange_name: &str, routing_key: &str, content: String) -> Result<(), Box<dyn std::error::Error>> {
let args = BasicPublishArguments::new(exchange_name, routing_key);
self.channel
.basic_publish(BasicProperties::default(), content.into_bytes(), args)
.await?;
Ok(())
}
pub async fn consume_messages(&self, queue_name: &str, consumer_tag: &str) -> Result<mpsc::Receiver<String>, Box<dyn std::error::Error>> {
let (tx, rx) = mpsc::channel(100);
let args = BasicConsumeArguments::new(queue_name, consumer_tag);
let consumer = DefaultConsumer::new(args.no_ack);
let tx_clone = tx.clone();
self.channel.basic_consume(consumer, args)
.await?
.set_delegate(move |deliver: amqprs::consumer::DeliverEvent| {
let content_str = String::from_utf8_lossy(&deliver.content).to_string();
let tx = tx_clone.clone();
tokio::spawn(async move {
if let Err(e) = tx.send(content_str).await {
eprintln!("Failed to send message: {}", e);
}
});
});
Ok(rx)
.await
.unwrap();
}
}