From 33c8f52db4c54121635d4c69a7fbddb09c0828fb Mon Sep 17 00:00:00 2001
From: Aleksandr Mikhailov <iam@fidonode.me>
Date: Thu, 7 Nov 2024 18:34:48 +0200
Subject: [PATCH] Add Flux.unfold

---
 .../java/reactor/core/publisher/Flux.java     | 25 +++++++++++++++++++
 1 file changed, 25 insertions(+)

diff --git a/reactor-core/src/main/java/reactor/core/publisher/Flux.java b/reactor-core/src/main/java/reactor/core/publisher/Flux.java
index 055a0e97be..71978ace16 100644
--- a/reactor-core/src/main/java/reactor/core/publisher/Flux.java
+++ b/reactor-core/src/main/java/reactor/core/publisher/Flux.java
@@ -29,6 +29,7 @@
 import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.Queue;
 import java.util.Set;
 import java.util.Spliterator;
@@ -2073,6 +2074,30 @@ public static <T> Flux<T> switchOnNext(Publisher<? extends Publisher<? extends T
 				Queues.unbounded(prefetch), prefetch));
 	}
 
+	/**
+ 	 * Creates a {@link Flux} that uses a function `f` to produce elements of type `T`
+ 	 * and update an internal state of type `S`.
+ 	 *
+ 	 * @param init State initial value
+ 	 * @param f    Computes the next element (or returns empty {@link Optional} to signal the end of the sequence)
+ 	 * @param <T>  Type of the elements
+ 	 * @param <S>  Type of the internal state
+ 	 *
+ 	 * @return a {@link Flux} that produces elements using `f` until `f` returns empty {@link Optional}
+ 	 */
+ 	public static <T, S> Flux<T> unfold(S init, Function<S, Optional<Tuple2<T, S>>> f) {
+ 		return Flux.generate(() -> init, (s, sink) -> {
+ 			Optional<Tuple2<T, S>> res = f.apply(s);
+ 			if (!res.isPresent()) {
+ 				sink.complete();
+ 				return s;
+ 			} else {
+ 				sink.next(res.get().getT1());
+ 				return res.get().getT2();
+ 			}
+ 		});
+ 	}
+
 	/**
 	 * Uses a resource, generated by a supplier for each individual Subscriber, while streaming the values from a
 	 * Publisher derived from the same resource and makes sure the resource is released if the sequence terminates or