java.lang.Object

cloud.opencode.base.collections.OpenGatherers

public final class OpenGatherers extends Object

OpenGatherers - JDK 25 Stream Gatherers Utilities OpenGatherers - JDK 25 流收集器工具

Provides custom Gatherers for advanced stream processing operations using JDK 25's Gatherer API (JEP 485).

使用 JDK 25 的 Gatherer API (JEP 485) 提供自定义收集器用于高级流处理操作。

Features | 主要功能:

Windowing operations (fixed, sliding, session) - 窗口操作
Batching and chunking - 批处理和分块
Distinct by key - 按键去重
Scan and fold operations - 扫描和折叠操作
Filtering with state - 带状态的过滤
Map with previous element - 带前一个元素的映射

Usage Examples | 使用示例:

// Fixed window - 固定窗口
Stream<List<Integer>> windows = stream.gather(OpenGatherers.windowFixed(3));

// Sliding window - 滑动窗口
Stream<List<Integer>> sliding = stream.gather(OpenGatherers.windowSliding(3));

// Distinct by key - 按键去重
Stream<User> unique = users.gather(OpenGatherers.distinctBy(User::getEmail));

// Take while with index - 带索引的takeWhile
Stream<T> limited = stream.gather(OpenGatherers.takeWhileIndexed((i, e) -> i < 10));

// Scan (running accumulation) - 扫描（运行累积）
Stream<Integer> sums = numbers.gather(OpenGatherers.scan(0, Integer::sum));

Performance | 性能特性:

Uses JDK 25 Gatherer API - 使用 JDK 25 Gatherer API
Supports parallel streams where applicable - 在适用时支持并行流
Lazy evaluation - 延迟求值

Security | 安全性:

Thread-safe: Depends on Gatherer - 线程安全: 取决于收集器
Null-safe: Yes - 空值安全: 是

Since:

JDK 25, opencode-base-collections V1.0.0

Author:

Leon Soo www.LeonSoo.com

See Also:

Nested Class Summary

Nested Classes

Modifier and Type

Class

Description

static final record

OpenGatherers.IndexedElement<T>

Indexed element record.
Method Summary

Modifier and Type

Method

Description

static <T> Gatherer<T,?,List<T>>

batch(int batchSize)

Creates a batch gatherer.

static <T,R> Gatherer<T,?,R>

batchProcess(int batchSize, Function<List<T>,R> processor)

Creates a batch gatherer with processing function.

static <T> Gatherer<T,?,T>

changed()

Creates a changed gatherer (emit only when value changes).

static <T,K> Gatherer<T,?,T>

changedBy(Function<? super T, ? extends K> keyExtractor)

Creates a changed-by gatherer (emit when key changes).

static <T,K> Gatherer<T,?,T>

distinctBy(Function<? super T, ? extends K> keyExtractor)

Creates a distinct-by-key gatherer.

static <T,K> Gatherer<T,?,T>

distinctBy(Function<? super T, ? extends K> keyMapper, BiPredicate<K,K> equals, ToIntFunction<K> hash)

Creates a distinct-by gatherer with custom equality.

static <T> Gatherer<T,?,T>

dropLast(int n)

Creates a drop-last gatherer.

static <T> Gatherer<T,?,T>

dropWhileIndexed(BiPredicate<Long, ? super T> predicate)

Creates a drop-while-indexed gatherer.

static <T> Gatherer<T,?,T>

filterIndexed(BiPredicate<Long, ? super T> predicate)

Creates a filter-indexed gatherer.

static <T,R> Gatherer<T,?,R>

fold(R initial, BiFunction<R, ? super T, R> accumulator)

Creates a fold gatherer (produces single result).

static <T,K> Gatherer<T,?,List<T>>

groupRuns(Function<? super T, ? extends K> keyExtractor)

Creates a group-runs gatherer.

static <T> Gatherer<T, ?, OpenGatherers.IndexedElement<T>>

indexed()

Creates an indexed wrapper gatherer.

static <T> Gatherer<T,?,T>

interleave(Iterator<? extends T> other)

Creates a gatherer that interleaves elements from the source stream with elements from the provided iterator, alternating between them.

