Sequential Aggregate Reader

I have written once a reader that I want to share (I can create a PR if you like it):

The idea of this reader is to read an ordered set of data and to group items based on a condition. Exemple :

Given :

Based on the fk column, the reader must give me :

Aggregate : (1, {data-1, data-2})
Aggregate : (2, {data-1, data-2})
Aggregate : (3, {data-1})

Implementation (Not thread safe) :

//Implemented by the model to tell us how to aggregate, which line match the next one
public interface AggregateMatcher<T> {
    boolean match(T t);
}

//The aggregate model
public class Aggregate<T> implements Iterable<T> {

    private final List<T> items = new ArrayList<>();

    public static <E> Aggregate<E> of(E item) {
        return new Aggregate<>(item);
    }

    @SafeVarargs
    public static <E> Aggregate<E> of(E... item) {
        Aggregate<E> aggregate = new Aggregate<>();
        Arrays.stream(item).forEach(aggregate::add);
        return aggregate;
    }
    
    private Aggregate() {
    }

    private Aggregate(T first) {
        add(first);
    }

    public Aggregate<T> add(T item) {
        items.add(item);
        return this;
    }

    public boolean isEmpty() {
        return items.isEmpty();
    }

    @Override
    public Iterator<T> iterator() {
        return items.iterator();
    }

    public Stream<T> stream() {
        return items.stream();
    }

    public int size() {
        return items.size();
    }
}

public class AggregateReader<T extends AggregateMatcher<T>> implements ItemReader<Aggregate<T>>, ItemStream {

    private final PeekableItemReader<T> delegate;

    public AggregateReader(PeekableItemReader<T> delegate) {
        this.delegate = delegate;
    }

    @Nullable
    @Override
    public Aggregate<T> read() throws Exception {

        T item = delegate.read();
        if (item == null) {
            return null;
        }

        Aggregate<T> aggregate = Aggregate.of(item);
        T candidate;
        while ((candidate = delegate.peek()) != null && candidate.match(item)) {
            aggregate.add(delegate.read());
        }

        return aggregate;
    }

    @Override
    public void open(ExecutionContext executionContext) throws ItemStreamException {
        if (delegate instanceof ItemStream) {
            ((ItemStream) delegate).open(executionContext);
        }
    }

    @Override
    public void update(ExecutionContext executionContext) throws ItemStreamException {
        if (delegate instanceof ItemStream) {
            ((ItemStream) delegate).update(executionContext);
        }
    }

    @Override
    public void close() throws ItemStreamException {
        if (delegate instanceof ItemStream) {
            ((ItemStream) delegate).close();
        }
    }
}

And finally the test class:

class AggregateReaderTest {

    @Test
    void should_read_multiple_aggregates_when_multiple_ides_found() throws Exception {

        AggregateReader<SomeData> reader = aggregateReader(List.of(
                SomeData.of(1, "1"),
                SomeData.of(1, "2"),
                SomeData.of(2, "1"),
                SomeData.of(2, "2"),
                SomeData.of(2, "3"),
                SomeData.of(2, "4"),
                SomeData.of(3, "1"),
                SomeData.of(3, "2")
        ));

        assertThat(reader.read())
                .containsExactly(
                        SomeData.of(1, "1"),
                        SomeData.of(1, "2")
                );
        assertThat(reader.read())
                .containsExactly(
                        SomeData.of(2, "1"),
                        SomeData.of(2, "2"),
                        SomeData.of(2, "3"),
                        SomeData.of(2, "4")
                );
        assertThat(reader.read())
                .containsExactly(
                        SomeData.of(3, "1"),
                        SomeData.of(3, "2")
                );

        assertThat(reader.read()).isNull();
    }

    @Test
    void should_read_one_aggregate_of_one_item_when_data_size_isOne() throws Exception {

        AggregateReader<SomeData> reader = aggregateReader(List.of(SomeData.of(1, "1")));
        assertThat(reader.read())
                .containsExactly(SomeData.of(1, "1"));
        assertThat(reader.read()).isNull();
    }

    @Test
    void should_read_one_aggregate_when_same_id_on_all_data() throws Exception {

        AggregateReader<SomeData> reader = aggregateReader(List.of(SomeData.of(1, "1"), SomeData.of(1, "2")));
        assertThat(reader.read())
                .containsExactly(SomeData.of(1, "1"), SomeData.of(1, "2"));
        assertThat(reader.read()).isNull();
    }

    @Test
    void should_read_aggregate_of_one_item_when_not_same_id_provided() throws Exception {

        AggregateReader<SomeData> reader = aggregateReader(List.of(SomeData.of(1, "1"), SomeData.of(2, "2"), SomeData.of(3, "3")));
        assertThat(reader.read())
                .containsExactly(SomeData.of(1, "1"));
        assertThat(reader.read())
                .containsExactly(SomeData.of(2, "2"));
        assertThat(reader.read())
                .containsExactly(SomeData.of(3, "3"));
        assertThat(reader.read()).isNull();
    }

    @Test
    void should_read_null_when_empty_data() throws Exception {

        AggregateReader<SomeData> reader = aggregateReader(List.of());
        assertThat(reader.read()).isNull();
    }

    private static AggregateReader<SomeData> aggregateReader(List<SomeData> someData) {
        SingleItemPeekableItemReader<SomeData> peekableItemReader = new SingleItemPeekableItemReader<>();
        peekableItemReader.setDelegate(new ListItemReader<>(someData));
        return new AggregateReader<>(peekableItemReader);
    }

    @Value(staticConstructor = "of")
    private static class SomeData implements AggregateMatcher<SomeData> {
        int id;
        String data;

        @Override
        public boolean match(SomeData someData) {
            return id == someData.id;
        }
    }
}

Start by reviewing the proposed AggregateMatcher, Aggregate, AggregateReader, and AggregateReaderTest entry points in the issue, along with the ItemReader and ItemStream contracts they use. Compare the behavior against existing Spring Batch reader patterns and determine whether the proposed API belongs in the project. Done means an agreed design, integrated implementation, and passing coverage for the listed aggregate cases.