Add Streams.forEachWithIndex

android/guava/src/com/google/common/collect/Streams.java

@@ -44,6 +44,7 @@
 import java.util.function.DoubleConsumer;
 import java.util.function.IntConsumer;
 import java.util.function.LongConsumer;
+import java.util.function.ObjLongConsumer;
 import java.util.stream.BaseStream;
 import java.util.stream.DoubleStream;
 import java.util.stream.IntStream;
@@ -414,6 +415,48 @@ public boolean tryAdvance(Consumer<? super R> action) {
     }
   }
 
+  /**
+   * Invokes {@code consumer} once for each element of {@code stream} and its index in the stream.
+   * For example,
+   *
+   * {@snippet :
+   * Streams.forEachWithIndex(
+   *     Stream.of("a", "b", "c"),
+   *     (e, index) -> System.out.println(index + ": " + e))
+   * }
+   *
+   * <p>would print:
+   *
+   * {@snippet :
+   * 0: a
+   * 1: b
+   * 2: c
+   * }
+   *
+   * <p><b>Warning:</b> If {@code stream} is a parallel stream, the elements may be passed to the
+   * consumer in any order. The index assigned to each element reflects its position in the stream's
+   * <i>encounter order</i>. If the stream has no defined encounter order, the index assigned to
+   * each element is arbitrary.
+   *
+   * <p>This method behaves equivalently to applying {@link #mapWithIndex(Stream,
+   * FunctionWithIndex)} and then invoking {@link Stream#forEach} on the resulting stream.
+   *
+   * @since 33.6.0
+   */
+  @Beta
+  public static <T extends @Nullable Object> void forEachWithIndex(
+      Stream<T> stream, ObjLongConsumer<? super T> consumer) {
+    checkNotNull(stream);
+    checkNotNull(consumer);
+    mapWithIndex(
+            stream,
+            (element, index) -> {
+              consumer.accept(element, index);
+              return null;
+            })
+        .forEach(ignored -> {});
+  }
+
   // Use this carefully - it doesn't implement value semantics
   private static final class TemporaryPair<A extends @Nullable Object, B extends @Nullable Object> {
     @ParametricNullness final A a;

guava-tests/test/com/google/common/collect/StreamsTest.java

@@ -19,6 +19,7 @@
 import static com.google.common.truth.Truth.assertThat;
 import static java.util.Arrays.asList;
 import static java.util.stream.Collectors.toCollection;
+import static org.junit.Assert.assertThrows;
 
 import com.google.common.annotations.GwtCompatible;
 import com.google.common.annotations.GwtIncompatible;
@@ -33,6 +34,7 @@
 import java.util.OptionalDouble;
 import java.util.OptionalInt;
 import java.util.OptionalLong;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Function;
 import java.util.stream.DoubleStream;
@@ -508,6 +510,66 @@ public void testForEachPair_finiteWithInfinite() {
     assertThat(list).containsExactly("a:1", "b:2", "c:3");
   }
 
+  private void testForEachWithIndex(Function<Collection<String>, Stream<String>> sourceFactory) {
+    List<String> result = new ArrayList<>();
+    Streams.forEachWithIndex(
+        sourceFactory.apply(ImmutableList.of()), (e, i) -> result.add(i + ":" + e));
+    assertThat(result).isEmpty();
+
+    Streams.forEachWithIndex(
+        sourceFactory.apply(ImmutableList.of("a", "b", "c")),
+        (e, i) -> result.add(i + ":" + e));
+    assertThat(result).containsExactly("0:a", "1:b", "2:c").inOrder();
+  }
+
+  public void testForEachWithIndex_sizedSource() {
+    testForEachWithIndex(Collection::stream);
+  }
+
+  public void testForEachWithIndex_nullElement() {
+    List<@Nullable String> collected = new ArrayList<>();
+    List<Long> indices = new ArrayList<>();
+    Streams.forEachWithIndex(
+        Stream.<@Nullable String>of("a", null, "c"),
+        (e, i) -> {
+          collected.add(e);
+          indices.add(i);
+        });
+    assertThat(collected).containsExactly("a", null, "c").inOrder();
+    assertThat(indices).containsExactly(0L, 1L, 2L).inOrder();
+  }
+
+  public void testForEachWithIndex_nullChecks() {
+    assertThrows(
+        NullPointerException.class, () -> Streams.forEachWithIndex(null, (e, i) -> {}));
+    assertThrows(
+        NullPointerException.class, () -> Streams.forEachWithIndex(Stream.of("a"), null));
+  }
+
+  public void testForEachWithIndex_unsizedSource() {
+    // flatMap strips SUBSIZED, exercising mapWithIndex's iterator-backed fallback.
+    testForEachWithIndex(
+        elems ->
+            Stream.<@Nullable Object>of((Object) null)
+                .flatMap(unused -> ImmutableList.copyOf(elems).stream()));
+  }
+
+  public void testForEachWithIndex_parallelStream() {
+    int n = 200;
+    List<String> input = new ArrayList<>();
+    for (int j = 0; j < n; j++) {
+      input.add("e" + j);
+    }
+
+    ConcurrentHashMap<String, Long> result = new ConcurrentHashMap<>();
+    Streams.forEachWithIndex(input.stream().parallel(), (e, i) -> result.put(e, i));
+
+    assertThat(result).hasSize(n);
+    for (int j = 0; j < n; j++) {
+      assertThat(result).containsEntry("e" + j, (long) j);
+    }
+  }
+
   public void testForEachPair_parallel() {
     Stream<String> streamA = IntStream.range(0, 100000).mapToObj(String::valueOf).parallel();
     Stream<Integer> streamB = IntStream.range(0, 100000).mapToObj(i -> i).parallel();

guava/src/com/google/common/collect/Streams.java

@@ -44,6 +44,7 @@
 import java.util.function.DoubleConsumer;
 import java.util.function.IntConsumer;
 import java.util.function.LongConsumer;
+import java.util.function.ObjLongConsumer;
 import java.util.stream.BaseStream;
 import java.util.stream.DoubleStream;
 import java.util.stream.IntStream;
@@ -408,6 +409,48 @@ public boolean tryAdvance(Consumer<? super R> action) {
     }
   }
 
+  /**
+   * Invokes {@code consumer} once for each element of {@code stream} and its index in the stream.
+   * For example,
+   *
+   * {@snippet :
+   * Streams.forEachWithIndex(
+   *     Stream.of("a", "b", "c"),
+   *     (e, index) -> System.out.println(index + ": " + e))
+   * }
+   *
+   * <p>would print:
+   *
+   * {@snippet :
+   * 0: a
+   * 1: b
+   * 2: c
+   * }
+   *
+   * <p><b>Warning:</b> If {@code stream} is a parallel stream, the elements may be passed to the
+   * consumer in any order. The index assigned to each element reflects its position in the stream's
+   * <i>encounter order</i>. If the stream has no defined encounter order, the index assigned to
+   * each element is arbitrary.
+   *
+   * <p>This method behaves equivalently to applying {@link #mapWithIndex(Stream,
+   * FunctionWithIndex)} and then invoking {@link Stream#forEach} on the resulting stream.
+   *
+   * @since 33.6.0
+   */
+  @Beta
+  public static <T extends @Nullable Object> void forEachWithIndex(
+      Stream<T> stream, ObjLongConsumer<? super T> consumer) {
+    checkNotNull(stream);
+    checkNotNull(consumer);
+    mapWithIndex(
+            stream,
+            (element, index) -> {
+              consumer.accept(element, index);
+              return null;
+            })
+        .forEach(ignored -> {});
+  }
+
   // Use this carefully - it doesn't implement value semantics
   private static final class TemporaryPair<A extends @Nullable Object, B extends @Nullable Object> {
     @ParametricNullness final A a;