static <T> Gatherer<T,?,T>

intersperse(T separator)

Creates an interleave gatherer with separators.

static <T,R> Gatherer<T,?,R>

mapIndexed(BiFunction<Long, ? super T, ? extends R> mapper)

Creates a map-indexed gatherer.

static <T,R> Gatherer<T,?,R>

mapWithPrevious(BiFunction<? super T, ? super T, ? extends R> mapper)

Creates a map-with-previous gatherer.

static <T> Gatherer<T,?,T>

peekWithIndex(BiConsumer<Long, ? super T> action)

Creates a peek gatherer for debugging.

static <T,A,R> Gatherer<T,?,R>

scan(Supplier<A> initial, BiFunction<A, ? super T, A> accumulator, Function<A,R> finisher)

Creates a scan gatherer with initial value and finisher.

static <T,R> Gatherer<T,?,R>

scan(R initial, BiFunction<R, ? super T, R> accumulator)

Creates a scan (running fold) gatherer.

static <T> Gatherer<T,?,T>

takeLast(int n)

Creates a take-last gatherer.

static <T> Gatherer<T,?,T>

takeWhileInclusive(Predicate<? super T> predicate)

Creates a gatherer that takes elements while the predicate is true, including the first element that fails the predicate.

static <T> Gatherer<T,?,T>

takeWhileIndexed(BiPredicate<Long, ? super T> predicate)

Creates a take-while-indexed gatherer.

static <T> Gatherer<T,?,List<T>>

windowFixed(int size)

Creates a fixed window gatherer.

static <T> Gatherer<T,?,List<T>>

windowSliding(int size)

Creates a sliding window gatherer.

static <T> Gatherer<T,?,List<T>>

windowSliding(int size, int step)

Creates a sliding window gatherer with custom step.

static <T> Gatherer<T, ?, Pair<Long,T>>

zipWithIndex()

Creates a gatherer that pairs each element with its zero-based index.

static <T,R> Gatherer<T,?,R>

zipWithNext(BiFunction<? super T, ? super T, ? extends R> zipper)

Creates a zip-with-next gatherer.

Methods inherited from class Object
clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Method Details
- windowFixed
  
  public static <T> Gatherer<T,?,List<T>> windowFixed(int size)
  
  Creates a fixed window gatherer. 创建固定窗口收集器。
  Groups elements into fixed-size lists (tumbling window).
  
  将元素分组为固定大小的列表（翻滚窗口）。
  
  Type Parameters:
  
  T - element type | 元素类型
  
  Parameters:
  
  size - window size | 窗口大小
  
  Returns:
  
  gatherer | 收集器
- windowSliding
  
  public static <T> Gatherer<T,?,List<T>> windowSliding(int size)
  
  Creates a sliding window gatherer. 创建滑动窗口收集器。
  Creates overlapping windows of the specified size.
  
  创建指定大小的重叠窗口。
  
  Type Parameters:
  
  T - element type | 元素类型
  
  Parameters:
  
  size - window size | 窗口大小
  
  Returns:
  
  gatherer | 收集器
- windowSliding
  
  public static <T> Gatherer<T,?,List<T>> windowSliding(int size, int step)
  
  Creates a sliding window gatherer with custom step. 创建自定义步长的滑动窗口收集器。
  
  Type Parameters:
  
  T - element type | 元素类型
  
  Parameters:
  
  size - window size | 窗口大小
  
  step - step size | 步长
  
  Returns:
  
  gatherer | 收集器
- batch
  
  public static <T> Gatherer<T,?,List<T>> batch(int batchSize)
  
  Creates a batch gatherer. 创建批处理收集器。
  Alias for windowFixed for semantic clarity.
  
  windowFixed 的别名，用于语义清晰。
  
  Type Parameters:
  
  T - element type | 元素类型
  
  Parameters:
  
  batchSize - batch size | 批次大小
  
  Returns:
  
  gatherer | 收集器
