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}