working impl

2026-03-26 19:31:32 +01:00 · 2024-09-20 11:39:39 +02:00
parent 7557769388
commit 10c48f37ed
13 changed files with 38043 additions and 107 deletions
--- a/src/consumer.rs
+++ b/src/consumer.rs
@@ -1,9 +1,73 @@
-use zettle_db::rabbitmq::RabbitMQ;
+use lapin::{
+    options::*, types::FieldTable, Connection, ConnectionProperties, Consumer, Result,
+};
+use tracing::{info, error};
+use futures_lite::stream::StreamExt;

-#[tokio::main(flavor = "multi_thread", worker_threads = 2)]
-async fn main() {
-    let rabbitmq = RabbitMQ::new().await;
-    let queue_name = rabbitmq.declare_queue("amqprs.examples.basic").await;
-    rabbitmq.bind_queue(&queue_name.0, "amq.topic", "amqprs.example").await;
-    //...
+pub struct RabbitMQConsumer {
+    pub connection: Connection,
+    pub    consumer: Consumer,
+}
+
+impl RabbitMQConsumer {
+    pub async fn new(addr: &str, queue: &str) -> Result<Self> {
+        let connection = Connection::connect(addr, ConnectionProperties::default()).await?;
+        let channel = connection.create_channel().await?;
+
+        // Declare the queue (in case it doesn't exist)
+        channel
+            .queue_declare(
+                queue,
+                QueueDeclareOptions::default(),
+                FieldTable::default(),
+            )
+            .await?;
+
+        let consumer = channel
+            .basic_consume(
+                queue,
+                "consumer",
+                BasicConsumeOptions::default(),
+                FieldTable::default(),
+            )
+            .await?;
+
+        Ok(Self { connection, consumer })
+    }
+
+    pub async fn run(&mut self) -> Result<()> {
+        info!("Consumer started - waiting for messages");
+        while let Some(delivery) = self.consumer.next().await {
+            match delivery {
+                Ok(delivery) => {
+                    let message = std::str::from_utf8(&delivery.data).unwrap_or("Invalid UTF-8");
+                    info!("Received message: {}", message);
+                    
+                    // Process the message here
+                    // For example, you could deserialize it and perform some action
+                    
+                    delivery.ack(BasicAckOptions::default()).await?;
+                },
+                Err(e) => error!("Failed to consume message: {:?}", e),
+            }
+        }
+
+        Ok(())
+    }
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    // Set up tracing
+    tracing_subscriber::fmt::init();
+
+    let addr = "amqp://guest:guest@localhost:5672";
+    let queue = "hello";
+
+    let mut consumer = RabbitMQConsumer::new(addr, queue).await?;
+    
+    info!("Starting consumer");
+    consumer.run().await?;
+
+    Ok(())
 }
--- a/src/example.rs
+++ b/src/example.rs
@@ -0,0 +1,104 @@
+use amqprs::{
+    callbacks::{DefaultChannelCallback, DefaultConnectionCallback},
+    channel::{
+        BasicConsumeArguments, BasicPublishArguments, QueueBindArguments, QueueDeclareArguments,
+    },
+    connection::{Connection, OpenConnectionArguments},
+    consumer::DefaultConsumer,
+    BasicProperties,
+};
+use tokio::time;
+
+use tracing_subscriber::{fmt, prelude::*, EnvFilter};
+
+#[tokio::main(flavor = "multi_thread", worker_threads = 2)]
+async fn main() {
+    // construct a subscriber that prints formatted traces to stdout
+    // global subscriber with log level according to RUST_LOG
+    tracing_subscriber::registry()
+        .with(fmt::layer())
+        .with(EnvFilter::from_default_env())
+        .try_init()
+        .ok();
+
+    // open a connection to RabbitMQ server
+    let connection = Connection::open(&OpenConnectionArguments::new(
+        "localhost",
+        5672,
+        "guest",
+        "guest",
+    ))
+    .await
+    .unwrap();
+    connection
+        .register_callback(DefaultConnectionCallback)
+        .await
+        .unwrap();
+
+    // open a channel on the connection
+    let channel = connection.open_channel(None).await.unwrap();
+    channel
+        .register_callback(DefaultChannelCallback)
+        .await
+        .unwrap();
+
+    // declare a durable queue
+    let (queue_name, _, _) = channel
+        .queue_declare(QueueDeclareArguments::durable_client_named(
+            "amqprs.examples.basic",
+        ))
+        .await
+        .unwrap()
+        .unwrap();
+
+    // bind the queue to exchange
+    let routing_key = "amqprs.example";
+    let exchange_name = "amq.topic";
+    channel
+        .queue_bind(QueueBindArguments::new(
+            &queue_name,
+            exchange_name,
+            routing_key,
+        ))
+        .await
+        .unwrap();
+
+    //////////////////////////////////////////////////////////////////////////////
+    // start consumer with given name
+    let args = BasicConsumeArguments::new(&queue_name, "example_basic_pub_sub");
+
+    channel
+        .basic_consume(DefaultConsumer::new(args.no_ack), args)
+        .await
+        .unwrap();
+
+    //////////////////////////////////////////////////////////////////////////////
+    // publish message
+    let content = String::from(
+        r#"
+            {
+                "publisher": "example"
+                "data": "Hello, amqprs!"
+            }
+        "#,
+    )
+    .into_bytes();
+
+    // create arguments for basic_publish
+    let args = BasicPublishArguments::new(exchange_name, routing_key);
+
+    channel
+        .basic_publish(BasicProperties::default(), content, args)
+        .await
+        .unwrap();
+    tracing::info!("finished");
+
+    // keep the `channel` and `connection` object from dropping before pub/sub is done.
+    // channel/connection will be closed when drop.
+    time::sleep(time::Duration::from_secs(1)).await;
+    // explicitly close
+
+    channel.close().await.unwrap();
+    connection.close().await.unwrap();
+}
+
--- a/src/rabbitmq/mod.rs
+++ b/src/rabbitmq/mod.rs
@@ -1,56 +1,44 @@
-use amqprs::{
-    callbacks::{DefaultChannelCallback, DefaultConnectionCallback},
-    channel::{
-        BasicConsumeArguments, BasicPublishArguments, QueueBindArguments, QueueDeclareArguments,
-    },
-    connection::{Connection, OpenConnectionArguments},
-    consumer::DefaultConsumer,
-    BasicProperties,
+use lapin::{
+    options::*, types::FieldTable, BasicProperties, Channel, Connection, ConnectionProperties, Result
 };

-pub struct RabbitMQ {
+pub struct RabbitMQClient {
    pub connection: Connection,
-    pub channel: amqprs::channel::Channel,
+    pub channel: Channel,
 }

-impl RabbitMQ {
-    pub async fn new() -> Self {
-        let connection = Connection::open(&OpenConnectionArguments::new(
-            "localhost",
-            5672,
-            "user",
-            "bitnami",
-        ))
-        .await
-        .unwrap();
+impl RabbitMQClient {
+    pub async fn new(addr: &str) -> Result<Self> {
+        let connection = Connection::connect(addr, ConnectionProperties::default()).await?;
+        let channel = connection.create_channel().await?;

-        connection
-            .register_callback(DefaultConnectionCallback)
-            .await
-            .unwrap();
-
-        let channel = connection.open_channel(None).await.unwrap();
-        channel
-            .register_callback(DefaultChannelCallback)
-            .await
-            .unwrap();
-
-        RabbitMQ { connection, channel }
+        Ok(Self { connection, channel })
    }

-    pub async fn declare_queue(&self, queue_name: &str) -> (String, u32, u32) {
+    pub async fn publish(&self, queue: &str, payload: &[u8]) -> Result<()> {
        self.channel
-            .queue_declare(QueueDeclareArguments::durable_client_named(queue_name))
-            .await
-            .unwrap()
-            .unwrap()
+            .basic_publish(
+                "",
+                queue,
+                Default::default(),
+                payload,
+                BasicProperties::default(),
+            )
+            .await?;
+
+        Ok(())
    }

-    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
-            .unwrap();
+    pub async fn consume(&self, queue: &str) -> Result<lapin::Consumer> {
+        let consumer = self.channel
+            .basic_consume(
+                queue,
+                "consumer",
+                BasicConsumeOptions::default(),
+                FieldTable::default(),
+            )
+            .await?;
+
+        Ok(consumer)
    }
 }
-
--- a/src/server.rs
+++ b/src/server.rs
@@ -1,9 +1,116 @@
-use zettle_db::rabbitmq::RabbitMQ;
+// use axum::{
+//     routing::post,
+//     Router,
+//     response::{IntoResponse, Response},
+//     Error, Extension, Json,
+// };
+// use serde::Deserialize;
+// use tracing_subscriber::{fmt, prelude::*, EnvFilter};
+
+// #[derive(Deserialize)]
+// struct IngressPayload {
+//     payload: String,
+// }
+
+// use tracing::{info, error};
+// use zettle_db::rabbitmq::RabbitMQClient;
+
+// async fn ingress_handler(
+//     // Extension(rabbitmq): Extension<RabbitMQ>,
+//     Json(payload): Json<IngressPayload>
+// ) -> Response {
+//     info!("Received payload: {:?}", payload.payload);
+//     let rabbitmqclient = RabbitMQClient::new("127.0.0.1").await;
+//     match rabbitmq.publish(&payload.payload).await {
+//         Ok(_) => {
+//             info!("Message published successfully");
+//             "thank you".to_string().into_response()
+//         },
+//         Err(e) => {
+//             error!("Failed to publish message: {:?}", e);
+//             (axum::http::StatusCode::INTERNAL_SERVER_ERROR, "Failed to publish message").into_response()
+//         }
+//     }
+// }
+
+// #[tokio::main(flavor = "multi_thread", worker_threads = 2)]
+// async fn main() -> Result<(), Error> {
+//     // Set up tracing
+//     tracing_subscriber::registry()
+//        .with(fmt::layer())
+//        .with(EnvFilter::from_default_env())
+//        .try_init()
+//        .ok();
+
+//     // Set up RabbitMQ
+//     // let rabbitmq = RabbitMQ::new("amqprs.examples.basic2", "amq.topic", "amqprs.example2").await;
+
+//     // Create Axum router
+//     let app = Router::new()
+//         .route("/ingress", post(ingress_handler));        // .layer(Extension(rabbitmq));
+
+//     tracing::info!("Listening on 0.0.0.0:3000");
+//     let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await.unwrap();
+//     axum::serve(listener, app).await.unwrap();
+
+//     Ok(())
+// }
+use axum::{
+    routing::post,
+    Router,
+    response::{IntoResponse, Response},
+    Error, Extension, Json,
+};
+use serde::Deserialize;
+use tracing_subscriber::{fmt, prelude::*, EnvFilter};
+use zettle_db::rabbitmq::RabbitMQClient;
+use std::sync::Arc;
+
+#[derive(Deserialize)]
+struct IngressPayload {
+    payload: String,
+}
+
+use tracing::{info, error};
+
+async fn ingress_handler(
+    Extension(rabbitmq): Extension<Arc<RabbitMQClient>>,
+    Json(payload): Json<IngressPayload>
+) -> Response {
+    info!("Received payload: {:?}", payload.payload);
+    match rabbitmq.publish("hello", payload.payload.as_bytes()).await {
+        Ok(_) => {
+            info!("Message published successfully");
+            "thank you".to_string().into_response()
+        },
+        Err(e) => {
+            error!("Failed to publish message: {:?}", e);
+            (axum::http::StatusCode::INTERNAL_SERVER_ERROR, "Failed to publish message").into_response()
+        }
+    }
+}

 #[tokio::main(flavor = "multi_thread", worker_threads = 2)]
-async fn main() {
-    let rabbitmq = RabbitMQ::new().await;
-    let queue_name = rabbitmq.declare_queue("amqprs.examples.basic").await;
-    rabbitmq.bind_queue(&queue_name.0, "amq.topic", "amqprs.example").await;
-    //...
+async fn main() -> Result<(), Error> {
+    // Set up tracing
+    tracing_subscriber::registry()
+       .with(fmt::layer())
+       .with(EnvFilter::from_default_env())
+       .try_init()
+       .ok();
+
+    // Set up RabbitMQ
+    let rabbitmq = Arc::new(RabbitMQClient::new("amqp://guest:guest@localhost:5672").await.expect("Failed to connect to RabbitMQ"));
+    
+    // Create Axum router
+    let app = Router::new()
+        .route("/ingress", post(ingress_handler))
+        .layer(Extension(rabbitmq));
+
+    tracing::info!("Listening on 0.0.0.0:3000");
+    let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await.unwrap();
+    axum::serve(listener, app).await.unwrap();
+
+    Ok(())
 }
+