// Integrates `element` from Stream. 
// Reads element and produces output
public interface Integrator<A, T, R> {
    boolean integrate(
        A state, // Mutable state object (optional, managed by Initializer)
        T element, // The current element from the upstream
        Downstream<R> downstream // Consumer for remaining downstream elements
    );

}
// Map example
// 1, 3, 2, 5, 1 -> A, C, B, E, A
Integrator<Void, T, R> mapIntegrator =
    (state, element, downstream) -> {
// Adding a mapper function: map(Function<T, R> mapper)
        R newElement = mapper.apply(element);
        downstream.push(newElement);
    };
Gatherer.of(mapIntegrator);

// Filter example
// 1, 3, 2, 5, 1 -> 1, 3, 5, 1
Integrator<Void, T, T> filterIntegrator =
    (state, element, downstream) -> {
        var passOn = 
// Suppying filter rule: filter(Predicate<T> filter)
filter.test(element);
        if (passOn) {
            downstream.push(element);
        }
    };
Gatherer.of(filterIntegrator);

// Creates a private `state` object which is used in `Integrator`
// limit(int numberOfElements)
// 1, 3, 2, 5, 1 -> 1, 3, 2

Supplier<AtomicInteger> initializer =
    AtomicInteger::new;

Integrator<AtomicInteger, T, T> integrator =
    (state, element, downstream) -> {
        int index = state.getAndIncrement();
        if (index < numberOfElements)
            downstream.push(element);
    };
Gatherer.ofSequential(initializer, integrator);

/* For any left over operation.
 Example processing the last element of odd size with window of 2 element
 [1,2,3,4,5] turns into [1,2], [2,3], [5]
 Here 5 is dangling, so 
*/

// fixedGroups(int size)
// 1, 3, 2, 5, 1 -> [1, 3], [2, 5], [1]

Supplier<List<T>> initializer = ArrayList::new;

Integrator<List<T>, T, List<T>> integrator =
    (state, element, downstream) -> {
        state.add(element);
        if (state.size() == size) {
            downstream.push(List.copyOf(state));
            state.clear();
        }
        return true; // Assuming this should be returned by the lambda
    };

BiConsumer<List<T>, Downstream<List<T>>> finisher =
    (state, downstream) -> {
        if (!state.isEmpty()) { // Corrected potential OCR error from 'Istate' to '!state'
            downstream.push(List.copyOf(state));
        }
    };

Gatherer.ofSequential(initializer, integrator, finisher);

// Combine `state` object for parallel streams
// No examples as it requires a lot of boiler plate code to understand

// 1. windowFixed
    List<List<Integer>> windows =    
Stream.of(1,2,3,4,5,6,7,8).gather(Gatherers.windowFixed(3)).toList();
    windows.forEach(System.out::println);
/* 
Input: [1,2,3,4,5,6,7,8]
Output:[1, 2, 3],[4, 5, 6],[7, 8]
*/
// 2. windowSliding
    List<List<Integer>> windows =    
Stream.of(1,2,3,4,5,6,7,8).gather(Gatherers. windowSliding(3)).toList();
    windows.forEach(System.out::println);
/* 
Input: [1,2,3,4,5,6,7,8]
Output:[1, 2, 3],[2, 3, 4],[3, 4, 5],
[4, 5, 6],[5, 6, 7],[6, 7, 8]
*/

synchronized byte[] getData() {
    byte[] buf = ...;
    int nread = socket.getInputStream().read(buf);    // Can block here through pinning
    ...
}