Merge pull request #22 from marcomq/dev

Add Status info

Author: marcomq

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "mq-bridge"
-version = "0.2.9"
+version = "0.2.10"
 edition = "2021"
 authors = ["Marco Mengelkoch"]
 rust-version = "1.75"  # 2021 edition was released in 1.56
@@ -76,6 +76,7 @@ hyper = { version = "1.8", optional = true, features = ["client", "server", "htt
 hyper-util = { version = "0.1", optional = true, features = ["tokio", "client", "client-legacy", "server", "http1", "http2"] }
 hyper-rustls = { version = "0.27", optional = true, features = ["http2"] }
 http-body-util = { version = "0.1", optional = true }
+http-body = { version = "1.0", optional = true }
 h2 = { version = "0.4", optional = true }
 flate2 = { version = "1.0", optional = true }
 base64 = { version = "0.22", optional = true }
@@ -129,7 +130,7 @@ amqp = ["lapin", "rustls", "rustls-pemfile", "url"]
 nats = ["async-nats", "rustls", "rustls-pemfile"]
 mongodb = ["dep:mongodb"]
 mqtt = ["rumqttc", "rustls", "tokio-rustls", "rustls-pemfile", "url"]
-http = ["hyper", "hyper-util", "hyper-rustls", "http-body-util", "rustls", "tokio-rustls", "rustls-pemfile", "webpki-roots", "h2", "dep:flate2", "dep:base64"]
+http = ["hyper", "hyper-util", "hyper-rustls", "http-body-util", "dep:http-body", "rustls", "tokio-rustls", "rustls-pemfile", "webpki-roots", "h2", "dep:flate2", "dep:base64"]
 aws = ["aws-config", "aws-sdk-sqs", "aws-sdk-sns"]
 ibm-mq = ["mqi"]
 zeromq = ["dep:zeromq"]

Cargo.toml

@@ -44,6 +44,12 @@ It may still be possible that there are issues with
 - nats, if jetstream support is disabled
 - TLS integration, as this also hasn't been tested a lot and is usually non-trivial to set up
 
+
+Due to the large code base, it may still be possible that some endpoints may show
+issues in production; therefore, they should be tested locally first. They all worked locally
+for me and didn't show data loss during simple in-flight broker restarts.
+Kafka, MongoDB, IBM-MQ, Files, and Memory are considered production-ready.
+
 ### When to use mq-bridge
 *   **Hybrid Messaging**: Connect systems speaking different protocols (e.g., MQTT to Kafka) without writing custom adapters.
 *   **Infrastructure Abstraction**: Write business logic that consumes `CanonicalMessage`s, allowing you to swap the underlying transport (e.g., switching from RabbitMQ to NATS) via configuration.
@@ -64,34 +70,40 @@ It may still be possible that there are issues with
 *   **Middleware**: Components that intercept and process messages (e.g., for error handling).
 *   **Handler**: A programmatic component for business logic, such as transforming/consuming messages (`CommandHandler`) or subscribe them (`EventHandler`).
 
-## Endpoint Behavior
+## Backend Features & Configuration
 
-`mq-bridge` endpoints generally default to a **Consumer** pattern (Queue), where messages are persisted (if supported by the backend) and distributed among workers.
+`mq-bridge` endpoints generally default to a **Consumer** pattern (Queue), where messages are persisted and distributed among workers. To achieve **Subscriber** (Pub/Sub) behavior, specific configuration is required.
 
-To achieve **Subscriber** (Pub/Sub) behavior—where messages are broadcast to all active instances—you must configure the specific backend accordingly. There is no global "subscriber mode" toggle; it is determined by the configuration of the endpoint.
+The table below summarizes the capabilities and configuration for each backend:
 
-| Backend | Default Behavior (Queue) | Configuration for Subscriber (Pub/Sub) | Response Support |
+| Backend | Subscriber Config (Pub/Sub) | Request-Reply | Nack Support |
 | :--- | :--- | :--- | :--- |
-| **Kafka** | Persistent (Consumer Group) | Omit `group_id` (generates unique ID) | No |
-| **NATS** | Persistent (JetStream Durable) | Set `subscriber_mode: true` | Yes |
-| **AMQP** | Persistent (Durable Queue) | Set `subscribe_mode: true` | No |
-| **MQTT** | Persistent Session | Set `clean_session: true` | No |
-| **IBM MQ** | Persistent Queue | Set `topic` instead of `queue` | No |
-| **MongoDB** | Persistent (Collection) | Set `change_stream: true` | Yes |
-| **SQLx** | Persistent (Table) | Not supported | No |
-| **AWS** | Persistent (SQS) | Not supported directly (Use SNS->SQS) | No |
-| **Memory** | Ephemeral (Channel) | Set `subscribe_mode: true` | Yes |
-| **File** | Queue (Reads from start) | Set `mode: subscribe` (Tails file) or `mode: group_subscribe` (Persistent tail) | No |
-| **HTTP** | Ephemeral (Request) | N/A | Yes (Implicit) |
-| **ZeroMQ** | Ephemeral (PULL) | Set `socket_type: "sub"` | No |
-
-### Response Mode
-The `response` output endpoint allows sending a reply back to the requester. This is useful for synchronous request-reply patterns (e.g., HTTP-to-NATS-to-HTTP).
-
-*   **Availability**: Only available if the **Input** endpoint supports request-reply (HTTP, NATS, Memory, MongoDB).
-*   **Configuration**: Use `response: {}` as the output endpoint.
+| **AMQP** | Set `subscribe_mode: true` | Emulated (Property) | **Yes** (Basic.nack) |
+| **AWS** | N/A (Use SNS) | No | **Yes** (Visibility Timeout) |
+| **File** | Set `mode: subscribe` | No | Simulated (In-Memory) |
+| **gRPC** | N/A | No | No |
+| **HTTP** | N/A | **Native** (Implicit) | **Yes** (HTTP 500) |
+| **IBM MQ** | Set `topic` | No | **Yes** (Tx Rollback) |
+| **Kafka** | Omit `group_id` | Emulated (Header) | Eventual (Skip Offset) |
+| **Memory** | Set `subscribe_mode: true` | Emulated (Metadata) | **Yes** (Re-queue), by default **disabled** |
+| **MongoDB** | Set `change_stream: true` | Emulated (Metadata) | **Yes** (Unlock) |
+| **MQTT** | Set `clean_session: true` | Emulated (Property) | Eventual (Skip Ack) |
+| **NATS** | Set `subscriber_mode: true` | **Native** (Inbox) | **Yes** (JetStream Nak) |
+| **Sled** | Set `delete_after_read: false` | No | **Yes** (Tx Rollback) |
+| **SQLx** | Not supported | No | Eventual (Skip Delete) |
+| **ZeroMQ** | Set `socket_type: "sub"` | **Native** (REQ/REP) | No |
+
+### Feature Details
+*   **Request-Reply**:
+    *   **Native**: Uses protocol-level correlation (e.g., HTTP connection, NATS reply subject).
+    *   **Emulated**: Publishes a new message to a reply destination (specified by the `reply_to` metadata field) carrying a `correlation_id` metadata field.
+*   **Nack Support**: If "Yes", the backend supports explicit negative acknowledgement triggering redelivery. "Eventual" means redelivery depends on timeout or connection drop. "Simulated" is handled in-memory by the bridge.
+
+### Response Endpoint
+The `response` output endpoint allows sending a reply back to the requester. This is useful for synchronous request-reply patterns (e.g., HTTP-to-NATS-to-HTTP). Use `response: {}` as the output endpoint configuration.
+
 *   **Caveats**:
-    *   If the input does not support responses (e.g., File, Kafka), the message sent to `response` will be dropped.
+    *   If the input does not support responses (e.g., File, SQLx), the message sent to `response` will be dropped.
     *   Ensure timeouts are configured correctly on the requester side, as the bridge processing time adds latency.
     *   Middleware that drops metadata (like `correlation_id`) may break the response chain.
 
@@ -339,18 +351,30 @@ The times are not stable yet, it is therefore recommended to perform the integra
 
 ## AI Disclaimer
 
-This library has been widely written with AI assistance. I used Gemini for planning and writing,
-CodeRabbit for reviews and Copilot/Claude for bugfixing and other small things.
+This library has been widely written with AI assistance. 
+
+Some of the code - the core for example, was originally written by myself, 
+but most other was generated by AI. I mostly used Gemini for 
+planning and writing, CodeRabbit for reviews and Claude for bugfixing and 
+more complicated tasks that Gemini couldn't solve properly.
 While some of the AI output was great, some other output wasn't.
 I am aware that in year 2026, AI is still not generating perfect code and sometimes
-even breaks simple stuff. I reviewed all the
-output code and re-specified it or changed the code manually whan insuficcient.
+breaks simple stuff or forgets important lines during refactorings that then cause
+severe issues. 
+I reviewed all the output code, cleaned it up manually, 
+re-specified and refactored it when insuficcient.
+**I do trust the current code as much as if it would be completely written by myself.**
+
 I didn't change the AI code appearance, so you will sometimes still see code that just
 looks as it is plain from AI and also most of the readme here was actually written
 by AI. I don't think it is bad practice, to keep the original code and text appearance. 
-I'm not an english native speaker, so the AI output for text is mostly just
-way better what I could write. For AI code, the readability is usually
-sufficient, even if it is sometimes much more verbose what I would write in code.
+I'm not an english native speaker, so the AI output for english text is just
+way better than my text. For AI code, the readability is usually
+good, even if it is more verbose than what I would write.
+However, especially for the different endpoints, there is already a lot of existing
+code and the AI could also just assist a lot there. Thats mostly the reason,
+why there are so many available endpoints in this library, they just could be added
+very easily and showed a sufficient code quality.
 
 ## License
 `mq-bridge` is licensed under the MIT License.

README.md

@@ -1,7 +1,7 @@
 use crate::canonical_message::tracing_support::LazyMessageIds;
 use crate::models::AmqpConfig;
 use crate::traits::{
-    BatchCommitFunc, BoxFuture, ConsumerError, MessageConsumer, MessageDisposition,
+    BatchCommitFunc, BoxFuture, ConsumerError, EndpointStatus, MessageConsumer, MessageDisposition,
     MessagePublisher, PublisherError, ReceivedBatch, Sent, SentBatch,
 };
 use crate::CanonicalMessage;
@@ -297,6 +297,33 @@ impl MessagePublisher for AmqpPublisher {
         }
     }
 
+    async fn status(&self) -> EndpointStatus {
+        let state = self.state.read().await;
+        let conn_status = state.connection.status();
+        let chan_status = state.channel.status();
+        let healthy = conn_status.connected() && chan_status.connected();
+        let error = if !healthy {
+            Some(format!(
+                "Connection: '{:?}', Channel: '{:?}'",
+                conn_status.state(),
+                chan_status.state()
+            ))
+        } else {
+            None
+        };
+        EndpointStatus {
+            healthy,
+            error,
+            target: if self.exchange.is_empty() {
+                self.queue.clone()
+            } else {
+                self.exchange.clone()
+            },
+            details: serde_json::json!({ "queue": self.queue, "delayed_ack": self.delayed_ack }),
+            ..Default::default()
+        }
+    }
+
     fn as_any(&self) -> &dyn Any {
         self
     }
@@ -308,6 +335,7 @@ pub struct AmqpConsumer {
     channel: Channel,
     queue: String,
     is_poisoned: Arc<AtomicBool>,
+    prefetch: u16,
 }
 
 impl AmqpConsumer {
@@ -409,6 +437,7 @@ impl AmqpConsumer {
             channel,
             queue: queue_name,
             is_poisoned: Arc::new(AtomicBool::new(false)),
+            prefetch: prefetch_count,
         })
     }
 }
