Stream Gatherers使い方メモ

# Stream Gatherers とは
- Stream API で任意の中間操作を実装できるようにする仕組み、API
- JDK 22 から試験的な導入が始まり、来年の2025年3月リリース予定の JDK 24 で正式リリースされる予定
- JDK 24 での JEP は [485](https://openjdk.org/jeps/485)
- Gatherer の読みは、「ぎゃざらー」（多分）

# 環境
```
------------------------------------------------------------
Gradle 8.11.1
------------------------------------------------------------

Build time:    2024-11-20 16:56:46 UTC
Revision:      481cb05a490e0ef9f8620f7873b83bd8a72e7c39

Kotlin:        2.0.20
Groovy:        3.0.22
Ant:           Apache Ant(TM) version 1.10.14 compiled on August 16 2023
Launcher JVM:  23.0.1 (Eclipse Adoptium 23.0.1+11)
Daemon JVM:    C:\pf\java\adoptium\jdk23 (no JDK specified, using current Java home)
OS:            Windows 10 10.0 amd64
```

# 日本語の公式参考情報
- Gatherer の公式の日本語情報は以下が参考になる
    - [Stream Gatherers | Java コアライブラリ開発者ガイド](https://docs.oracle.com/javase/jp/22/core/stream-gatherers-01.html)
    - [Gatherer (Java SE 22 & JDK 22)](https://docs.oracle.com/javase/jp/22/docs/api/java.base/java/util/stream/Gatherer.html)
- いずれも JDK 22 の情報だが、 Stream Gatherers に関する API は 22 から 24 まで一切変更されていないようなので 22 の情報でも多分問題ない
    - [JEP 473 (JDK 23)](https://openjdk.org/jeps/473) と [JEP 485 (JDK 24)](https://openjdk.org/jeps/485) の冒頭の History のところで、それぞれ過去バージョンから変更していないことが明記されている

# Hello World
**実装**

```gradle:build.gradle
plugins {
    id "application"
}

sourceCompatibility = 23
targetCompatibility = 23

application {
    mainClassName = "sandbox.gatherer.Main"
    applicationDefaultJvmArgs = ["--enable-preview"]
}

tasks.withType(JavaCompile).configureEach {
    options.compilerArgs += "--enable-preview"
}
```

- JDK 23 ではまだ試験的な機能なので、 `--enable-preview` を指定する必要がある

```java:Main.java
package sandbox.gatherer;

import java.util.stream.Gatherer;
import java.util.stream.Stream;

public class Main {
    public static void main(String[] args) {
        Gatherer.Integrator<Void, Integer, String> squared =
            Gatherer.Integrator.ofGreedy((_, element, downstream) -> {
                downstream.push(
                    String.format("%s * %s = %s", element, element, (element * element))
                );
                return true;
            });

        Stream.of(1, 2, 3, 4, 5)
                .gather(Gatherer.of(squared))
                .forEach(System.out::println);
    }
}
```

**実行結果**

```text:実行結果
1 * 1 = 1
2 * 2 = 4
3 * 3 = 9
4 * 4 = 16
5 * 5 = 25
```

**説明**

```java:Main.java
        Stream.of(1, 2, 3, 4, 5)
                .gather(Gatherer.of(squared))
                .forEach(System.out::println);
```

- `Stream` に `gather` という中間操作のためのメソッドが追加されている
- このメソッドに中間操作の処理を定義した `Gatherer` を渡すことで、任意の中間操作を実装できる

```java:Main.java
        Gatherer.Integrator<Void, Integer, String> squared =
            Gatherer.Integrator.ofGreedy((_, element, downstream) -> {
                downstream.push(
                    String.format("%s * %s = %s", element, element, (element * element))
                );
                return true;
            });
```

- ここでは、入力から受け取った数値（`element`）を二乗する数式（文字列）に変換して出力（`downstream`）に渡している

# Gatherer


```java:Gatherer(重要なメソッドのみ抽出)
public interface Gatherer<T, A, R> {
    ...
    Integrator<A, T, R> integrator();
    default Supplier<A> initializer() {...}
    default BiConsumer<A, Downstream<? super R>> finisher() {...}
    default BinaryOperator<A> combiner() {...}
    ...
}
```

- 中間操作は [Gatherer インタフェース](https://docs.oracle.com/javase/jp/22/docs/api/java.base/java/util/stream/Gatherer.html)を実装して作成する
- `Gatherer` に定義されているメソッドの内、重要なものは以下の４つになる
    - `integrator`
    - `initializer`
    - `finisher`
    - `combiner`
- 必須のメソッドは `integrator` のみで、残りはデフォルト実装が提供されていて上書きは任意となっている

## Gatherer の生成について
```java:Main.java
        Stream.of(1, 2, 3, 4, 5)
                .gather(Gatherer.of(squared))
                .forEach(System.out::println);
```

- `Gatherer` インタフェースには、 [of メソッド](https://docs.oracle.com/javase/jp/22/docs/api/java.base/java/util/stream/Gatherer.html#of(java.util.stream.Gatherer.Integrator)) などのファクトリメソッドが用意されている
- `Gatherer` インタフェースの抽象メソッドは `integrator` メソッドのみなので、ラムダ式で直接 `Gatherer` のインスタンスを生成することもできる
- このため、わざわざ `of` メソッドを使う意味が無いようにも思える
- しかし、このファクトリメソッドはストリームを **並列処理するときに意味が出てくる**
- 詳細は並列処理のところで説明するので、今はとりあえず「`Gatherer` のインスタンスは `of` メソッドで生成する」ということにしておく


## integrator
```java:Gaherer
public interface Gatherer<T, A, R> {
    ...
    Integrator<A, T, R> integrator();
    ...
}
```

- `integrator` メソッドは、 [Integrator インタフェース](https://docs.oracle.com/javase/jp/22/docs/api/java.base/java/util/stream/Gatherer.Integrator.html) を実装したオブジェクトを返すように実装する
- `Integrator` は中間操作の具体的な処理を提供するためのインタフェースで、以下のような定義になっている

```java:Integrator
interface Integrator<A, T, R> {
    boolean integrate(A state, T element, Downstream<? super R> downstream);
}
```

- [integrate メソッド](https://docs.oracle.com/javase/jp/22/docs/api/java.base/java/util/stream/Gatherer.Integrator.html#integrate(A,T,java.util.stream.Gatherer.Downstream)) を実装する必要があり、この中で具体的な中間操作の処理を実装する
- `integrate` メソッドは、３つの引数を受け取る
    - `state` は中間操作のあいだだけ保持したい情報を記録しておくためのオブジェクトで、**非公開状態オブジェクト (private state object)** と呼ぶ
        - 詳細は `initializer` メソッドを説明するところで解説
    - `element` は中間操作への入力値で、前のストリームから渡ってきた値になる
    - `downstream` は中間操作の結果を後続のストリームに書き出すための API を提供する
        - [Gatherer.Downstream (Java SE 22 & JDK 22)](https://docs.oracle.com/javase/jp/22/docs/api/java.base/java/util/stream/Gatherer.Downstream.html)
- `integrate` メソッドは `boolean` を返す必要がある
    - `true` を返した場合、ストリームの処理はそのまま続行される
    - `false` を返した場合、ストリームの処理はその時点で中断される（これを **短絡 (Short-circuit)** と呼ぶ）

### 後続ストリームへの値の受け渡しを制御する

```java
package sandbox.gatherer;

import java.util.stream.Gatherer;
import java.util.stream.Stream;

public class Main {
    public static void main(String[] args) {
        Gatherer.Integrator<Void, Integer, Integer> filterOddNumber =
            Gatherer.Integrator.ofGreedy((_, element, downstream) -> {
                if (element % 2 != 0) {
                    downstream.push(element);
                }
                return true;
            });

        Stream.of(1, 2, 3, 4, 5)
                .gather(Gatherer.of(filterOddNumber))
                .forEach(System.out::println);
    }
}
```

```text:実行結果
1
3
5
```

- `Downstream` の `push` メソッドを使うことで、後続のストリームに値を受け渡すことができる
- `push` を呼ばなければ、現在の要素（`element`）に対応する値は次のストリームへ渡されないことになる
- 逆に `push` を複数回呼べば、１つの入力要素から複数の出力に変換するようなこともできる

### 短絡を制御する

```java:Main.java
package sandbox.gatherer;

import java.util.stream.Gatherer;
import java.util.stream.Stream;

public class Main {
    public static void main(String[] args) {
        Gatherer.Integrator<Void, Integer, Integer> filter =
            Gatherer.Integrator.of((_, element, downstream) -> {
                if (element == 9) {
                    return false;
                }

                downstream.push(element);
                return true;
            });

        Stream.of(1, 2, 3, 9, 4, 5)
                .gather(Gatherer.of(filter))
                .forEach(System.out::println);
    }
}
```

```text:実行結果
1
2
3
```

- `integrate` メソッドが `false` を返すと、その時点でストリームの処理が中断される
- 上記例では、 `9` が来たら `false` を返して短絡するようにしている

### Integrator インスタンスの生成方法

- ここまでの実装例の中で、 `Integrator` のインスタンスを生成するために以下の二種類のファクトリメソッドを使用した例を記載した
    - [ofGreedy メソッド (Javaodoc)](https://docs.oracle.com/javase/jp/22/docs/api/java.base/java/util/stream/Gatherer.Integrator.html#ofGreedy(java.util.stream.Gatherer.Integrator.Greedy))
    - [of メソッド (Javadoc)](https://docs.oracle.com/javase/jp/22/docs/api/java.base/java/util/stream/Gatherer.Integrator.html#of(java.util.stream.Gatherer.Integrator))

```java:ofGreedyを使ったケース
        Gatherer.Integrator<Void, Integer, Integer> filterOddNumber =
            Gatherer.Integrator.ofGreedy((_, element, downstream) -> {
                if (element % 2 != 0) {
                    downstream.push(element);
                }
                return true;
            });
```

```java:ofを使ったケース
        Gatherer.Integrator<Void, Integer, Integer> filter =
            Gatherer.Integrator.of((_, element, downstream) -> {
                if (element == 9) {
                    return false;
                }

                downstream.push(element);
                return true;
            });
```

- そもそも `Integrator` は関数型インタフェースなので、ファクトリメソッドを使わなくてもラムダ式で直接インスタンスを生成することもできる

```java:ラムダ式で直接生成した場合
        Gatherer.Integrator<Void, Integer, Integer> filter =
            (_, element, downstream) -> {
                if (element == 9) {
                    return false;
                }

                downstream.push(element);
                return true;
            };
```

- 関数型インタフェースを利用した場合も、機能的な差は無い
- これらの生成方法の差は、 **処理の最適化に関係している**
- `ofGreedy` メソッドは、**短絡を行わない `Integrator`** を生成するときに利用することで、処理を最適化できるようになっている
- `ofGreedy` メソッドは [Greedy インタフェース](https://docs.oracle.com/javase/jp/22/docs/api/java.base/java/util/stream/Gatherer.Integrator.Greedy.html) を実装したインスタンスを返却する
- `Greedy` は `Integrator` を継承したマーカーインタフェースで、**短絡を行わない `Integrator`** であることを表している
    - greedy は「貪欲」や「強欲」という意味
    - 途中で短絡せずに全ての要素を処理するという意味合いから来ていると考えられる
- ストリームが処理される際、 `Integrator` の実体が `Greedy` かどうかで処理が調整されるようになっている
- 以下は OpenJDK の実装の一部[^1] を抜粋したものになる

[^1]: [jdk/src/java.base/share/classes/java/util/stream/GathererOp.java at jdk-25+0 · openjdk/jdk](https://github.com/openjdk/jdk/blob/jdk-25%2B0/src/java.base/share/classes/java/util/stream/GathererOp.java#L363)

```java:GathererOp
    ...
    private <CA, CR> CR evaluate(...) {
        ...

        // Optimization
        final boolean greedy = integrator instanceof Integrator.Greedy<A, T, R>;

        ...
        final class Sequential implements Consumer<T>, Gatherer.Downstream<R> {
            ...

            @ForceInline
            Sequential evaluateUsing(Spliterator<T> spliterator) {
                if (greedy)
                    spliterator.forEachRemaining(this);
                else
                    do {
                    } while (proceed && spliterator.tryAdvance(this));

                return this;
            }
    ...
```

- `integrator` が `Greedy` かどうかを判定して、結果を `greedy` 変数に記録している
- そして `greedy` が `true` か `false` かで、 `Spliterator` をイテレートするときの処理を `forEachRemaning` か `tryAdvance` で切り分けている
- `forEachRemaning` は残りの要素を一気にイテレートするのに対して、 `tryAdvance` は次の要素があるか都度確認しながらイテレートする形になる
- `Greedy` は **短絡しない `Integrator`** であることを表しているので、都度確認しながらイテレートする必要がなく `forEachRemaning` で一気にイテレートする形になっている
- つまり、**短絡しないことが保証されている `Integrator` を使うのであれば** `ofGreedy` メソッドでインスタンスを生成した方が、より最適化された実装を選択できることになる


### initializer, finisher

```java:Main.java
package sandbox.gatherer;

import java.util.ArrayList;
import java.util.List;
import java.util.function.BiConsumer;
import java.util.function.BinaryOperator;
import java.util.function.Supplier;
import java.util.stream.Gatherer;
import java.util.stream.Stream;

public class Main {
    public static void main(String[] args) {
        Supplier<List<Integer>> initializer = () -> {
            System.out.println("** initializer **");
            return new ArrayList<>(List.of(0));
        };

        Gatherer.Integrator<List<Integer>, String, Integer> integrator =
            Gatherer.Integrator.ofGreedy((state, element, downstream) -> {
                int totalLength = state.getFirst() + element.length();
                state.set(0, totalLength);

                System.out.printf("""
                == integrator ==
                  element = %s
                  totalLength = %s
                """, element, totalLength);
                return true;
            });

        BinaryOperator<List<Integer>> combiner = Gatherer.defaultCombiner();

        BiConsumer<List<Integer>, Gatherer.Downstream<? super Integer>> finisher =
            (state, downstream) -> {
                int totalLength = state.getFirst();
                System.out.printf("""
                ++ finisher ++
                  totalLength = %s
                """, totalLength);
                downstream.push(totalLength);
            };

        int result = Stream.of("one", "two", "three", "four", "five")
                    .gather(Gatherer.of(initializer, integrator, combiner, finisher))
                    .findFirst()
                    .get();

        System.out.println("result = " + result);
    }
}
```

```text:実行結果
** initializer **
== integrator ==
  element = one
  totalLength = 3
== integrator ==
  element = two
  totalLength = 6
== integrator ==
  element = three
  totalLength = 11
== integrator ==
  element = four
  totalLength = 15
== integrator ==
  element = five
  totalLength = 19
++ finisher ++
  totalLength = 19
result = 19
```

- ここで実装している中間操作は、入力から受け取った文字列の文字数を合計して次に渡す、という実装になっている


```java:initializerの実装
        Supplier<List<Integer>> initializer = () -> {
            System.out.println("** initializer **");
            return new ArrayList<>(List.of(0));
        };
```

- initializer は、ストリームの処理を開始する前に一度だけ呼ばれる
- initializer は、**非公開状態オブジェクト(private state object)** を返却するように実装する
- この非公開状態オブジェクトは、 integrator の処理が呼ばれるときに毎回渡されるので、イテレートのあいだ保持しておきたい情報を格納するために利用できる
- この実装では、文字数の合計を記録するために使用している

```java:integratorの実装
        Gatherer.Integrator<List<Integer>, String, Integer> integrator =
            Gatherer.Integrator.ofGreedy((state, element, downstream) -> {
                int totalLength = state.getFirst() + element.length();
                state.set(0, totalLength);

                System.out.printf("""
                == integrator ==
                  element = %s
                  totalLength = %s
                """, element, totalLength);
                return true;
            });
```

- ストリーム内の全ての要素が処理されると、最後に finisher の処理が呼ばれる
- finisher には非公開状態オブジェクトと `Downstream` が渡されるので、必要に応じて `Downstream` に追加の要素を渡すことができる
- この実装では、非公開状態オブジェクトに溜めていた文字数の合計を `Downstream` に流している

```java:finisherの実装
        BiConsumer<List<Integer>, Gatherer.Downstream<? super Integer>> finisher =
            (state, downstream) -> {
                int totalLength = state.getFirst();
                System.out.printf("""
                ++ finisher ++
                  totalLength = %s
                """, totalLength);
                downstream.push(totalLength);
            };
```

- なお、 combiner についてはデフォルトの `Gatherer.defaultCombiner()` を利用している
- combiner についてはこの後で説明する

## combiner (並列実行)
```java:Main.java
package sandbox.gatherer;

import java.util.ArrayList;
import java.util.List;
import java.util.function.BiConsumer;
import java.util.function.BinaryOperator;
import java.util.function.Supplier;
import java.util.stream.Gatherer;
import java.util.stream.Stream;

public class Main {
    public static void main(String[] args) {
        Supplier<List<Integer>> initializer = () -> {
            print("initializer");
            return new ArrayList<>(List.of(0));
        };

        Gatherer.Integrator<List<Integer>, String, Integer> integrator =
            Gatherer.Integrator.ofGreedy((state, element, downstream) -> {
                int totalLength = state.getFirst() + element.length();
                state.set(0, totalLength);

                print("integrator");
                return true;
            });

        BinaryOperator<List<Integer>> combiner = (state1, state2) -> {
            int totalOfState1 = state1.getFirst();
            int totalOfState2 = state2.getFirst();
            int total = totalOfState1 + totalOfState2;

            print("combiner");
            return new ArrayList<>(List.of(total));
        };

        BiConsumer<List<Integer>, Gatherer.Downstream<? super Integer>> finisher =
            (state, downstream) -> {
                print("finisher");
                downstream.push(state.getFirst());
            };

        int result = Stream.of("one", "two", "three", "four", "five")
                    .gather(Gatherer.of(initializer, integrator, combiner, finisher))
                    .parallel()
                    .findFirst()
                    .get();

        System.out.println("result = " + result);
    }

    private static void print(String message) {
        System.out.println(Thread.currentThread().getName() + ": " + message);
    }
}
```

```text:実行結果
ForkJoinPool.commonPool-worker-2: initializer
main: initializer
ForkJoinPool.commonPool-worker-3: initializer
ForkJoinPool.commonPool-worker-3: integrator
main: integrator
ForkJoinPool.commonPool-worker-2: integrator
ForkJoinPool.commonPool-worker-4: initializer
ForkJoinPool.commonPool-worker-1: initializer
ForkJoinPool.commonPool-worker-4: integrator
ForkJoinPool.commonPool-worker-1: integrator
ForkJoinPool.commonPool-worker-1: combiner
ForkJoinPool.commonPool-worker-4: combiner
ForkJoinPool.commonPool-worker-4: combiner
ForkJoinPool.commonPool-worker-4: combiner
main: finisher
result = 19
```

- 文字数の合計を処理するという実装は initializer, finisher の例と同じにしている
- 異なるのは、 `combiner` を実装している点と、ストリームの実行に `parallel` を加えて並列処理にしている点
- `initializer`, `integrator` がそれぞれ異なるスレッドで並列に実行されていることがわかる
    - `initializer` がスレッドごとに実行されているということは、非公開状態オブジェクトはスレッドごとに生成されていることが分かる
    - したがって、非公開状態オブジェクトはスレッドセーフでないオブジェクトでも問題ない（スレッド間で共有されることはない）


```java:combinerの実装
        BinaryOperator<List<Integer>> combiner = (state1, state2) -> {
            int totalOfState1 = state1.getFirst();
            int totalOfState2 = state2.getFirst();
            int total = totalOfState1 + totalOfState2;

            print("combiner");
            return new ArrayList<>(List.of(total));
        };
```

- `combiner` は、ストリームが並列で実行された際に呼び出される
- 引数として、2つの非公開状態オブジェクトが渡される
- これらは並列実行された各スレッドによって処理された非公開状態オブジェクトであり、必要に応じて1つの非公開状態オブジェクトに統合して返却するように実装する


### of と ofSequential
- `Gatherer` インタフェースには、 `Gatherer` インスタンスを生成するためのファクトリメソッドが大きく2種類用意されている
    - [of メソッド](https://docs.oracle.com/javase/jp/22/docs/api/java.base/java/util/stream/Gatherer.html#of(java.util.stream.Gatherer.Integrator))
    - [ofSequential メソッド](https://docs.oracle.com/javase/jp/22/docs/api/java.base/java/util/stream/Gatherer.html#ofSequential(java.util.stream.Gatherer.Integrator))
- 一見すると、いずれも `integrator` を受け取って `Gatherer` を返しているのは同じで違いが無いように見える
- これらは、ストリームを並列にして実行すると違いがわかる

```java:Main.java
package sandbox.gatherer;

import java.util.stream.Gatherer;
import java.util.stream.Stream;

public class Main {
    public static void main(String[] args) {
        Gatherer.Integrator<Void, Integer, Integer> integrator =
            Gatherer.Integrator.ofGreedy((_, element, downstream) -> {
                downstream.push(element);
                System.out.println(Thread.currentThread().getName() + ": integrator");
                return true;
            });

        Gatherer<Integer, Void, Integer> gatherer
            = Gatherer.of(integrator);
        Gatherer<Integer, Void, Integer> sequentialGatherer
            = Gatherer.ofSequential(integrator);

        System.out.println("[gatherer]");
        Stream.of(1, 2, 3, 4, 5)
            .parallel()
            .gather(gatherer)
            .forEach(_ -> {});

        System.out.println("[sequentialGatherer]");
        Stream.of(1, 2, 3, 4, 5)
            .parallel()
            .gather(sequentialGatherer)
            .forEach(_ -> {});
    }
}
```

- 同じ `Integrator` を元に、 `of` と `ofSequential` の2つで `Gatherer` インスタンスを生成し、同じ内容の `Stream` にそれぞれ適用して並列で起動している

```text:実行結果
[gatherer]
ForkJoinPool.commonPool-worker-4: integrator
main: integrator
ForkJoinPool.commonPool-worker-2: integrator
ForkJoinPool.commonPool-worker-1: integrator
ForkJoinPool.commonPool-worker-3: integrator

[sequentialGatherer]
ForkJoinPool.commonPool-worker-5: integrator
ForkJoinPool.commonPool-worker-5: integrator
ForkJoinPool.commonPool-worker-5: integrator
ForkJoinPool.commonPool-worker-5: integrator
ForkJoinPool.commonPool-worker-5: integrator
```

- `of` で生成した方は、 `Integrator` の処理が異なるスレッドで実行されている
- 一方で、 `ofSequential` で生成した `Integrator` は、全て同じスレッドで実行されている
- `ofSequential` で生成された `Gatherer` は**シーケンシャルな Gatherer**と判断され、ストリームを並列実行しても単一のスレッドでしか実行されないようになっている
- したがって、マルチスレッドで実行されても問題ない `Gatherer` を作りたい場合は `of` メソッドでインスタンスを生成しておく必要がある
    - 間違えて `ofSequential` で生成すると、実装上は並列実行になっていても実際には単一スレッドで実行されるようなことになりかねない
- 逆にマルチスレッドでの実行は想定していない `Gatherer` を用意する場合は、 `ofSequential` メソッドでインスタンスを生成しておかないといけない

#### ラムダ式で Gatherer を生成するとシーケンシャルな Gatherer になる

- ところで `Gatherer` インタフェースは抽象メソッドが `integrator` の1つだけなので、関数型インタフェースとなっている
- したがって、 `Gatherer` のインスタンスはラムダ式で生成することでもきる

```java:Gathererをラムダ式で生成した場合
package sandbox.gatherer;

import java.util.stream.Gatherer;
import java.util.stream.Stream;

public class Main {
    public static void main(String[] args) {
        Gatherer.Integrator<Void, Integer, Integer> integrator =
            Gatherer.Integrator.ofGreedy((_, element, downstream) -> {
                downstream.push(element);
                System.out.println(Thread.currentThread().getName() + ": integrator");
                return true;
            });

        Gatherer<Integer, Void, Integer> gatherer = () -> integrator;

        System.out.println("[gatherer]");
        Stream.of(1, 2, 3, 4, 5)
            .parallel()
            .gather(gatherer)
            .forEach(_ -> {});
    }
}
```

- `of` および `ofSequential` を使わずに、ラムダ式だけで `Gatherer` インスタンスを生成している
- このラムダ式で生成した `Gatherer` を使ってストリームを並列で起動している

```text:実行結果
ForkJoinPool.commonPool-worker-2: integrator
ForkJoinPool.commonPool-worker-2: integrator
ForkJoinPool.commonPool-worker-2: integrator
ForkJoinPool.commonPool-worker-2: integrator
ForkJoinPool.commonPool-worker-2: integrator
```

- ラムダ式だけで生成した `Gatherer` を使うと、ストリームを並列実行しても単一スレッドで実行されてしまう
- つまり、シーケンシャルな Gatherer となっている
- `Gatherer` の Javadoc には以下のように書かれている

> コンビナがdefaultCombiner()であるGatherersは、**順次しか評価できません**。 
> https://docs.oracle.com/javase/jp/22/docs/api/java.base/java/util/stream/Gatherer.html

- `Gatherer` の `combiner` メソッドはデフォルト実装が提供されており `defaultCombiner()` の結果を返すように実装されている
- したがって、ラムダ式で生成した `Gatherer` の combiner は `defaultCombiner()` と同じモノを返すことになり、順次しか評価できない＝シーケンシャルな Gatherer として扱われることになる
- このため、並列処理されると困るような実装でない限り、 `Gatherer` のインスタンスは `of` メソッドで生成しておくのが無難と考えられる
    - ラムダ式で生成していると並列で実行しても単一スレッドでしか動かず、パフォーマンスがよくならない可能性がある

# 組み込みの Gatherer
[Gatherers](https://docs.oracle.com/javase/jp/22/docs/api/java.base/java/util/stream/Gatherers.html) というクラスに、組み込みの `Gatherer` を返すメソッドがいくつか用意されている。

## fold
```java:Main.java
package sandbox.gatherer;

import java.util.stream.Gatherer;
import java.util.stream.Gatherers;
import java.util.stream.Stream;

public class Main {
    public static void main(String[] args) {
        Gatherer<Integer, ?, String> fold =
            Gatherers.fold(() -> "@", (state, element) -> state + element);

        String result = Stream.of(1, 2, 3, 4, 5)
                .gather(fold)
                .findFirst()
                .get();

        System.out.println("result = " + result);
    }
}
```

```text:実行結果
result = @12345
```

- [fold](https://docs.oracle.com/javase/jp/22/docs/api/java.base/java/util/stream/Gatherers.html#fold(java.util.function.Supplier,java.util.function.BiFunction)) メソッドを使用すると、ストリームの各要素を単一の結果にまとめる（畳みこむ）処理を実装できる
- fold は「畳みこむ」という意味
- `fold` メソッドの第一引数には、畳み込み処理の初期値を返す関数を渡す
- 第二引数は、畳み込み後の値と現在の要素が渡されるので、畳み込み後の値を返すように実装した関数を渡す

## mapConcurrent
```java:Main.java
package sandbox.gatherer;

import java.util.stream.Gatherer;
import java.util.stream.Gatherers;
import java.util.stream.Stream;

public class Main {
    public static void main(String[] args) {
        Gatherer<Integer, ?, String> mapConcurrent
            = Gatherers.mapConcurrent(3, element -> {
                Thread thread = Thread.currentThread();
                System.out.printf("""
                id=%s, isVirtual=%s
                """, thread.threadId(), thread.isVirtual());
                return "[" + element + "]";
            });

        Stream.of(1, 2, 3, 4, 5)
                .gather(mapConcurrent)
                .forEach(System.out::println);
    }
}
```

```text:実行結果
id=30, isVirtual=true
id=32, isVirtual=true
id=34, isVirtual=true
[1]
[2]
[3]
id=38, isVirtual=true
id=37, isVirtual=true
[4]
[5]
```

- [mapConcurrent メソッド](https://docs.oracle.com/javase/jp/22/docs/api/java.base/java/util/stream/Gatherers.html#mapConcurrent(int,java.util.function.Function)) を使用すると、変換を仮装スレッドで行う `Gatherer` を生成できる
- 第一引数には同時実行数を渡し、第二引数には変換処理を渡す
- ストリームの順序は保存（維持）される

## scan
```java:Main.java
package sandbox.gatherer;

import java.util.stream.Gatherer;
import java.util.stream.Gatherers;
import java.util.stream.Stream;

public class Main {
    public static void main(String[] args) {
        Gatherer<Integer, ?, String> fold =
                Gatherers.scan(() -> "@", (state, element) -> state + element);

        Stream.of(1, 2, 3, 4, 5)
                .gather(fold)
                .forEach(System.out::println);
    }
}
```

```text:実行結果
@1
@12
@123
@1234
@12345
```

- `fold` は結果を1つに畳み込んでいたのに対して、 [scan メソッド](https://docs.oracle.com/javase/jp/22/docs/api/java.base/java/util/stream/Gatherers.html#scan(java.util.function.Supplier,java.util.function.BiFunction)) は結果を蓄積しつつ、各イテレートの結果はそのまま次のストリームに渡される

## windowFixed
```java:Main.java
package sandbox.gatherer;

import java.util.List;
import java.util.stream.Gatherer;
import java.util.stream.Gatherers;
import java.util.stream.Stream;

public class Main {
    public static void main(String[] args) {
        Gatherer<Integer, ?, List<Integer>> windowFixed
            = Gatherers.windowFixed(2);

        List<List<Integer>> result = Stream.of(1, 2, 3, 4, 5)
                .gather(windowFixed)
                .toList();

        System.out.println("result = " + result);
    }
}
```

```text:実行結果
result = [[1, 2], [3, 4], [5]]
```

- [windowFixed メソッド](https://docs.oracle.com/javase/jp/22/docs/api/java.base/java/util/stream/Gatherers.html#windowFixed(int)) を使うと、各要素を指定された要素数の `List` で切り分けることができる

## windowSliding
```java:Main.java
package sandbox.gatherer;

import java.util.List;
import java.util.stream.Gatherer;
import java.util.stream.Gatherers;
import java.util.stream.Stream;

public class Main {
    public static void main(String[] args) {
        Gatherer<Integer, ?, List<Integer>> windowFixed
            = Gatherers.windowSliding(3);

        List<List<Integer>> result = Stream.of(1, 2, 3, 4, 5)
                .gather(windowFixed)
                .toList();

        System.out.println("result = " + result);
    }
}
```

```text:実行結果
result = [[1, 2, 3], [2, 3, 4], [3, 4, 5]]
```

- [windowSliding メソッド](https://docs.oracle.com/javase/jp/22/docs/api/java.base/java/util/stream/Gatherers.html#windowSliding(int)) を使用すると、各要素を指定された要素数の `List` で切り分けることができる
- このとき、切り分けられた各 `List` の中身は、1つずつスライドするように格納される

# Gatherer の連結

```java:Main.java
package sandbox.gatherer;

import java.util.stream.Gatherer;
import java.util.stream.Stream;

public class Main {
    public static void main(String[] args) {
        Gatherer<String, Void, String> toUpperCase
            = Gatherer.of((_, element, downstream) -> {
                downstream.push(element.toUpperCase());
                return true;
            });
        Gatherer<String, Void, String> wrapWithParentheses
            = Gatherer.of((_, element, downstream) -> {
                downstream.push("(" + element + ")");
                return true;
            });

        Gatherer<String, ?, String> gatherer = toUpperCase.andThen(wrapWithParentheses);

        Stream.of("one", "two", "three")
                .gather(gatherer)
                .forEach(System.out::println);
    }
}
```

- 入力値を大文字に変換する `Gatherer` と括弧で囲う `Gatherer` を用意して、 `andThen` で連結してから利用している

```text:実行結果
(ONE)
(TWO)
(THREE)
```

- `Gatherer` は `andThen` であらかじめ連結しておくことができる


# 参考
- [JEP 485: Stream Gatherers](https://openjdk.org/jeps/485)
- [Stream Gatherers | Java SE 22 Core Libraries](https://docs.oracle.com/javase/jp/22/core/stream-gatherers-01.html#GUID-A8BF409A-A4F5-4A1E-98D8-62F2829A1399)