Add Flux.unfold

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