crates/arkflow-plugin/src/input/amqp09.rs (1)

86-86:Consider making the internal channel size configurable.

The flume channel is hardcoded to 1000 messages, which might not be suitable for all use cases. High-throughput scenarios might benefit from a larger buffer.

Add a configuration option:

 pub struct Amqp09InputConfig {     /// Auto-acknowledge messages     pub auto_ack: Option<bool>,     /// Prefetch count     pub prefetch_count: Option<u16>,+    /// Internal channel buffer size (default: 1000)+    pub channel_buffer_size: Option<usize>, }

Then use it:

-let (sender, receiver) = flume::bounded::<Amqp09Msg>(1000);+let buffer_size = config.channel_buffer_size.unwrap_or(1000);+let (sender, receiver) = flume::bounded::<Amqp09Msg>(buffer_size);

crates/arkflow-plugin/src/output/amqp09.rs (1)

50-62:⚠️ Potential issue

Implement or remove unused configuration fields.

The following config fields are defined but never used in the implementation:

• persistent (line 51)
• priority (line 55)
• expiration (line 57)
• content_type (line 59)
• content_encoding (line 61)

These fields should be applied when constructingBasicProperties in thewrite method.

Apply this diff to use the configured properties:

-                    BasicProperties::default(),+                    {+                        let mut props = BasicProperties::default();+                        if let Some(persistent) = self.config.persistent {+                            props = props.with_delivery_mode(if persistent { 2 } else { 1 });+                        }+                        if let Some(priority) = self.config.priority {+                            props = props.with_priority(priority);+                        }+                        if let Some(ref expiration) = self.config.expiration {+                            props = props.with_expiration(expiration.clone());+                        }+                        if let Some(ref content_type) = self.config.content_type {+                            props = props.with_content_type(content_type.clone().into());+                        }+                        if let Some(ref content_encoding) = self.config.content_encoding {+                            props = props.with_content_encoding(content_encoding.clone().into());+                        }+                        props+                    },

Committable suggestion skipped: line range outside the PR's diff.

🤖 Prompt for AI Agents

In crates/arkflow-plugin/src/output/amqp09.rs around lines 50 to 62, the fieldspersistent, priority, expiration, content_type, and content_encoding are definedbut not used when constructing BasicProperties in the write method. To fix this,update the write method to apply these configuration fields to theBasicProperties builder, setting each property accordingly if the option isSome, so that the message properties reflect the configured values.

crates/arkflow-plugin/src/input/amqp09.rs (1)

216-256:🛠️ Refactor suggestion

Add error recovery for the consumer task.

The spawned task could panic and silently stop consuming messages. Consider adding panic recovery or at least logging when the task exits unexpectedly.

Store the task handle and monitor it:

+use tokio::task::JoinHandle; pub struct Amqp09Input {     // ... existing fields ...+    consumer_task: Arc<Mutex<Option<JoinHandle<()>>>>, }

Then in connect:

-tokio::spawn(async move {+let handle = tokio::spawn(async move {     // existing task code });+{+    let mut task_guard = self.consumer_task.lock().await;+    *task_guard = Some(handle);+}

This allows you to check task status in read() or implement recovery logic.

📝 Committable suggestion

‼️IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

        // Spawn the consumer loop and keep its JoinHandle for monitoring        let handle = tokio::spawn(async move {            loop {                tokio::select! {                    _ = cancellation_token.cancelled() => {                        break;                    }                    result = async {                        let mut consumer_guard = consumer_arc.lock().await;                        if let Some(consumer) = &mut *consumer_guard {                            consumer.next().await                        } else {                            None                        }                    } => {                        match result {                            Some(delivery_result) => {                                match delivery_result {                                    Ok(delivery) => {                                        if let Err(_) = sender_clone.send_async(Amqp09Msg::Delivery(delivery)).await {                                            break;                                        }                                    }                                    Err(e) => {                                        error!("RabbitMQ delivery error: {}", e);                                        if let Err(_) = sender_clone.send_async(Amqp09Msg::Err(Error::Disconnection)).await {                                            break;                                        }                                    }                                }                            }                            None => {                                // Consumer ended                                if let Err(_) = sender_clone.send_async(Amqp09Msg::Err(Error::EOF)).await {                                    break;                                }                            }                        }                    }                }            }        });        // Store the handle so we can .await it or check for panics later        {            let mut task_guard = self.consumer_task.lock().await;            *task_guard = Some(handle);        }

🤖 Prompt for AI Agents

In crates/arkflow-plugin/src/input/amqp09.rs around lines 216 to 256, thespawned consumer task lacks error recovery and may silently stop on panic.Modify the code to store the JoinHandle returned by tokio::spawn so you canmonitor the task status. Add panic recovery by wrapping the async block with acatch_unwind or use a task supervision pattern to log unexpected exits. Then, inthe connect function, keep the JoinHandle and check its status in read() orimplement logic to restart the task if it has stopped.

crates/arkflow-plugin/src/output/amqp09.rs (2)

57-57:Consider using a numeric type for the expiration field.

Theexpiration field is currently aString representing milliseconds. Consider usingOption<u64> orOption<std::time::Duration> for better type safety and to avoid runtime parsing errors.

-    /// Message expiration (in milliseconds)-    expiration: Option<String>,+    /// Message expiration+    expiration: Option<std::time::Duration>,

---

112-118:Add validation for unrecognized exchange types.

The code silently defaults toDirect for unrecognized exchange types, which could hide configuration errors.

 let exchange_type = match self.config.exchange_type.as_deref() {     Some("direct") => ExchangeKind::Direct,     Some("fanout") => ExchangeKind::Fanout,     Some("topic") => ExchangeKind::Topic,     Some("headers") => ExchangeKind::Headers,+    Some(unknown) => {+        return Err(Error::Config(format!("Unknown exchange type: {}", unknown)));+    }-    _ => ExchangeKind::Direct,+    None => ExchangeKind::Direct, };

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro