refactor: use iterator with into_iter for cleaner event processing

- Move events instead of borrowing
- Use collect with Result for fail-fast on config/serialization errors

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: psteinroe

src/sink/rabbitmq.rs

@@ -207,39 +207,37 @@ impl Sink for RabbitmqSink {
             return Ok(());
         }
 
-        // Prepare all messages first.
-        let mut prepared: Vec<(String, String, Vec<u8>)> = Vec::with_capacity(events.len());
-
-        for event in &events {
-            // Resolve exchange from event metadata or config.
-            let exchange = self.resolve_exchange(event).ok_or_else(|| {
-                etl::etl_error!(
-                    etl::error::ErrorKind::ConfigError,
-                    "No exchange configured",
-                    "Exchange must be provided in sink config or event metadata"
-                )
-            })?;
-
-            // Resolve routing key from event metadata or config.
-            let routing_key = self.resolve_routing_key(event).ok_or_else(|| {
-                etl::etl_error!(
-                    etl::error::ErrorKind::ConfigError,
-                    "No routing_key configured",
-                    "Routing key must be provided in sink config or event metadata"
-                )
-            })?;
-
-            // Serialize payload to JSON.
-            let payload = serde_json::to_vec(&event.payload).map_err(|e| {
-                etl::etl_error!(
-                    etl::error::ErrorKind::InvalidData,
-                    "Failed to serialize payload to JSON",
-                    e.to_string()
-                )
-            })?;
+        // Resolve routing and serialize payloads upfront (fail fast, minimize lock time).
+        let prepared: Vec<_> = events
+            .into_iter()
+            .map(|event| {
+                let exchange = self.resolve_exchange(&event).ok_or_else(|| {
+                    etl::etl_error!(
+                        etl::error::ErrorKind::ConfigError,
+                        "No exchange configured",
+                        "Exchange must be provided in sink config or event metadata"
+                    )
+                })?;
 
-            prepared.push((exchange.to_string(), routing_key.to_string(), payload));
-        }
+                let routing_key = self.resolve_routing_key(&event).ok_or_else(|| {
+                    etl::etl_error!(
+                        etl::error::ErrorKind::ConfigError,
+                        "No routing_key configured",
+                        "Routing key must be provided in sink config or event metadata"
+                    )
+                })?;
+
+                let payload = serde_json::to_vec(&event.payload).map_err(|e| {
+                    etl::etl_error!(
+                        etl::error::ErrorKind::InvalidData,
+                        "Failed to serialize payload to JSON",
+                        e.to_string()
+                    )
+                })?;
+
+                Ok((exchange.to_string(), routing_key.to_string(), payload))
+            })
+            .collect::<EtlResult<Vec<_>>>()?;
 
         // Fire all publishes and collect confirms (holding lock minimally).
         let confirms = {