@@ -620,6 +649,51 @@ impl MessageConsumer for AmqpConsumer {
         Ok(ReceivedBatch { messages, commit })
     }
 
+    async fn status(&self) -> EndpointStatus {
+        let conn_status = self._conn.status();
+        let chan_status = self.channel.status();
+        let mut healthy = conn_status.connected() && chan_status.connected();
+        let mut pending: Option<usize> = None;
+        let mut error: Option<String> = None;
+
+        if healthy {
+            let passive_declare = self.channel.queue_declare(
+                &self.queue,
+                lapin::options::QueueDeclareOptions {
+                    passive: true,
+                    ..Default::default()
+                },
+                lapin::types::FieldTable::default(),
+            );
+            match tokio::time::timeout(Duration::from_secs(2), passive_declare).await {
+                Ok(Ok(q)) => pending = Some(q.message_count() as usize),
+                Ok(Err(e)) => {
+                    healthy = false;
+                    error = Some(e.to_string());
+                }
+                Err(e) => {
+                    healthy = false;
+                    error = Some(e.to_string());
+                }
+            }
+        } else {
+            error = Some(format!(
+                "Connection: '{:?}', Channel: '{:?}'",
+                conn_status.state(),
+                chan_status.state()
+            ));
+        }
+
+        EndpointStatus {
+            healthy,
+            target: self.queue.clone(),
+            pending,
+            error,
+            capacity: Some(self.prefetch as usize),
+            ..Default::default()
+        }
+    }
+
     fn as_any(&self) -> &dyn Any {
         self
     }
@@ -631,7 +705,16 @@ async fn handle_replies(
     dispositions: &[MessageDisposition],
 ) {
     for ((reply_to, correlation_id), disposition) in reply_infos.iter().zip(dispositions.iter()) {
-        if let (Some(rt), MessageDisposition::Reply(resp)) = (reply_to, disposition) {
+        let payload = match (disposition, reply_to) {
+            (MessageDisposition::Reply(resp), Some(_)) => Some(resp.payload.clone()),
+            (MessageDisposition::Reply(_), None) => {
+                tracing::warn!("MessageDisposition::Reply received but no reply_to address found in original message");
+                None
+            }
+            _ => None,
+        };
+
+        if let (Some(rt), Some(body)) = (reply_to, payload) {
             let mut props = BasicProperties::default();
             if let Some(cid) = correlation_id {
                 props = props.with_correlation_id(cid.clone().into());
@@ -643,7 +726,7 @@ async fn handle_replies(
                     "", // Default exchange
                     rt,
                     BasicPublishOptions::default(),
-                    &resp.payload,
+                    &body,
                     props,
                 )
                 .await