From fbe513446d2aeb9c75435c7dd474134fe35d9731 Mon Sep 17 00:00:00 2001
From: Oliver Drotbohm <odrotbohm@vmware.com>
Date: Mon, 9 Aug 2021 17:04:11 +0200
Subject: [PATCH] =?UTF-8?q?StreamUtils.zip(=E2=80=A6)=20now=20treats=20inf?=
 =?UTF-8?q?inite=20streams=20correctly.?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

When an infinite Stream was handed into StreamUtils.zip(…) as first argument, the resulting stream was infinite, too, while inverting the argument order was limiting the resulting stream to the length of the finite one. This is now fixed by actually evaluating whether we can advance on both of the streams and shortcutting the process if that is not possible on either of the streams, limiting the processing of the overall Stream to the shorter of the two as already advertised in the Javadoc.

Fixes #2426.
---
 .../org/springframework/data/util/Sink.java   | 34 +++++++++++++++++++
 .../data/util/StreamUtils.java                | 21 ++++++++++--
 .../data/util/StreamUtilsTests.java           |  9 +++++
 3 files changed, 62 insertions(+), 2 deletions(-)
 create mode 100644 src/main/java/org/springframework/data/util/Sink.java

diff --git a/src/main/java/org/springframework/data/util/Sink.java b/src/main/java/org/springframework/data/util/Sink.java
new file mode 100644
index 000000000..55d3587e6
--- /dev/null
+++ b/src/main/java/org/springframework/data/util/Sink.java
@@ -0,0 +1,34 @@
+package org.springframework.data.util;
+
+import java.util.function.Consumer;
+
+import org.springframework.lang.Nullable;
+
+/**
+ * A simple {@link Consumer} that captures the instance handed into it.
+ *
+ * @author Oliver Drotbohm
+ * @since 2.4.12
+ */
+class Sink<T> implements Consumer<T> {
+
+	private T value;
+
+	/**
+	 * Returns the value captured.
+	 *
+	 * @return
+	 */
+	public T getValue() {
+		return value;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see java.util.function.Consumer#accept(java.lang.Object)
+	 */
+	@Override
+	public void accept(@Nullable T t) {
+		this.value = t;
+	}
+}
diff --git a/src/main/java/org/springframework/data/util/StreamUtils.java b/src/main/java/org/springframework/data/util/StreamUtils.java
index 0f8abfef8..f53a6736a 100644
--- a/src/main/java/org/springframework/data/util/StreamUtils.java
+++ b/src/main/java/org/springframework/data/util/StreamUtils.java
@@ -140,9 +140,26 @@ public interface StreamUtils {
 			@Override
 			@SuppressWarnings("null")
 			public boolean tryAdvance(Consumer<? super T> action) {
-				return lefts.tryAdvance(left -> rights.tryAdvance(right -> action.accept(combiner.apply(left, right))));
-			}
 
+				Sink<L> leftSink = new Sink<L>();
+				Sink<R> rightSink = new Sink<R>();
+
+				boolean leftAdvance = lefts.tryAdvance(leftSink);
+
+				if (!leftAdvance) {
+					return false;
+				}
+
+				boolean rightAdvance = rights.tryAdvance(rightSink);
+
+				if (!rightAdvance) {
+					return false;
+				}
+
+				action.accept(combiner.apply(leftSink.getValue(), rightSink.getValue()));
+
+				return true;
+			}
 		}, parallel);
 	}
 }
diff --git a/src/test/java/org/springframework/data/util/StreamUtilsTests.java b/src/test/java/org/springframework/data/util/StreamUtilsTests.java
index 296473a74..1e961f306 100755
--- a/src/test/java/org/springframework/data/util/StreamUtilsTests.java
+++ b/src/test/java/org/springframework/data/util/StreamUtilsTests.java
@@ -42,4 +42,13 @@ public class StreamUtilsTests {
 
 		assertThat(input).isEqualTo(output);
 	}
+
+	@Test // #2426
+	void combinesInfiniteStreamCorrectly() {
+
+		Stream<Long> indices = Stream.iterate(1L, n -> n + 1);
+		Stream<String> lines = Stream.of("first line", "second line");
+
+		assertThat(StreamUtils.zip(indices, lines, (index, line) -> index + ":" + line).count()).isEqualTo(2);
+	}
 }