Module recipient_list 

Recipient List pattern: fan-out a single Exchange sequentially to multiple processors.

Implements the Enterprise Integration Pattern (EIP) “Recipient List”. Each recipient (processor) operates on the same mutable Exchange instance in the order added.

Use Cases

• Sequential enrichment steps (add headers, transform payload).
• Applying multiple independent processors where order matters (validation -> normalization -> formatting).
• Simple fan-out where recipients mutate the same exchange rather than produce clones.

Behavior

• Recipients execute strictly in insertion order.
• If a recipient returns an Err, processing stops and the error is propagated; remaining recipients are skipped.
• All recipients share and mutate the same Exchange (no copying or cloning performed).
• Last mutation wins (e.g. later recipient can overwrite out_msg set by earlier recipients).

Async vs Sync

Under the async feature the process method awaits each recipient’s async process call. Without it, a synchronous loop invokes process directly.

Example

use allora_core::{patterns::recipient_list::RecipientList, processor::ClosureProcessor, route::Route, Exchange, Message};
let r1 = ClosureProcessor::new(|exchange| { exchange.in_msg.set_header("step", "1"); Ok(()) });
let r2 = ClosureProcessor::new(|exchange| { exchange.out_msg = Some(Message::from_text("done")); Ok(()) });
let list = RecipientList::new().add(r1).add(r2);
let route = Route::new().add(list).build();
let mut exchange = Exchange::new(Message::from_text("payload"));
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async { route.run(&mut exchange).await.unwrap(); });
assert_eq!(exchange.in_msg.header("step"), Some("1"));
assert_eq!(exchange.out_msg.unwrap().body_text(), Some("done"));

Error Short-Circuit Example

use allora_core::{patterns::recipient_list::RecipientList, processor::ClosureProcessor, route::Route, Error, Exchange, Message };
let ok = ClosureProcessor::new(|exchange| { exchange.in_msg.set_header("visited", "yes"); Ok(()) });
let fail = ClosureProcessor::new(|_| Err(Error::processor("boom"))); // stop here
let after = ClosureProcessor::new(|exchange| { exchange.in_msg.set_header("after", "should_not"); Ok(()) });
let list = RecipientList::new().add(ok).add(fail).add(after);
let route = Route::new().add(list).build();
let mut exchange = Exchange::new(Message::from_text("payload"));
let rt = tokio::runtime::Runtime::new().unwrap();
let res = rt.block_on(async { route.run(&mut exchange).await });
assert!(res.is_err());
assert!(exchange.in_msg.header("visited").is_some());
assert!(exchange.in_msg.header("after").is_none());

Testing Tips

• Verify ordering by setting incremental headers (step=1, step=2).
• Inject an error in the middle to assert short-circuit behavior.
• Confirm that later recipients can override out_msg.
• Edge case: empty RecipientList should be a no-op (returns Ok(())).
• Ensure that the same Exchange instance is passed to all recipients.

Structs

RecipientList