- batchProcess
  
  public static <T,R> Gatherer<T,?,R> batchProcess(int batchSize, Function<List<T>,R> processor)
  
  Creates a batch gatherer with processing function. 创建带处理函数的批处理收集器。
  
  Type Parameters:
  
  T - element type | 元素类型
  
  R - result type | 结果类型
  
  Parameters:
  
  batchSize - batch size | 批次大小
  
  processor - batch processor | 批次处理器
  
  Returns:
  
  gatherer | 收集器
- distinctBy
  
  public static <T,K> Gatherer<T,?,T> distinctBy(Function<? super T, ? extends K> keyExtractor)
  
  Creates a distinct-by-key gatherer. 创建按键去重收集器。
  
  Type Parameters:
  
  T - element type | 元素类型
  
  K - key type | 键类型
  
  Parameters:
  
  keyExtractor - key extractor | 键提取器
  
  Returns:
  
  gatherer | 收集器
- distinctBy
  
  public static <T,K> Gatherer<T,?,T> distinctBy(Function<? super T, ? extends K> keyMapper, BiPredicate<K,K> equals, ToIntFunction<K> hash)
  
  Creates a distinct-by gatherer with custom equality. 创建自定义相等性的去重收集器。
  
  Type Parameters:
  
  T - element type | 元素类型
  
  K - key type | 键类型
  
  Parameters:
  
  keyMapper - key mapper | 键映射器
  
  equals - equality function | 相等函数
  
  hash - hash function | 哈希函数
  
  Returns:
  
  gatherer | 收集器
- scan
  
  public static <T,R> Gatherer<T,?,R> scan(R initial, BiFunction<R, ? super T, R> accumulator)
  
  Creates a scan (running fold) gatherer. 创建扫描（运行折叠）收集器。
  Produces intermediate accumulation results.
  
  产生中间累积结果。
  
  Type Parameters:
  
  T - element type | 元素类型
  
  R - result type | 结果类型
  
  Parameters:
  
  initial - initial value | 初始值
  
  accumulator - accumulator function | 累加器函数
  
  Returns:
  
  gatherer | 收集器
- scan
  
  public static <T,A,R> Gatherer<T,?,R> scan(Supplier<A> initial, BiFunction<A, ? super T, A> accumulator, Function<A,R> finisher)
  
  Creates a scan gatherer with initial value and finisher. 创建带初始值和完成器的扫描收集器。
  
  Type Parameters:
  
  T - element type | 元素类型
  
  A - accumulator type | 累加器类型
  
  R - result type | 结果类型
  
  Parameters:
  
  initial - initial value supplier | 初始值提供者
  
  accumulator - accumulator function | 累加器函数
  
  finisher - finisher function | 完成器函数
  
  Returns:
  
  gatherer | 收集器
- takeWhileIndexed
  
  public static <T> Gatherer<T,?,T> takeWhileIndexed(BiPredicate<Long, ? super T> predicate)
  
  Creates a take-while-indexed gatherer. 创建带索引的takeWhile收集器。
  
  Type Parameters:
  
  T - element type | 元素类型
  
  Parameters:
  
  predicate - predicate (index, element) | 谓词 (索引, 元素)
  
  Returns:
  
  gatherer | 收集器
- dropWhileIndexed
  
  public static <T> Gatherer<T,?,T> dropWhileIndexed(BiPredicate<Long, ? super T> predicate)
  
  Creates a drop-while-indexed gatherer. 创建带索引的dropWhile收集器。
  
  Type Parameters:
  
  T - element type | 元素类型
  
  Parameters:
  
  predicate - predicate (index, element) | 谓词 (索引, 元素)
  
  Returns:
  
  gatherer | 收集器
- filterIndexed
  
  public static <T> Gatherer<T,?,T> filterIndexed(BiPredicate<Long, ? super T> predicate)
  
  Creates a filter-indexed gatherer. 创建带索引的过滤收集器。
  
  Type Parameters:
  
  T - element type | 元素类型
  
  Parameters:
  
  predicate - predicate (index, element) | 谓词 (索引, 元素)
  
  Returns:
  
  gatherer | 收集器
