// 종단 연산: Collector로 자유롭게 커스텀 가능
stream.collect(Collectors.groupingBy(...));
stream.collect(myCustomCollector());

// 중간 연산: 정해진 것만 사용 가능
stream.filter(...)
      .map(...)
      .sorted(...)  // 여기에 커스텀 중간 연산을 끼워넣을 방법이 없었음
      .collect(...);

Collector : collect() = Gatherer : gather()
(종단 연산 커스텀)       (중간 연산 커스텀)

stream.gather(myCustomGatherer)   // 커스텀 중간 연산
      .filter(...)
      .collect(...);

public interface Gatherer<T, A, R> {
    // T: 입력 타입, A: 중간 상태 타입, R: 출력 타입

    Supplier<A> initializer();          // 1. 초기 상태 생성
    Integrator<A, T, R> integrator();   // 2. 각 요소 처리
    BinaryOperator<A> combiner();       // 3. 병렬 스트림 결합
    BiConsumer<A, Downstream<R>> finisher(); // 4. 마무리 처리
}

@FunctionalInterface
interface Integrator<A, T, R> {
    boolean integrate(A state, T element, Downstream<R> downstream);
    // state: 현재 상태
    // element: 현재 스트림 요소
    // downstream: 결과를 다음 단계로 보내는 인터페이스
    // 반환값: true면 계속, false면 스트림 조기 종료
}

// 3개씩 묶기
List<List<Integer>> windows = Stream.of(1, 2, 3, 4, 5, 6, 7)
    .gather(Gatherers.windowFixed(3))
    .toList();
// [[1, 2, 3], [4, 5, 6], [7]]

// 크기 3의 슬라이딩 윈도우
List<List<Integer>> windows = Stream.of(1, 2, 3, 4, 5)
    .gather(Gatherers.windowSliding(3))
    .toList();
// [[1, 2, 3], [2, 3, 4], [3, 4, 5]]

// 누적 합계를 중간 결과로 내보냄
List<Integer> runningSums = Stream.of(1, 2, 3, 4, 5)
    .gather(Gatherers.fold(() -> 0, Integer::sum))
    .toList();
// [15] — fold는 최종 결과만 하나 내보냄

// 누적 합계를 매 단계마다 출력
List<Integer> runningSums = Stream.of(1, 2, 3, 4, 5)
    .gather(Gatherers.scan(() -> 0, Integer::sum))
    .toList();
// [1, 3, 6, 10, 15]

// 최대 5개를 동시에 처리하되 순서 유지
List<String> results = Stream.of("url1", "url2", "url3", "url4", "url5")
    .gather(Gatherers.mapConcurrent(5, url -> fetchData(url)))
    .toList();

public static <T, K> Gatherer<T, ?, T> distinctBy(Function<T, K> keyExtractor) {
    return Gatherer.ofSequential(
        // initializer: 이미 본 키를 저장할 Set
        HashSet<K>::new,

        // integrator: 새 키면 통과, 이미 본 키면 무시
        (seen, element, downstream) -> {
            K key = keyExtractor.apply(element);
            if (seen.add(key)) {
                downstream.push(element);
            }
            return true;  // 계속 처리
        }
    );
}

// 사용
record Employee(String dept, String name) {}

var employees = List.of(
    new Employee("개발", "Alice"),
    new Employee("개발", "Bob"),
    new Employee("인사", "Charlie"),
    new Employee("인사", "Dave")
);

// 부서별 첫 번째 직원만
var firstPerDept = employees.stream()
    .gather(distinctBy(Employee::dept))
    .toList();
// [Employee[dept=개발, name=Alice], Employee[dept=인사, name=Charlie]]

public static <T> Gatherer<T, ?, T> takeWhileChanged(
        Function<T, ?> keyExtractor) {

    // 상태: 이전 키를 추적
    class State {
        Object previousKey = new Object();  // 센티넬 값 (아직 시작 안 함)
        boolean started = false;
    }

    return Gatherer.ofSequential(
        State::new,
        (state, element, downstream) -> {
            Object currentKey = keyExtractor.apply(element);
            if (!state.started) {
                state.started = true;
                state.previousKey = currentKey;
                return downstream.push(element);
            }
            if (Objects.equals(state.previousKey, currentKey)) {
                return downstream.push(element);
            }
            return false;  // 키가 변했으면 스트림 종료
        }
    );
}

// 같은 카테고리의 연속된 항목만 가져오기
var items = List.of("A:1", "A:2", "A:3", "B:4", "A:5");
var result = items.stream()
    .gather(takeWhileChanged(s -> s.split(":")[0]))
    .toList();
// [A:1, A:2, A:3]  — B에서 키가 변하므로 종료

public record Indexed<T>(int index, T value) {}

public static <T> Gatherer<T, ?, Indexed<T>> zipWithIndex() {
    return Gatherer.ofSequential(
        // initializer: 카운터
        () -> new int[]{0},

        // integrator: 인덱스를 붙여서 전달
        (counter, element, downstream) -> {
            downstream.push(new Indexed<>(counter[0]++, element));
            return true;
        }
    );
}

// 사용
List.of("Apple", "Banana", "Cherry").stream()
    .gather(zipWithIndex())
    .forEach(System.out::println);
// Indexed[index=0, value=Apple]
// Indexed[index=1, value=Banana]
// Indexed[index=2, value=Cherry]

public List<Response> batchProcess(List<Request> requests) {
    return requests.stream()
        .gather(Gatherers.windowFixed(100))        // 100개씩 묶기
        .gather(Gatherers.mapConcurrent(3,         // 최대 3 배치 동시 호출
            batch -> apiClient.bulkSend(batch)))
        .flatMap(List::stream)
        .toList();
}

public static Gatherer<Double, ?, Double> movingAverage(int windowSize) {
    return Gatherer.<Double, LinkedList<Double>, Double>ofSequential(
        LinkedList::new,
        (window, value, downstream) -> {
            window.addLast(value);
            if (window.size() > windowSize) {
                window.removeFirst();
            }
            if (window.size() == windowSize) {
                double avg = window.stream()
                    .mapToDouble(Double::doubleValue)
                    .average()
                    .orElse(0);
                downstream.push(avg);
            }
            return true;
        }
    );
}

// 3일 이동 평균
Stream.of(10.0, 20.0, 30.0, 40.0, 50.0)
    .gather(movingAverage(3))
    .forEach(System.out::println);
// 20.0 (10+20+30)/3
// 30.0 (20+30+40)/3
// 40.0 (30+40+50)/3

javac --enable-preview --source 25 MyApp.java
java --enable-preview MyApp

// 병렬 지원 Gatherer
Gatherer.of(
    initializer,
    integrator,
    combiner,     // 병렬 처리 시 상태 병합 로직
    finisher
);

stream.gather(windowFixed(5))
      .gather(mapConcurrent(3, this::process))
      .gather(distinctBy(Result::id))
      .toList();