Skip to main content

spring_batch_rs/core/
item.rs

1use crate::error::BatchError;
2
3/// Represents the result of reading an item from the reader.
4///
5/// This type is a specialized `Result` that can be:
6/// - `Ok(Some(R))` when an item is successfully read
7/// - `Ok(None)` when there are no more items to read (end of data)
8/// - `Err(BatchError)` when an error occurs during reading
9pub type ItemReaderResult<I> = Result<Option<I>, BatchError>;
10
11/// Represents the result of processing an item by the processor.
12///
13/// This type is a specialized `Result` that can be:
14/// - `Ok(Some(O))` when an item is successfully processed and should be passed to the writer
15/// - `Ok(None)` when an item is intentionally filtered out (not an error)
16/// - `Err(BatchError)` when an error occurs during processing
17pub type ItemProcessorResult<O> = Result<Option<O>, BatchError>;
18
19/// Represents the result of writing items by the writer.
20///
21/// This type is a specialized `Result` that can be:
22/// - `Ok(())` when items are successfully written
23/// - `Err(BatchError)` when an error occurs during writing
24pub type ItemWriterResult = Result<(), BatchError>;
25
26/// A trait for reading items.
27///
28/// This trait defines the contract for components that read items from a data source.
29/// It is one of the fundamental building blocks of the batch processing pipeline.
30///
31/// # Design Pattern
32///
33/// This follows the Strategy Pattern, allowing different reading strategies to be
34/// interchangeable while maintaining a consistent interface.
35///
36/// # Implementation Note
37///
38/// Implementors of this trait should:
39/// - Return `Ok(Some(item))` when an item is successfully read
40/// - Return `Ok(None)` when there are no more items to read (end of data)
41/// - Return `Err(BatchError)` when an error occurs during reading
42///
43/// # Example
44///
45/// ```compile_fail
46/// use spring_batch_rs::core::item::{ItemReader, ItemReaderResult};
47/// use spring_batch_rs::error::BatchError;
48///
49/// struct StringReader {
50///     items: Vec<String>,
51///     position: usize,
52/// }
53///
54/// impl ItemReader<String> for StringReader {
55///     fn read(&mut self) -> ItemReaderResult<String> {
56///         if self.position < self.items.len() {
57///             let item = self.items[self.position].clone();
58///             self.position += 1;
59///             Ok(Some(item))
60///         } else {
61///             Ok(None) // End of data
62///         }
63///     }
64/// }
65/// ```
66pub trait ItemReader<I> {
67    /// Reads an item from the reader.
68    ///
69    /// # Returns
70    /// - `Ok(Some(item))` when an item is successfully read
71    /// - `Ok(None)` when there are no more items to read (end of data)
72    /// - `Err(BatchError)` when an error occurs during reading
73    fn read(&self) -> ItemReaderResult<I>;
74}
75
76/// A trait for processing items.
77///
78/// This trait defines the contract for components that transform or process items
79/// in a batch processing pipeline. It takes an input item of type `I` and produces
80/// an output item of type `O`.
81///
82/// # Filtering
83///
84/// Returning `Ok(None)` filters the item silently: it is not passed to the writer
85/// and is counted in [`crate::core::step::StepExecution::filter_count`]. This is different from returning
86/// `Err(BatchError)` which counts as a processing error and may trigger fault tolerance.
87///
88/// # Design Pattern
89///
90/// This follows the Strategy Pattern, allowing different processing strategies to be
91/// interchangeable while maintaining a consistent interface.
92///
93/// # Type Parameters
94///
95/// - `I`: The input item type
96/// - `O`: The output item type
97///
98/// # Example
99///
100/// ```
101/// use spring_batch_rs::core::item::{ItemProcessor, ItemProcessorResult};
102/// use spring_batch_rs::error::BatchError;
103///
104/// struct AdultFilter;
105///
106/// struct Person { name: String, age: u32 }
107///
108/// impl ItemProcessor<Person, Person> for AdultFilter {
109///     fn process(&self, item: Person) -> ItemProcessorResult<Person> {
110///         if item.age >= 18 {
111///             Ok(Some(item)) // keep adults
112///         } else {
113///             Ok(None) // filter out minors
114///         }
115///     }
116/// }
117/// ```
118pub trait ItemProcessor<I, O> {
119    /// Processes an item and returns the processed result.
120    ///
121    /// # Parameters
122    /// - `item`: The item to process (consumed by value)
123    ///
124    /// # Returns
125    /// - `Ok(Some(processed_item))` when the item is successfully processed
126    /// - `Ok(None)` when the item is intentionally filtered out
127    /// - `Err(BatchError)` when an error occurs during processing
128    fn process(&self, item: I) -> ItemProcessorResult<O>;
129}
130
131/// A trait for writing items.
132///
133/// This trait defines the contract for components that write items to a data destination.
134/// It is one of the fundamental building blocks of the batch processing pipeline.
135///
136/// # Design Pattern
137///
138/// This follows the Strategy Pattern, allowing different writing strategies to be
139/// interchangeable while maintaining a consistent interface.
140///
141/// # Lifecycle Methods
142///
143/// This trait includes additional lifecycle methods:
144/// - `flush()`: Flushes any buffered data
145/// - `open()`: Initializes resources before writing starts
146/// - `close()`: Releases resources after writing completes
147///
148/// # Example
149///
150/// ```
151/// use spring_batch_rs::core::item::{ItemWriter, ItemWriterResult};
152/// use spring_batch_rs::error::BatchError;
153///
154/// struct ConsoleWriter;
155///
156/// impl ItemWriter<String> for ConsoleWriter {
157///     fn write(&self, items: &[String]) -> ItemWriterResult {
158///         for item in items {
159///             println!("{}", item);
160///         }
161///         Ok(())
162///     }
163/// }
164/// ```
165pub trait ItemWriter<O> {
166    /// Writes the given items.
167    ///
168    /// # Parameters
169    /// - `items`: A slice of items to write
170    ///
171    /// # Returns
172    /// - `Ok(())` when items are successfully written
173    /// - `Err(BatchError)` when an error occurs during writing
174    fn write(&self, items: &[O]) -> ItemWriterResult;
175
176    /// Flushes any buffered data.
177    ///
178    /// This method is called after a chunk of items has been written, and
179    /// allows the writer to flush any internally buffered data to the destination.
180    ///
181    /// # Default Implementation
182    ///
183    /// The default implementation does nothing and returns `Ok(())`.
184    ///
185    /// # Returns
186    /// - `Ok(())` when the flush operation succeeds
187    /// - `Err(BatchError)` when an error occurs during flushing
188    fn flush(&self) -> ItemWriterResult {
189        Ok(())
190    }
191
192    /// Opens the writer.
193    ///
194    /// This method is called before any items are written, and allows the writer
195    /// to initialize any resources it needs.
196    ///
197    /// # Default Implementation
198    ///
199    /// The default implementation does nothing and returns `Ok(())`.
200    ///
201    /// # Returns
202    /// - `Ok(())` when the open operation succeeds
203    /// - `Err(BatchError)` when an error occurs during opening
204    fn open(&self) -> ItemWriterResult {
205        Ok(())
206    }
207
208    /// Closes the writer.
209    ///
210    /// This method is called after all items have been written, and allows the writer
211    /// to release any resources it acquired.
212    ///
213    /// # Default Implementation
214    ///
215    /// The default implementation does nothing and returns `Ok(())`.
216    ///
217    /// # Returns
218    /// - `Ok(())` when the close operation succeeds
219    /// - `Err(BatchError)` when an error occurs during closing
220    fn close(&self) -> ItemWriterResult {
221        Ok(())
222    }
223}
224
225/// A pass-through processor that returns items unchanged.
226///
227/// Internal identity-function processor. [`ChunkOrientedStepBuilder::build`](crate::core::step::ChunkOrientedStepBuilder::build)
228/// uses it as the implicit processor when `.processor(...)` is not called and the
229/// reader's output type matches the writer's input type — callers never need to
230/// name this type directly.
231///
232/// # Type Parameters
233///
234/// - `T`: The item type that will be passed through unchanged.
235///
236/// # Performance
237///
238/// This processor takes ownership of the item and returns it directly,
239/// with no allocation or clone.
240pub(crate) struct PassThroughProcessor<T> {
241    _phantom: std::marker::PhantomData<T>,
242}
243
244impl<T> Default for PassThroughProcessor<T> {
245    /// Returns a new `PassThroughProcessor`.
246    ///
247    /// Implemented manually (rather than via `#[derive(Default)]`) so that
248    /// `Default` is available for every `T`, not just `T: Default`.
249    /// `PhantomData<T>` does not require `T: Default` to construct, but the
250    /// derive macro adds that bound anyway since it can't see through
251    /// `PhantomData`.
252    fn default() -> Self {
253        Self::new()
254    }
255}
256
257impl<T> ItemProcessor<T, T> for PassThroughProcessor<T> {
258    /// Processes an item by returning it unchanged.
259    ///
260    /// # Parameters
261    /// - `item`: The item to process (consumed by value and returned unchanged)
262    ///
263    /// # Returns
264    /// - `Ok(Some(item))` - Always succeeds and returns the input item
265    fn process(&self, item: T) -> ItemProcessorResult<T> {
266        Ok(Some(item))
267    }
268}
269
270impl<T> PassThroughProcessor<T> {
271    /// Creates a new `PassThroughProcessor`.
272    ///
273    /// # Returns
274    /// A new instance of `PassThroughProcessor` that will pass through items of type `T`.
275    ///
276    /// `const fn` so instances can be created in const contexts (e.g. a
277    /// crate-level `static` for a fixed, non-generic `T`).
278    pub(crate) const fn new() -> Self {
279        Self {
280            _phantom: std::marker::PhantomData,
281        }
282    }
283}
284
285/// A composite processor that chains two processors sequentially using static dispatch.
286///
287/// The output of the first processor becomes the input of the second.
288/// If the first processor filters an item (returns `Ok(None)`), the chain
289/// stops immediately and `Ok(None)` is returned — the second processor is
290/// never called.
291///
292/// Both processors are stored by value — no heap allocation occurs inside the
293/// struct itself. This mirrors the pattern used by standard library iterator
294/// adapters such as [`std::iter::Chain`].
295///
296/// Construct chains using [`CompositeItemProcessorBuilder`] rather than
297/// instantiating this struct directly.
298///
299/// # Type Parameters
300///
301/// - `P1`: The first processor type. Must implement `ItemProcessor<I, M>` for
302///   some input type `I` and intermediate type `M`.
303/// - `P2`: The second processor type. Must implement `ItemProcessor<M, O>` where
304///   `M` is the output type of `P1` and `O` is the final output type.
305/// - `M`: The intermediate type — output of `P1`, input of `P2`. Tracked via
306///   `PhantomData` so it participates in type inference without being stored.
307///
308/// # Examples
309///
310/// ```
311/// use spring_batch_rs::core::item::{ItemProcessor, CompositeItemProcessorBuilder};
312/// use spring_batch_rs::BatchError;
313///
314/// struct DoubleProcessor;
315/// impl ItemProcessor<i32, i32> for DoubleProcessor {
316///     fn process(&self, item: i32) -> Result<Option<i32>, BatchError> {
317///         Ok(Some(item * 2))
318///     }
319/// }
320///
321/// struct ToStringProcessor;
322/// impl ItemProcessor<i32, String> for ToStringProcessor {
323///     fn process(&self, item: i32) -> Result<Option<String>, BatchError> {
324///         Ok(Some(item.to_string()))
325///     }
326/// }
327///
328/// let composite = CompositeItemProcessorBuilder::new(DoubleProcessor)
329///     .link(ToStringProcessor)
330///     .build();
331///
332/// // 21 * 2 = 42, then converted to "42"
333/// assert_eq!(composite.process(21).unwrap(), Some("42".to_string()));
334/// ```
335///
336/// # Errors
337///
338/// Returns [`BatchError`] if any processor in the chain returns an error.
339pub struct CompositeItemProcessor<P1, P2, M> {
340    first: P1,
341    second: P2,
342    /// Tracks the intermediate type `M` (output of `P1`, input of `P2`).
343    /// Uses `fn(M) -> M` to keep the type parameter invariant and avoid
344    /// unintended variance.
345    _marker: std::marker::PhantomData<fn(M) -> M>,
346}
347
348impl<I, M, O, P1, P2> ItemProcessor<I, O> for CompositeItemProcessor<P1, P2, M>
349where
350    P1: ItemProcessor<I, M>,
351    P2: ItemProcessor<M, O>,
352{
353    /// Applies the first processor, then — if the result is `Some` — applies
354    /// the second. Returns `Ok(None)` immediately if the first processor
355    /// filters the item.
356    ///
357    /// # Errors
358    ///
359    /// Returns [`BatchError`] if either processor fails.
360    fn process(&self, item: I) -> ItemProcessorResult<O> {
361        match self.first.process(item)? {
362            Some(intermediate) => self.second.process(intermediate),
363            None => Ok(None),
364        }
365    }
366}
367
368/// Builder for creating a chain of [`ItemProcessor`]s using static dispatch.
369///
370/// Start the chain with [`new`](CompositeItemProcessorBuilder::new), append
371/// processors with [`link`](CompositeItemProcessorBuilder::link), and finalise
372/// with [`build`](CompositeItemProcessorBuilder::build). Each call to `link`
373/// wraps the accumulated chain in a [`CompositeItemProcessor`], changing the
374/// output type. Mismatched types are caught at compile time.
375///
376/// The built chain stores all processors by value — no heap allocations occur
377/// inside the processor itself. The type of the built value encodes the full
378/// chain structure (e.g. `CompositeItemProcessor<P1, CompositeItemProcessor<P2, P3>>`),
379/// similar to how `Iterator` adapters compose in the standard library.
380///
381/// # Type Parameters
382///
383/// - `P`: The accumulated processor type. Starts as the first processor and
384///   is wrapped in [`CompositeItemProcessor`] with each [`link`](CompositeItemProcessorBuilder::link) call.
385///
386/// # Examples
387///
388/// Two processors (`i32 → i32 → String`):
389///
390/// ```
391/// use spring_batch_rs::core::item::{ItemProcessor, CompositeItemProcessorBuilder};
392/// use spring_batch_rs::BatchError;
393///
394/// struct DoubleProcessor;
395/// impl ItemProcessor<i32, i32> for DoubleProcessor {
396///     fn process(&self, item: i32) -> Result<Option<i32>, BatchError> {
397///         Ok(Some(item * 2))
398///     }
399/// }
400///
401/// struct ToStringProcessor;
402/// impl ItemProcessor<i32, String> for ToStringProcessor {
403///     fn process(&self, item: i32) -> Result<Option<String>, BatchError> {
404///         Ok(Some(item.to_string()))
405///     }
406/// }
407///
408/// let composite = CompositeItemProcessorBuilder::new(DoubleProcessor)
409///     .link(ToStringProcessor)
410///     .build();
411///
412/// assert_eq!(composite.process(21).unwrap(), Some("42".to_string()));
413/// ```
414///
415/// Three processors (`i32 → i32 → i32 → String`):
416///
417/// ```
418/// use spring_batch_rs::core::item::{ItemProcessor, CompositeItemProcessorBuilder};
419/// use spring_batch_rs::BatchError;
420///
421/// struct AddOneProcessor;
422/// impl ItemProcessor<i32, i32> for AddOneProcessor {
423///     fn process(&self, item: i32) -> Result<Option<i32>, BatchError> {
424///         Ok(Some(item + 1))
425///     }
426/// }
427///
428/// struct DoubleProcessor;
429/// impl ItemProcessor<i32, i32> for DoubleProcessor {
430///     fn process(&self, item: i32) -> Result<Option<i32>, BatchError> {
431///         Ok(Some(item * 2))
432///     }
433/// }
434///
435/// struct ToStringProcessor;
436/// impl ItemProcessor<i32, String> for ToStringProcessor {
437///     fn process(&self, item: i32) -> Result<Option<String>, BatchError> {
438///         Ok(Some(item.to_string()))
439///     }
440/// }
441///
442/// let composite = CompositeItemProcessorBuilder::new(AddOneProcessor)
443///     .link(DoubleProcessor)
444///     .link(ToStringProcessor)
445///     .build();
446///
447/// // (4 + 1) * 2 = 10 → "10"
448/// assert_eq!(composite.process(4).unwrap(), Some("10".to_string()));
449/// ```
450pub struct CompositeItemProcessorBuilder<P> {
451    processor: P,
452}
453
454impl<P> CompositeItemProcessorBuilder<P> {
455    /// Creates a new builder with the given processor as the first in the chain.
456    ///
457    /// # Parameters
458    ///
459    /// - `first`: The first processor in the chain.
460    ///
461    /// # Examples
462    ///
463    /// ```
464    /// use spring_batch_rs::core::item::{ItemProcessor, CompositeItemProcessorBuilder};
465    /// use spring_batch_rs::BatchError;
466    ///
467    /// struct UppercaseProcessor;
468    /// impl ItemProcessor<String, String> for UppercaseProcessor {
469    ///     fn process(&self, item: String) -> Result<Option<String>, BatchError> {
470    ///         Ok(Some(item.to_uppercase()))
471    ///     }
472    /// }
473    ///
474    /// let builder = CompositeItemProcessorBuilder::new(UppercaseProcessor);
475    /// let composite = builder.build();
476    /// assert_eq!(composite.process("hello".to_string()).unwrap(), Some("HELLO".to_string()));
477    /// ```
478    pub fn new(first: P) -> Self {
479        Self { processor: first }
480    }
481
482    /// Appends a processor to the end of the chain.
483    ///
484    /// Returns a new builder whose accumulated type is
485    /// `CompositeItemProcessor<P, P2>`. The input/output types are verified
486    /// at compile time when the chain is used.
487    ///
488    /// # Type Parameters
489    ///
490    /// - `P2`: The processor type to append.
491    /// - `M`: The intermediate type connecting `P` and `P2`. Inferred by the
492    ///   compiler from the `ItemProcessor` impls on `P` and `P2`.
493    ///
494    /// # Parameters
495    ///
496    /// - `next`: The processor to append to the chain.
497    ///
498    /// # Examples
499    ///
500    /// ```
501    /// use spring_batch_rs::core::item::{ItemProcessor, CompositeItemProcessorBuilder};
502    /// use spring_batch_rs::BatchError;
503    ///
504    /// struct AddOneProcessor;
505    /// impl ItemProcessor<i32, i32> for AddOneProcessor {
506    ///     fn process(&self, item: i32) -> Result<Option<i32>, BatchError> {
507    ///         Ok(Some(item + 1))
508    ///     }
509    /// }
510    ///
511    /// struct ToStringProcessor;
512    /// impl ItemProcessor<i32, String> for ToStringProcessor {
513    ///     fn process(&self, item: i32) -> Result<Option<String>, BatchError> {
514    ///         Ok(Some(item.to_string()))
515    ///     }
516    /// }
517    ///
518    /// let composite = CompositeItemProcessorBuilder::new(AddOneProcessor)
519    ///     .link(ToStringProcessor)
520    ///     .build();
521    ///
522    /// assert_eq!(composite.process(41).unwrap(), Some("42".to_string()));
523    /// ```
524    pub fn link<P2, M>(
525        self,
526        next: P2,
527    ) -> CompositeItemProcessorBuilder<CompositeItemProcessor<P, P2, M>> {
528        CompositeItemProcessorBuilder {
529            processor: CompositeItemProcessor {
530                first: self.processor,
531                second: next,
532                _marker: std::marker::PhantomData,
533            },
534        }
535    }
536
537    /// Builds and returns the composite processor.
538    ///
539    /// Returns the accumulated processor value `P`. When chained via `link`,
540    /// `P` will be a nested `CompositeItemProcessor` such as
541    /// `CompositeItemProcessor<P1, CompositeItemProcessor<P2, P3>>`.
542    ///
543    /// Pass `&composite` to the step builder's `.processor()` method — Rust
544    /// will coerce it to `&dyn ItemProcessor<I, O>` automatically.
545    ///
546    /// # Examples
547    ///
548    /// ```
549    /// use spring_batch_rs::core::item::{ItemProcessor, CompositeItemProcessorBuilder};
550    /// use spring_batch_rs::BatchError;
551    ///
552    /// struct DoubleProcessor;
553    /// impl ItemProcessor<i32, i32> for DoubleProcessor {
554    ///     fn process(&self, item: i32) -> Result<Option<i32>, BatchError> {
555    ///         Ok(Some(item * 2))
556    ///     }
557    /// }
558    ///
559    /// struct AddTenProcessor;
560    /// impl ItemProcessor<i32, i32> for AddTenProcessor {
561    ///     fn process(&self, item: i32) -> Result<Option<i32>, BatchError> {
562    ///         Ok(Some(item + 10))
563    ///     }
564    /// }
565    ///
566    /// let composite = CompositeItemProcessorBuilder::new(DoubleProcessor)
567    ///     .link(AddTenProcessor)
568    ///     .build();
569    ///
570    /// // 5 * 2 = 10, then 10 + 10 = 20
571    /// assert_eq!(composite.process(5).unwrap(), Some(20));
572    /// ```
573    pub fn build(self) -> P {
574        self.processor
575    }
576}
577
578/// A composite writer that fans out the same chunk to two writers sequentially using static dispatch.
579///
580/// Both writers receive identical slices on every `write` call. All four lifecycle
581/// methods (`write`, `flush`, `open`, `close`) are forwarded to `first` then `second`,
582/// short-circuiting on the first `Err`. If `open()` on `first` fails, `second.open()`
583/// is never called — lifecycle management is the step's responsibility.
584///
585/// Both writers are stored by value — no heap allocation occurs inside the struct.
586/// The type encodes the full chain:
587/// `CompositeItemWriter<CompositeItemWriter<W1, W2>, W3>` for three writers.
588///
589/// Prefer constructing instances via [`CompositeItemWriterBuilder`] rather than
590/// direct struct literal syntax.
591///
592/// # Type Parameters
593///
594/// - `W1`: The first writer type. Must implement `ItemWriter<T>`.
595/// - `W2`: The second writer type. Must implement `ItemWriter<T>`.
596///
597/// # Examples
598///
599/// ```
600/// use spring_batch_rs::core::item::{ItemWriter, CompositeItemWriterBuilder};
601/// use std::rc::Rc;
602/// use std::cell::Cell;
603///
604/// struct CountingWriter { count: Rc<Cell<usize>> }
605/// impl CountingWriter {
606///     fn new(count: Rc<Cell<usize>>) -> Self { Self { count } }
607/// }
608/// impl ItemWriter<i32> for CountingWriter {
609///     fn write(&self, items: &[i32]) -> Result<(), spring_batch_rs::BatchError> {
610///         self.count.set(self.count.get() + items.len());
611///         Ok(())
612///     }
613/// }
614///
615/// let c1 = Rc::new(Cell::new(0usize));
616/// let c2 = Rc::new(Cell::new(0usize));
617/// let composite = CompositeItemWriterBuilder::new(CountingWriter::new(c1.clone()))
618///     .link(CountingWriter::new(c2.clone()))
619///     .build();
620/// composite.write(&[1, 2, 3]).unwrap();
621/// assert_eq!(c1.get(), 3);
622/// assert_eq!(c2.get(), 3);
623/// ```
624///
625/// # Errors
626///
627/// Returns [`BatchError`] if any writer in the chain returns an error.
628pub struct CompositeItemWriter<W1, W2> {
629    first: W1,
630    second: W2,
631}
632
633impl<T, W1, W2> ItemWriter<T> for CompositeItemWriter<W1, W2>
634where
635    W1: ItemWriter<T>,
636    W2: ItemWriter<T>,
637{
638    /// Writes `items` to `first`, then to `second`. Short-circuits on the first error.
639    ///
640    /// # Errors
641    ///
642    /// Returns [`BatchError::ItemWriter`] if either writer fails.
643    fn write(&self, items: &[T]) -> ItemWriterResult {
644        self.first.write(items)?;
645        self.second.write(items)
646    }
647
648    /// Flushes both writers regardless of errors. Returns the first error encountered.
649    ///
650    /// Both `first` and `second` are always flushed, even if `first` fails,
651    /// to avoid silently skipping buffered output in the second writer.
652    ///
653    /// # Errors
654    ///
655    /// Returns [`BatchError::ItemWriter`] if either flush fails. If both fail,
656    /// the error from `first` is returned.
657    fn flush(&self) -> ItemWriterResult {
658        let r1 = self.first.flush();
659        let r2 = self.second.flush();
660        r1.and(r2)
661    }
662
663    /// Opens `first`, then `second`. Short-circuits on the first error.
664    ///
665    /// # Errors
666    ///
667    /// Returns [`BatchError::ItemWriter`] if either open fails.
668    fn open(&self) -> ItemWriterResult {
669        self.first.open()?;
670        self.second.open()
671    }
672
673    /// Closes both writers regardless of errors. Returns the first error encountered.
674    ///
675    /// Both `first` and `second` are always closed, even if `first` fails,
676    /// to avoid resource leaks.
677    ///
678    /// # Errors
679    ///
680    /// Returns [`BatchError::ItemWriter`] if either close fails. If both fail,
681    /// the error from `first` is returned.
682    fn close(&self) -> ItemWriterResult {
683        let r1 = self.first.close();
684        let r2 = self.second.close();
685        r1.and(r2)
686    }
687}
688
689/// Builder for creating a fan-out chain of [`ItemWriter`]s using static dispatch.
690///
691/// Start the chain with [`new`](CompositeItemWriterBuilder::new), append writers
692/// with [`add`](CompositeItemWriterBuilder::add), and finalise with
693/// [`build`](CompositeItemWriterBuilder::build). Each call to `add` wraps the
694/// accumulated chain in a [`CompositeItemWriter`]. The built chain stores all
695/// writers by value — no heap allocations occur inside the chain itself.
696///
697/// # Type Parameters
698///
699/// - `W`: The accumulated writer type. Starts as the first writer and is wrapped
700///   in [`CompositeItemWriter`] with each [`add`](CompositeItemWriterBuilder::add) call.
701///
702/// # Examples
703///
704/// Two writers:
705///
706/// ```
707/// use spring_batch_rs::core::item::{ItemWriter, CompositeItemWriterBuilder};
708/// use std::rc::Rc;
709/// use std::cell::Cell;
710///
711/// struct CountingWriter { count: Rc<Cell<usize>> }
712/// impl CountingWriter {
713///     fn new(count: Rc<Cell<usize>>) -> Self { Self { count } }
714/// }
715/// impl ItemWriter<i32> for CountingWriter {
716///     fn write(&self, items: &[i32]) -> Result<(), spring_batch_rs::BatchError> {
717///         self.count.set(self.count.get() + items.len());
718///         Ok(())
719///     }
720/// }
721///
722/// let c1 = Rc::new(Cell::new(0usize));
723/// let c2 = Rc::new(Cell::new(0usize));
724/// let composite = CompositeItemWriterBuilder::new(CountingWriter::new(c1.clone()))
725///     .link(CountingWriter::new(c2.clone()))
726///     .build();
727///
728/// composite.write(&[1, 2, 3]).unwrap();
729/// assert_eq!(c1.get(), 3);
730/// assert_eq!(c2.get(), 3);
731/// ```
732///
733/// Three writers:
734///
735/// ```
736/// use spring_batch_rs::core::item::{ItemWriter, CompositeItemWriterBuilder};
737/// use std::rc::Rc;
738/// use std::cell::Cell;
739///
740/// struct CountingWriter { count: Rc<Cell<usize>> }
741/// impl CountingWriter {
742///     fn new(count: Rc<Cell<usize>>) -> Self { Self { count } }
743/// }
744/// impl ItemWriter<i32> for CountingWriter {
745///     fn write(&self, items: &[i32]) -> Result<(), spring_batch_rs::BatchError> {
746///         self.count.set(self.count.get() + items.len());
747///         Ok(())
748///     }
749/// }
750///
751/// let c1 = Rc::new(Cell::new(0usize));
752/// let c2 = Rc::new(Cell::new(0usize));
753/// let c3 = Rc::new(Cell::new(0usize));
754/// let composite = CompositeItemWriterBuilder::new(CountingWriter::new(c1.clone()))
755///     .link(CountingWriter::new(c2.clone()))
756///     .link(CountingWriter::new(c3.clone()))
757///     .build();
758///
759/// composite.write(&[1, 2]).unwrap();
760/// assert_eq!(c1.get(), 2);
761/// assert_eq!(c2.get(), 2);
762/// assert_eq!(c3.get(), 2);
763/// ```
764pub struct CompositeItemWriterBuilder<W> {
765    writer: W,
766}
767
768impl<W> CompositeItemWriterBuilder<W> {
769    /// Creates a new builder with the given writer as the first delegate.
770    ///
771    /// # Parameters
772    ///
773    /// - `first`: The first writer in the fan-out chain.
774    ///
775    /// # Examples
776    ///
777    /// ```
778    /// use spring_batch_rs::core::item::{ItemWriter, CompositeItemWriterBuilder};
779    ///
780    /// struct CountingWriter { count: std::cell::Cell<usize> }
781    /// impl CountingWriter { fn new() -> Self { Self { count: std::cell::Cell::new(0) } } }
782    /// impl ItemWriter<i32> for CountingWriter {
783    ///     fn write(&self, items: &[i32]) -> Result<(), spring_batch_rs::BatchError> {
784    ///         self.count.set(self.count.get() + items.len());
785    ///         Ok(())
786    ///     }
787    /// }
788    ///
789    /// let writer = CompositeItemWriterBuilder::new(CountingWriter::new()).build();
790    /// writer.write(&[1, 2, 3]).unwrap();
791    /// assert_eq!(writer.count.get(), 3, "writer should receive all items");
792    /// ```
793    pub fn new(first: W) -> Self {
794        Self { writer: first }
795    }
796
797    /// Appends a writer to the fan-out chain.
798    ///
799    /// Returns a new builder whose accumulated type is `CompositeItemWriter<W, W2>`.
800    /// Both writers must implement `ItemWriter<T>` for the same `T` — verified at
801    /// compile time.
802    ///
803    /// # Parameters
804    ///
805    /// - `next`: The writer to link into the chain.
806    ///
807    /// # Examples
808    ///
809    /// ```
810    /// use spring_batch_rs::core::item::{ItemWriter, CompositeItemWriterBuilder};
811    /// use std::rc::Rc;
812    /// use std::cell::Cell;
813    ///
814    /// struct CountingWriter { count: Rc<Cell<usize>> }
815    /// impl CountingWriter {
816    ///     fn new(count: Rc<Cell<usize>>) -> Self { Self { count } }
817    /// }
818    /// impl ItemWriter<i32> for CountingWriter {
819    ///     fn write(&self, items: &[i32]) -> Result<(), spring_batch_rs::BatchError> {
820    ///         self.count.set(self.count.get() + items.len());
821    ///         Ok(())
822    ///     }
823    /// }
824    ///
825    /// let c1 = Rc::new(Cell::new(0usize));
826    /// let c2 = Rc::new(Cell::new(0usize));
827    /// let composite = CompositeItemWriterBuilder::new(CountingWriter::new(c1.clone()))
828    ///     .link(CountingWriter::new(c2.clone()))
829    ///     .build();
830    ///
831    /// composite.write(&[1, 2, 3]).unwrap();
832    /// assert_eq!(c1.get(), 3, "first writer should receive all items");
833    /// assert_eq!(c2.get(), 3, "second writer should receive all items");
834    /// ```
835    #[allow(clippy::should_implement_trait)]
836    pub fn link<W2>(self, next: W2) -> CompositeItemWriterBuilder<CompositeItemWriter<W, W2>> {
837        CompositeItemWriterBuilder {
838            writer: CompositeItemWriter {
839                first: self.writer,
840                second: next,
841            },
842        }
843    }
844
845    /// Builds and returns the composite writer.
846    ///
847    /// Returns the accumulated writer value `W`. When chained via `link`, `W` is a
848    /// nested `CompositeItemWriter` such as
849    /// `CompositeItemWriter<CompositeItemWriter<W1, W2>, W3>`.
850    ///
851    /// Pass `&composite` to the step builder's `.writer()` method.
852    ///
853    /// # Examples
854    ///
855    /// ```
856    /// use spring_batch_rs::core::item::{ItemWriter, CompositeItemWriterBuilder};
857    /// use std::rc::Rc;
858    /// use std::cell::Cell;
859    ///
860    /// struct CountingWriter { count: Rc<Cell<usize>> }
861    /// impl CountingWriter {
862    ///     fn new(count: Rc<Cell<usize>>) -> Self { Self { count } }
863    /// }
864    /// impl ItemWriter<i32> for CountingWriter {
865    ///     fn write(&self, items: &[i32]) -> Result<(), spring_batch_rs::BatchError> {
866    ///         self.count.set(self.count.get() + items.len());
867    ///         Ok(())
868    ///     }
869    /// }
870    ///
871    /// let c1 = Rc::new(Cell::new(0usize));
872    /// let c2 = Rc::new(Cell::new(0usize));
873    /// let composite = CompositeItemWriterBuilder::new(CountingWriter::new(c1.clone()))
874    ///     .link(CountingWriter::new(c2.clone()))
875    ///     .build();
876    ///
877    /// composite.write(&[1, 2, 3]).unwrap();
878    /// assert_eq!(c1.get(), 3);
879    /// assert_eq!(c2.get(), 3);
880    /// ```
881    pub fn build(self) -> W {
882        self.writer
883    }
884}
885
886/// Allows any `Box<P>` where `P: ItemProcessor<I, O>` to be used wherever
887/// `&dyn ItemProcessor<I, O>` is expected — including boxed concrete types
888/// (`Box<MyProcessor>`) and boxed trait objects (`Box<dyn ItemProcessor<I, O>>`).
889///
890/// The `?Sized` bound is what makes this cover trait objects: `dyn Trait` is
891/// unsized, so without `?Sized` the impl would not apply to them.
892impl<I, O, P: ItemProcessor<I, O> + ?Sized> ItemProcessor<I, O> for Box<P> {
893    fn process(&self, item: I) -> ItemProcessorResult<O> {
894        (**self).process(item)
895    }
896}
897
898/// Allows any `Box<W>` where `W: ItemWriter<T>` to be used wherever
899/// `&dyn ItemWriter<T>` is expected — including boxed concrete types
900/// (`Box<MyWriter>`) and boxed trait objects (`Box<dyn ItemWriter<T>>`).
901///
902/// The `?Sized` bound makes this cover trait objects: `dyn Trait` is
903/// unsized, so without `?Sized` the impl would not apply to them.
904impl<T, W: ItemWriter<T> + ?Sized> ItemWriter<T> for Box<W> {
905    fn write(&self, items: &[T]) -> ItemWriterResult {
906        (**self).write(items)
907    }
908    fn flush(&self) -> ItemWriterResult {
909        (**self).flush()
910    }
911    fn open(&self) -> ItemWriterResult {
912        (**self).open()
913    }
914    fn close(&self) -> ItemWriterResult {
915        (**self).close()
916    }
917}
918
919#[cfg(test)]
920mod tests {
921    use super::*;
922
923    #[test]
924    fn should_create_new_pass_through_processor() {
925        let _processor = PassThroughProcessor::<String>::new();
926        // Test that we can create the processor without panicking
927        // Verify it's a zero-sized type (only contains PhantomData)
928        assert_eq!(std::mem::size_of::<PassThroughProcessor<String>>(), 0);
929    }
930
931    #[test]
932    fn should_create_pass_through_processor_with_default() {
933        let _processor = PassThroughProcessor::<i32>::default();
934        // Test that we can create the processor using Default trait
935        // Verify it's a zero-sized type (only contains PhantomData)
936        assert_eq!(std::mem::size_of::<PassThroughProcessor<i32>>(), 0);
937    }
938
939    #[test]
940    fn should_pass_through_string_unchanged() -> Result<(), BatchError> {
941        let processor = PassThroughProcessor::new();
942        let result = processor.process("Hello, World!".to_string())?;
943        assert_eq!(result, Some("Hello, World!".to_string()));
944        Ok(())
945    }
946
947    #[test]
948    fn should_pass_through_integer_unchanged() -> Result<(), BatchError> {
949        let processor = PassThroughProcessor::new();
950        let result = processor.process(42i32)?;
951        assert_eq!(result, Some(42));
952        Ok(())
953    }
954
955    #[test]
956    fn should_pass_through_vector_unchanged() -> Result<(), BatchError> {
957        let processor = PassThroughProcessor::new();
958        let result = processor.process(vec![1, 2, 3, 4, 5])?;
959        assert_eq!(result, Some(vec![1, 2, 3, 4, 5]));
960        Ok(())
961    }
962
963    #[test]
964    fn should_pass_through_custom_struct_unchanged() -> Result<(), BatchError> {
965        #[derive(PartialEq, Debug)]
966        struct TestData {
967            id: u32,
968            name: String,
969            values: Vec<f64>,
970        }
971
972        let processor = PassThroughProcessor::new();
973        let result = processor.process(TestData {
974            id: 123,
975            name: "Test Item".to_string(),
976            values: vec![1.1, 2.2, 3.3],
977        })?;
978        assert_eq!(
979            result,
980            Some(TestData {
981                id: 123,
982                name: "Test Item".to_string(),
983                values: vec![1.1, 2.2, 3.3],
984            })
985        );
986        Ok(())
987    }
988
989    #[test]
990    fn should_pass_through_option_unchanged() -> Result<(), BatchError> {
991        let processor = PassThroughProcessor::new();
992        let result_some = processor.process(Some("test".to_string()))?;
993        assert_eq!(result_some, Some(Some("test".to_string())));
994        let result_none = processor.process(None::<String>)?;
995        assert_eq!(result_none, Some(None::<String>));
996        Ok(())
997    }
998
999    #[test]
1000    fn should_handle_empty_collections() -> Result<(), BatchError> {
1001        let result_vec = PassThroughProcessor::new().process(Vec::<i32>::new())?;
1002        assert_eq!(result_vec, Some(vec![]));
1003        let result_string = PassThroughProcessor::new().process(String::new())?;
1004        assert_eq!(result_string, Some(String::new()));
1005        Ok(())
1006    }
1007
1008    #[test]
1009    fn should_take_ownership_of_input() {
1010        let processor = PassThroughProcessor::new();
1011        let input = "original".to_string();
1012        let result = processor.process(input).unwrap();
1013        assert_eq!(result, Some("original".to_string()));
1014    }
1015
1016    #[test]
1017    fn should_work_with_multiple_processors() -> Result<(), BatchError> {
1018        let processor1 = PassThroughProcessor::<String>::new();
1019        let processor2 = PassThroughProcessor::<String>::new();
1020        let inner = processor1.process("test data".to_string())?.unwrap();
1021        let result = processor2.process(inner)?;
1022        assert_eq!(result, Some("test data".to_string()));
1023        Ok(())
1024    }
1025
1026    #[test]
1027    fn should_handle_large_data_structures() -> Result<(), BatchError> {
1028        let processor = PassThroughProcessor::new();
1029        let large_input: Vec<i32> = (0..10000).collect();
1030        let expected_len = large_input.len();
1031        let result = processor.process(large_input)?;
1032        // PassThroughProcessor always returns Some — unwrap is safe
1033        assert_eq!(result.unwrap().len(), expected_len);
1034        Ok(())
1035    }
1036
1037    #[test]
1038    fn should_use_default_flush_open_close_implementations() {
1039        struct MinimalWriter;
1040        impl ItemWriter<String> for MinimalWriter {
1041            fn write(&self, _: &[String]) -> ItemWriterResult {
1042                Ok(())
1043            }
1044            // flush, open, close use the trait's default implementations
1045        }
1046        let w = MinimalWriter;
1047        assert!(w.flush().is_ok(), "default flush should return Ok");
1048        assert!(w.open().is_ok(), "default open should return Ok");
1049        assert!(w.close().is_ok(), "default close should return Ok");
1050    }
1051
1052    // --- CompositeItemProcessor / CompositeItemProcessorBuilder ---
1053
1054    struct DoubleProcessor;
1055    impl ItemProcessor<i32, i32> for DoubleProcessor {
1056        fn process(&self, item: i32) -> ItemProcessorResult<i32> {
1057            Ok(Some(item * 2))
1058        }
1059    }
1060
1061    struct AddTenProcessor;
1062    impl ItemProcessor<i32, i32> for AddTenProcessor {
1063        fn process(&self, item: i32) -> ItemProcessorResult<i32> {
1064            Ok(Some(item + 10))
1065        }
1066    }
1067
1068    struct ToStringProcessor;
1069    impl ItemProcessor<i32, String> for ToStringProcessor {
1070        fn process(&self, item: i32) -> ItemProcessorResult<String> {
1071            Ok(Some(item.to_string()))
1072        }
1073    }
1074
1075    struct FilterEvenProcessor;
1076    impl ItemProcessor<i32, i32> for FilterEvenProcessor {
1077        fn process(&self, item: i32) -> ItemProcessorResult<i32> {
1078            if item % 2 == 0 {
1079                Ok(Some(item))
1080            } else {
1081                Ok(None) // filter odd numbers
1082            }
1083        }
1084    }
1085
1086    struct FailingProcessor;
1087    impl ItemProcessor<i32, i32> for FailingProcessor {
1088        fn process(&self, _item: i32) -> ItemProcessorResult<i32> {
1089            Err(BatchError::ItemProcessor("forced failure".to_string()))
1090        }
1091    }
1092
1093    #[test]
1094    fn should_chain_two_same_type_processors() -> Result<(), BatchError> {
1095        let composite = CompositeItemProcessorBuilder::new(DoubleProcessor)
1096            .link(AddTenProcessor)
1097            .build();
1098
1099        // 5 * 2 = 10, then 10 + 10 = 20
1100        assert_eq!(
1101            composite.process(5)?,
1102            Some(20),
1103            "5 * 2 + 10 should equal 20"
1104        );
1105        Ok(())
1106    }
1107
1108    #[test]
1109    fn should_chain_two_type_changing_processors() -> Result<(), BatchError> {
1110        let composite = CompositeItemProcessorBuilder::new(DoubleProcessor)
1111            .link(ToStringProcessor)
1112            .build();
1113
1114        // 21 * 2 = 42, then "42"
1115        assert_eq!(composite.process(21)?, Some("42".to_string()));
1116        Ok(())
1117    }
1118
1119    #[test]
1120    fn should_chain_three_processors() -> Result<(), BatchError> {
1121        let composite = CompositeItemProcessorBuilder::new(DoubleProcessor)
1122            .link(AddTenProcessor)
1123            .link(ToStringProcessor)
1124            .build();
1125
1126        // 5 * 2 = 10, then 10 + 10 = 20, then "20"
1127        assert_eq!(composite.process(5)?, Some("20".to_string()));
1128        Ok(())
1129    }
1130
1131    #[test]
1132    fn should_stop_chain_when_first_processor_filters_item() -> Result<(), BatchError> {
1133        let composite = CompositeItemProcessorBuilder::new(FilterEvenProcessor)
1134            .link(ToStringProcessor)
1135            .build();
1136
1137        // 3 is odd → filtered by first processor → second processor never called
1138        assert_eq!(composite.process(3)?, None, "odd number should be filtered");
1139        // 4 is even → passes through → converted to string
1140        assert_eq!(
1141            composite.process(4)?,
1142            Some("4".to_string()),
1143            "even number should pass"
1144        );
1145        Ok(())
1146    }
1147
1148    #[test]
1149    fn should_propagate_error_from_first_processor() {
1150        let composite = CompositeItemProcessorBuilder::new(FailingProcessor)
1151            .link(ToStringProcessor)
1152            .build();
1153
1154        let result = composite.process(1);
1155        assert!(
1156            result.is_err(),
1157            "error from first processor should propagate"
1158        );
1159    }
1160
1161    #[test]
1162    fn should_propagate_error_from_second_processor() {
1163        struct AlwaysFailI32;
1164        impl ItemProcessor<i32, i32> for AlwaysFailI32 {
1165            fn process(&self, _: i32) -> ItemProcessorResult<i32> {
1166                Err(BatchError::ItemProcessor("second failed".to_string()))
1167            }
1168        }
1169
1170        let composite = CompositeItemProcessorBuilder::new(DoubleProcessor)
1171            .link(AlwaysFailI32)
1172            .build();
1173
1174        let result = composite.process(5);
1175        assert!(
1176            result.is_err(),
1177            "error from second processor should propagate"
1178        );
1179    }
1180
1181    #[test]
1182    fn should_use_box_blanket_impl_as_item_processor() -> Result<(), BatchError> {
1183        // build() returns the concrete type; Box::new() it to get a trait object.
1184        // Box<dyn ItemProcessor<I, O>> implements ItemProcessor<I, O> via the ?Sized blanket impl.
1185        let composite = CompositeItemProcessorBuilder::new(DoubleProcessor)
1186            .link(ToStringProcessor)
1187            .build();
1188        let boxed: Box<dyn ItemProcessor<i32, String>> = Box::new(composite);
1189
1190        let result = boxed.process(3)?;
1191        assert_eq!(
1192            result,
1193            Some("6".to_string()),
1194            "boxed trait object should delegate to inner processor"
1195        );
1196        Ok(())
1197    }
1198
1199    #[test]
1200    fn should_use_box_concrete_type_as_item_processor() -> Result<(), BatchError> {
1201        // Box<ConcreteProcessor> also implements ItemProcessor<I, O> via the ?Sized blanket impl
1202        let boxed: Box<DoubleProcessor> = Box::new(DoubleProcessor);
1203
1204        let result = boxed.process(7)?;
1205        assert_eq!(
1206            result,
1207            Some(14),
1208            "boxed concrete processor should delegate to inner processor"
1209        );
1210        Ok(())
1211    }
1212
1213    // --- CompositeItemWriter ---
1214
1215    use std::cell::Cell;
1216
1217    struct RecordingWriter {
1218        write_calls: Cell<usize>,
1219        items_written: Cell<usize>,
1220        open_calls: Cell<usize>,
1221        close_calls: Cell<usize>,
1222        flush_calls: Cell<usize>,
1223        fail_write: bool,
1224        fail_open: bool,
1225        fail_flush: bool,
1226        fail_close: bool,
1227    }
1228
1229    impl RecordingWriter {
1230        fn new() -> Self {
1231            Self {
1232                write_calls: Cell::new(0),
1233                items_written: Cell::new(0),
1234                open_calls: Cell::new(0),
1235                close_calls: Cell::new(0),
1236                flush_calls: Cell::new(0),
1237                fail_write: false,
1238                fail_open: false,
1239                fail_flush: false,
1240                fail_close: false,
1241            }
1242        }
1243        fn failing_write() -> Self {
1244            Self {
1245                fail_write: true,
1246                ..Self::new()
1247            }
1248        }
1249        fn failing_open() -> Self {
1250            Self {
1251                fail_open: true,
1252                ..Self::new()
1253            }
1254        }
1255    }
1256
1257    impl ItemWriter<i32> for RecordingWriter {
1258        fn write(&self, items: &[i32]) -> ItemWriterResult {
1259            if self.fail_write {
1260                return Err(BatchError::ItemWriter("forced write failure".to_string()));
1261            }
1262            self.write_calls.set(self.write_calls.get() + 1);
1263            self.items_written
1264                .set(self.items_written.get() + items.len());
1265            Ok(())
1266        }
1267        fn open(&self) -> ItemWriterResult {
1268            if self.fail_open {
1269                return Err(BatchError::ItemWriter("forced open failure".to_string()));
1270            }
1271            self.open_calls.set(self.open_calls.get() + 1);
1272            Ok(())
1273        }
1274        fn close(&self) -> ItemWriterResult {
1275            if self.fail_close {
1276                return Err(BatchError::ItemWriter("forced close failure".to_string()));
1277            }
1278            self.close_calls.set(self.close_calls.get() + 1);
1279            Ok(())
1280        }
1281        fn flush(&self) -> ItemWriterResult {
1282            if self.fail_flush {
1283                return Err(BatchError::ItemWriter("forced flush failure".to_string()));
1284            }
1285            self.flush_calls.set(self.flush_calls.get() + 1);
1286            Ok(())
1287        }
1288    }
1289
1290    #[test]
1291    fn should_write_to_both_writers() -> Result<(), BatchError> {
1292        let w1 = RecordingWriter::new();
1293        let w2 = RecordingWriter::new();
1294        let composite = CompositeItemWriter {
1295            first: w1,
1296            second: w2,
1297        };
1298        composite.write(&[1, 2, 3])?;
1299        assert_eq!(
1300            composite.first.write_calls.get(),
1301            1,
1302            "first writer should be called"
1303        );
1304        assert_eq!(
1305            composite.first.items_written.get(),
1306            3,
1307            "first writer should receive 3 items"
1308        );
1309        assert_eq!(
1310            composite.second.write_calls.get(),
1311            1,
1312            "second writer should be called"
1313        );
1314        assert_eq!(
1315            composite.second.items_written.get(),
1316            3,
1317            "second writer should receive 3 items"
1318        );
1319        Ok(())
1320    }
1321
1322    #[test]
1323    fn should_open_both_writers_in_order() -> Result<(), BatchError> {
1324        let w1 = RecordingWriter::new();
1325        let w2 = RecordingWriter::new();
1326        let composite = CompositeItemWriter {
1327            first: w1,
1328            second: w2,
1329        };
1330        composite.open()?;
1331        assert_eq!(
1332            composite.first.open_calls.get(),
1333            1,
1334            "first writer should be opened"
1335        );
1336        assert_eq!(
1337            composite.second.open_calls.get(),
1338            1,
1339            "second writer should be opened"
1340        );
1341        Ok(())
1342    }
1343
1344    #[test]
1345    fn should_close_both_writers_in_order() -> Result<(), BatchError> {
1346        let w1 = RecordingWriter::new();
1347        let w2 = RecordingWriter::new();
1348        let composite = CompositeItemWriter {
1349            first: w1,
1350            second: w2,
1351        };
1352        composite.close()?;
1353        assert_eq!(
1354            composite.first.close_calls.get(),
1355            1,
1356            "first writer should be closed"
1357        );
1358        assert_eq!(
1359            composite.second.close_calls.get(),
1360            1,
1361            "second writer should be closed"
1362        );
1363        Ok(())
1364    }
1365
1366    #[test]
1367    fn should_flush_both_writers() -> Result<(), BatchError> {
1368        let w1 = RecordingWriter::new();
1369        let w2 = RecordingWriter::new();
1370        let composite = CompositeItemWriter {
1371            first: w1,
1372            second: w2,
1373        };
1374        composite.flush()?;
1375        assert_eq!(
1376            composite.first.flush_calls.get(),
1377            1,
1378            "first writer should be flushed"
1379        );
1380        assert_eq!(
1381            composite.second.flush_calls.get(),
1382            1,
1383            "second writer should be flushed"
1384        );
1385        Ok(())
1386    }
1387
1388    #[test]
1389    fn should_short_circuit_on_write_error() {
1390        let w1 = RecordingWriter::failing_write();
1391        let w2 = RecordingWriter::new();
1392        let composite = CompositeItemWriter {
1393            first: w1,
1394            second: w2,
1395        };
1396        let result = composite.write(&[1, 2, 3]);
1397        assert!(result.is_err(), "error should propagate");
1398        assert_eq!(
1399            composite.second.write_calls.get(),
1400            0,
1401            "second writer should not be called after first fails"
1402        );
1403    }
1404
1405    #[test]
1406    fn should_short_circuit_on_open_error() {
1407        let w1 = RecordingWriter::failing_open();
1408        let w2 = RecordingWriter::new();
1409        let composite = CompositeItemWriter {
1410            first: w1,
1411            second: w2,
1412        };
1413        let result = composite.open();
1414        assert!(result.is_err(), "error should propagate");
1415        assert_eq!(
1416            composite.second.open_calls.get(),
1417            0,
1418            "second writer should not be opened after first fails"
1419        );
1420    }
1421
1422    #[test]
1423    fn should_flush_both_writers_even_when_first_fails() {
1424        let w1 = RecordingWriter {
1425            fail_flush: true,
1426            ..RecordingWriter::new()
1427        };
1428        let w2 = RecordingWriter::new();
1429        let composite = CompositeItemWriter {
1430            first: w1,
1431            second: w2,
1432        };
1433        let result = composite.flush();
1434        assert!(result.is_err(), "error should propagate");
1435        assert_eq!(
1436            composite.second.flush_calls.get(),
1437            1,
1438            "second writer should still be flushed even when first fails"
1439        );
1440    }
1441
1442    #[test]
1443    fn should_close_both_writers_even_when_first_fails() {
1444        let w1 = RecordingWriter {
1445            fail_close: true,
1446            ..RecordingWriter::new()
1447        };
1448        let w2 = RecordingWriter::new();
1449        let composite = CompositeItemWriter {
1450            first: w1,
1451            second: w2,
1452        };
1453        let result = composite.close();
1454        assert!(result.is_err(), "error should propagate");
1455        assert_eq!(
1456            composite.second.close_calls.get(),
1457            1,
1458            "second writer should still be closed even when first fails"
1459        );
1460    }
1461
1462    #[test]
1463    fn should_chain_two_writers_via_builder() -> Result<(), BatchError> {
1464        let composite = CompositeItemWriterBuilder::new(RecordingWriter::new())
1465            .link(RecordingWriter::new())
1466            .build();
1467        composite.write(&[10, 20])?;
1468        assert_eq!(
1469            composite.first.items_written.get(),
1470            2,
1471            "first writer should receive 2 items"
1472        );
1473        assert_eq!(
1474            composite.second.items_written.get(),
1475            2,
1476            "second writer should receive 2 items"
1477        );
1478        Ok(())
1479    }
1480
1481    #[test]
1482    fn should_chain_three_writers() -> Result<(), BatchError> {
1483        let composite = CompositeItemWriterBuilder::new(RecordingWriter::new())
1484            .link(RecordingWriter::new())
1485            .link(RecordingWriter::new())
1486            .build();
1487        composite.write(&[1, 2, 3, 4])?;
1488        // composite type: CompositeItemWriter<CompositeItemWriter<W1, W2>, W3>
1489        // composite.first is CompositeItemWriter<W1, W2>
1490        // composite.second is W3
1491        assert_eq!(
1492            composite.first.first.items_written.get(),
1493            4,
1494            "writer 1 should receive 4 items"
1495        );
1496        assert_eq!(
1497            composite.first.second.items_written.get(),
1498            4,
1499            "writer 2 should receive 4 items"
1500        );
1501        assert_eq!(
1502            composite.second.items_written.get(),
1503            4,
1504            "writer 3 should receive 4 items"
1505        );
1506        Ok(())
1507    }
1508
1509    #[test]
1510    fn should_use_box_blanket_impl_as_item_writer() -> Result<(), BatchError> {
1511        let composite = CompositeItemWriterBuilder::new(RecordingWriter::new())
1512            .link(RecordingWriter::new())
1513            .build();
1514        let boxed: Box<dyn ItemWriter<i32>> = Box::new(composite);
1515        boxed.write(&[5, 6, 7])?;
1516        // The test verifies that Box<dyn ItemWriter<T>> can be used as an ItemWriter<T>.
1517        // We can't inspect the inner writers through Box<dyn>, so asserting Ok is sufficient.
1518        Ok(())
1519    }
1520
1521    #[test]
1522    fn should_use_box_concrete_writer_as_item_writer() -> Result<(), BatchError> {
1523        let boxed: Box<RecordingWriter> = Box::new(RecordingWriter::new());
1524        boxed.open()?;
1525        boxed.write(&[1, 2])?;
1526        boxed.flush()?;
1527        boxed.close()?;
1528        assert_eq!(
1529            boxed.items_written.get(),
1530            2,
1531            "boxed concrete writer should delegate write"
1532        );
1533        assert_eq!(
1534            boxed.open_calls.get(),
1535            1,
1536            "boxed concrete writer should delegate open"
1537        );
1538        assert_eq!(
1539            boxed.flush_calls.get(),
1540            1,
1541            "boxed concrete writer should delegate flush"
1542        );
1543        assert_eq!(
1544            boxed.close_calls.get(),
1545            1,
1546            "boxed concrete writer should delegate close"
1547        );
1548        Ok(())
1549    }
1550}