- changed
  
  public static <T> Gatherer<T,?,T> changed()
  
  Creates a changed gatherer (emit only when value changes). 创建变化收集器（仅当值变化时发出）。
  
  Type Parameters:
  
  T - element type | 元素类型
  
  Returns:
  
  gatherer | 收集器
- changedBy
  
  public static <T,K> Gatherer<T,?,T> changedBy(Function<? super T, ? extends K> keyExtractor)
  
  Creates a changed-by gatherer (emit when key changes). 创建按键变化收集器（当键变化时发出）。
  
  Type Parameters:
  
  T - element type | 元素类型
  
  K - key type | 键类型
  
  Parameters:
  
  keyExtractor - key extractor | 键提取器
  
  Returns:
  
  gatherer | 收集器
- mapIndexed
  
  public static <T,R> Gatherer<T,?,R> mapIndexed(BiFunction<Long, ? super T, ? extends R> mapper)
  
  Creates a map-indexed gatherer. 创建带索引的映射收集器。
  
  Type Parameters:
  
  T - element type | 元素类型
  
  R - result type | 结果类型
  
  Parameters:
  
  mapper - mapper (index, element) | 映射器 (索引, 元素)
  
  Returns:
  
  gatherer | 收集器
- mapWithPrevious
  
  public static <T,R> Gatherer<T,?,R> mapWithPrevious(BiFunction<? super T, ? super T, ? extends R> mapper)
  
  Creates a map-with-previous gatherer. 创建带前一个元素的映射收集器。
  
  Type Parameters:
  
  T - element type | 元素类型
  
  R - result type | 结果类型
  
  Parameters:
  
  mapper - mapper (previous, current) | 映射器 (前一个, 当前)
  
  Returns:
  
  gatherer | 收集器
- zipWithNext
  
  public static <T,R> Gatherer<T,?,R> zipWithNext(BiFunction<? super T, ? super T, ? extends R> zipper)
  
  Creates a zip-with-next gatherer. 创建与下一个元素配对的收集器。
  
  Type Parameters:
  
  T - element type | 元素类型
  
  R - result type | 结果类型
  
  Parameters:
  
  zipper - zipper function | 配对函数
  
  Returns:
  
  gatherer | 收集器
- groupRuns
  
  public static <T,K> Gatherer<T,?,List<T>> groupRuns(Function<? super T, ? extends K> keyExtractor)
  
  Creates a group-runs gatherer. 创建连续分组收集器。
  Groups consecutive elements with the same key.
  
  将具有相同键的连续元素分组。
  
  Type Parameters:
  
  T - element type | 元素类型
  
  K - key type | 键类型
  
  Parameters:
  
  keyExtractor - key extractor | 键提取器
  
  Returns:
  
  gatherer | 收集器
- takeLast
  
  public static <T> Gatherer<T,?,T> takeLast(int n)
  
  Creates a take-last gatherer. 创建获取最后n个元素的收集器。
  
  Type Parameters:
  
  T - element type | 元素类型
  
  Parameters:
  
  n - number of elements | 元素数量
  
  Returns:
  
  gatherer | 收集器
- dropLast
  
  public static <T> Gatherer<T,?,T> dropLast(int n)
  
  Creates a drop-last gatherer. 创建丢弃最后n个元素的收集器。
  
  Type Parameters:
  
  T - element type | 元素类型
  
  Parameters:
  
  n - number of elements | 元素数量
  
  Returns:
  
  gatherer | 收集器
- fold
  
  public static <T,R> Gatherer<T,?,R> fold(R initial, BiFunction<R, ? super T, R> accumulator)
  
  Creates a fold gatherer (produces single result). 创建折叠收集器（产生单个结果）。
  
  Type Parameters:
  
  T - element type | 元素类型
  
  R - result type | 结果类型
  
  Parameters:
  
  initial - initial value | 初始值
  
  accumulator - accumulator function | 累加器函数
  
  Returns:
  
  gatherer | 收集器
- intersperse
  
  public static <T> Gatherer<T,?,T> intersperse(T separator)
  
  Creates an interleave gatherer with separators. 创建带分隔符的交错收集器。
  
  Type Parameters:
  
  T - element type | 元素类型
  
  Parameters:
  
  separator - separator element | 分隔符元素
  
  Returns:
  
  gatherer | 收集器
- indexed
  
  public static <T> Gatherer<T, ?, OpenGatherers.IndexedElement<T>> indexed()
  
  Creates an indexed wrapper gatherer. 创建索引包装收集器。
  
  Type Parameters:
  
  T - element type | 元素类型
  
  Returns:
  
  gatherer producing indexed elements | 产生索引元素的收集器
- zipWithIndex
  
  public static <T> Gatherer<T, ?, Pair<Long,T>> zipWithIndex()
  
  Creates a gatherer that pairs each element with its zero-based index. 创建一个将每个元素与其从零开始的索引配对的收集器。
  Unlike indexed() which returns OpenGatherers.IndexedElement, this method returns Pair<Long, T> for better interoperability with other Pair-based APIs.
  
  与返回 OpenGatherers.IndexedElement 的 indexed() 不同，此方法返回 Pair<Long, T> 以便与其他基于 Pair 的 API 更好地互操作。
  
  Example: [a, b, c] → [Pair(0,a), Pair(1,b), Pair(2,c)]
  
  Type Parameters:
  
  T - element type | 元素类型
  
  Returns:
  
  gatherer producing index-element pairs | 产生索引-元素配对的收集器
  
  Since:
  
  JDK 25, opencode-base-collections V1.0.3
- takeWhileInclusive
  
  public static <T> Gatherer<T,?,T> takeWhileInclusive(Predicate<? super T> predicate)
  
  Creates a gatherer that takes elements while the predicate is true, including the first element that fails the predicate. 创建一个在谓词为真时获取元素的收集器，包括第一个不满足谓词的元素。
  This differs from JDK's takeWhile which excludes the boundary element.
  
  这与 JDK 的 takeWhile（排除边界元素）不同。
  
  Example: [1,2,3,4,5].takeWhileInclusive(x -> x < 3) → [1,2,3]
  
  Type Parameters:
  
  T - element type | 元素类型
  
  Parameters:
  
  predicate - the predicate to test elements | 用于测试元素的谓词
  
  Returns:
  
  gatherer | 收集器
  
  Throws:
  
  NullPointerException - if predicate is null | 如果谓词为 null
  
  Since:
  
  JDK 25, opencode-base-collections V1.0.3
- interleave
  
  public static <T> Gatherer<T,?,T> interleave(Iterator<? extends T> other)
  
  Creates a gatherer that interleaves elements from the source stream with elements from the provided iterator, alternating between them. 创建一个将源流元素与提供的迭代器元素交替合并的收集器。
  If one source is exhausted before the other, remaining elements from the longer source are appended.
  
  如果一个源在另一个之前耗尽，较长源的剩余元素将被追加。
  
  Example: [1,2,3].interleave([a,b]) → [1,a,2,b,3]
  
  Type Parameters:
  
  T - element type | 元素类型
  
  Parameters:
  
  other - the iterator to interleave with | 要交错合并的迭代器
  
  Returns:
  
  gatherer | 收集器
  
  Throws:
  
  NullPointerException - if other is null | 如果 other 为 null
  
  Since:
  
  JDK 25, opencode-base-collections V1.0.3
- peekWithIndex
  
  public static <T> Gatherer<T,?,T> peekWithIndex(BiConsumer<Long, ? super T> action)
  
  Creates a peek gatherer for debugging. 创建用于调试的peek收集器。
  
  Type Parameters:
  
  T - element type | 元素类型
  
  Parameters:
  
  action - action to perform | 要执行的操作
  
  Returns:
  
  gatherer | 收集器

Class OpenGatherers

Nested Class Summary

Method Summary

Methods inherited from class Object

Method Details

windowFixed

windowSliding

windowSliding

batch

batchProcess

distinctBy

distinctBy

scan

scan

takeWhileIndexed

dropWhileIndexed

filterIndexed

changed

changedBy

mapIndexed

mapWithPrevious

zipWithNext

groupRuns

takeLast

dropLast

fold

intersperse

indexed

zipWithIndex

takeWhileInclusive

interleave

peekWithIndex