Add Flux.unfold

[asakaev]

Summary

This PR introduces a new unfold operator to Flux, providing a functional approach for generating a sequence from an initial seed value. The unfold method offers a declarative alternative to the existing generate operator, inspired by similar functionality in Scala and Haskell.

Motivation

The unfold operator enhances the API by allowing users to construct sequences in a functional style, which may be more familiar to developers coming from FP-centric languages. While functionally similar to generate, unfold focuses on a clean, declarative approach to defining recursive sequences.

[pivotal-cla]

@asakaev Please sign the Contributor License Agreement!

Click here to manually synchronize the status of this Pull Request.

See the FAQ for frequently asked questions.

[pivotal-cla]

@asakaev Please sign the Contributor License Agreement!

Click here to manually synchronize the status of this Pull Request.

See the FAQ for frequently asked questions.

[pivotal-cla]

@asakaev Thank you for signing the Contributor License Agreement!

[chemicL]

Thanks for the proposal @asakaev. I can't say whether we'd like to integrate this, but it looks interesting as it's rather simple and encapsulates a familiar pattern. However, adding an operator opens up doors for more ideas. E.g. generate has 3 overloads. For unfold I can start to imagine someone with a desire for asynchronous variant the same way akka has. Consider the related discussion in #1974.

Would you be able to provide a few use cases comparing the usage of this new operator vs utilizing existing factories? That would be useful to guide the decision process. Thanks in advance.

[asakaev]

Thank you for considering my proposal! I'm excited to discuss how this aligns with the existing API.

Functional approach

The current SynchronousSink<T> API, while effective, operates at a low level and encourages side effects. In contrast, using a pure function S -> (T, S) eliminates these risks and offers a more reliable way to model state transitions.

The proposed unfold operator introduces a declarative approach to codata definition, where each step produces the next value and state. This leads to clearer, more maintainable code:

// evens [0, 2, 4, 6, ...]
Flux.unfold(0, s -> Optional.of(Tuples.of(s, s + 2)));

// fib [0, 1, 1, 2, 3, 5, 8, ...]
Flux.unfold(Tuples.of(0, 1), s -> {
    var a = s.getT1();
    var b = s.getT2();
    return Optional.of(Tuples.of(a, Tuples.of(b, a + b)));
});

// countdown [3, 2, 1]
Flux.unfold(3, s -> s == 0 ? Optional.empty() : Optional.of(Tuples.of(s, s - 1)));

Asynchronous variant

State-of-the-art libraries like fs2 and zio-streams have set the pattern with unfoldEval and unfoldZIO. In our case, an API like this would be a logical extension:

abstract <A, S> Flux<A> unfoldMono(S init, Function<S, Mono<Optional<Tuple2<A, S>>>> f);

However, this is beyond the scope of this PR, as Flux.generate requires immediate state calculation, which doesn't align with asynchronous Mono.

Next steps

If the proposal is interesting, I’ll add documentation and tests. For a deeper theoretical foundation, I recommend Conal Elliott’s talk on folds and unfolds.

Thank you again for your consideration!

[orchestr7]

Looks very promising. Any plans to add this functionality?

[declspec-cdecl]

Seems that it will be great for programmers that familiar with FP.

[chemicL]

Am I getting it right to compare the implementations as below?

Even numbers:

	Flux.unfold(0, s -> Optional.of(Tuples.of(s, s + 2)));

	Flux.generate(() -> 0, (current, sink) -> {
		sink.next(current);
		return current + 2;
	})

Fibonacci:

	Flux.unfold(Tuples.of(0, 1), s -> {
		var a = s.getT1();
		var b = s.getT2();
		return Optional.of(Tuples.of(a, Tuples.of(b, a + b)));
	});

	Flux.generate(() -> Tuples.of(0, 1), (current, sink) -> {
		sink.next(current);
		var a = s.getT1();
		var b = s.getT2();
		return Tuples.of(b, a + b);
	})

Countdown:

	Flux.unfold(3, s -> s == 0 ? Optional.empty() : Optional.of(Tuples.of(s, s - 1)));

	Flux.generate(() -> 3, (current, sink) -> {
			if (current == 0) {
				sink.complete();
			} else {
				sink.next(current--);
			}
			return current;
	    })

[asakaev]

Thanks for your detailed response and the examples @chemicL! You're absolutely on point with your comparisons.

I'd like to emphasize a few key advantages that the unfold operator brings:

Robustness & Simplicity

The unfold step function is pure, which ensures greater robustness by eliminating side effects. This makes the code not only easier to test in isolation but also more predictable and less prone to errors. Its concise implementation—typically 2.5x shorter than generate—further reduces complexity, enhancing both reliability and maintainability.

Type-Safe Predictability

The unfold operator promotes type-driven, predictable design, giving developers a clearer view of all possible states and transitions. This contrasts with the added complexity of generate, which involves more manual control through the SynchronousSink<T> API.

Referential Transparency

One of the most important benefits of unfold step function is its referential transparency. In functional programming, referential transparency means that a function can be replaced with its value without changing the program's behavior. This leads to several key advantages:

Predictability: The unfold step function does nothing other than compute the next value and state. No hidden interactions, no side effects. This makes it much easier to reason about the code, trace execution, and maintain correctness.

Mental ease: Code that's referentially transparent is easier to inspect and understand at a glance. This frees developers' mental space to focus on shipping quality software instead of worrying about intricate control flows or side effects.

In essence, by introducing unfold, we provide an elegant, functional approach to defining sequences, which is more predictable, easier to test, and safer to use.

If this direction makes sense, I’d be glad to move forward with the next steps.

Thanks again for your thoughtful consideration!

[asakaev]

Hi @chemicL,

I wanted to bring up another key aspect of this proposal. Currently, generate seems to be the primary option for building pull-based systems in Flux. These systems are crucial when you need control over the number and rate of calls, especially for more complex or resource-sensitive flows.

With unfold, we gain a more reliable and type-safe API for modeling such processes, while keeping things declarative and concise. Here are some additional examples demonstrating how unfold can simplify common patterns:

• Factorial sequence:

Flux<Integer> factorial() {
  return Flux.unfold(Tuples.of(1, 1), s -> {
    var n = s.getT1();
    var acc = s.getT2();
    return Optional.of(Tuples.of(acc, Tuples.of(n + 1, acc * (n + 1))));
  });
}

• Collatz conjecture:

Flux<Integer> collatz(Integer num) {
  return Flux.unfold(num, n ->
    n == 1 ?
      Optional.empty() :
      Optional.of(Tuples.of(n, n % 2 == 0 ? n / 2 : 3 * n + 1))
  );
}

• Other useful patterns:

<T> Flux<T> constant(T x) {
  return Flux.unfold(0, s -> Optional.of(Tuples.of(x, s)));
}

Flux<Boolean> alternate() {
  return Flux.unfold(true, x -> Optional.of(Tuples.of(x, !x)));
}

Flux<Integer> powersOfTen() {
  return Flux.unfold(1, x -> Optional.of(Tuples.of(x, x * 10)));
}

Flux<Integer> digits(Integer n) {
  return Flux.unfold(n, x ->
    x == 0 ?
      Optional.empty() :
      Optional.of(Tuples.of(x % 10, x / 10))
  );
}

<T> Flux<T> repeatN(Integer n, T x) {
  return Flux.unfold(n, i ->
    i == 0 ?
      Optional.empty() :
      Optional.of(Tuples.of(x, i - 1))
  );
}

Flux<Integer> random(Long seed) {
  return Flux.unfold(new Random(seed), rng ->
    Optional.of(Tuples.of(rng.nextInt(), rng))
  );
}

These examples show how clean and intuitive the unfold API can be, especially when compared to generate. It offers a declarative and type-driven alternative that helps developers reason more easily about the data flow and lifecycle management of state.

I’m excited to hear your thoughts and hope we can take this forward. Let me know if there's anything you'd like to discuss further!

[chemicL]

@asakaev thank you for the thorough explanation. I've been busy with other obligations recently, but now had the time to review your points. I am convinced with the proposal and I believe it would be a handy addition to the API. My request then is - can you please enhance your PR with:

• Relevant tests covering the operator
• Updating the reference documentation where the generator is used and consider whether we can do a better job at suggesting the right choice for use
  • Specifically: add a section next to https://projectreactor.io/docs/core/snapshot/reference/coreFeatures/programmatically-creating-sequence.html#producing.generate to explain the functional alternative of unfold
  • Also: mention the alternative in https://projectreactor.io/docs/core/snapshot/reference/apdx-operatorChoice.html#which.create
  • We could also use your examples in the reference documentation to highlight the ease of use and ideas for implementation to users are aware of the potential which you've shown.
• Please also consider the debugging perspective - it is important to make users understand that unfold will appear as with FluxGenerate and SynchronousSink in the stacktraces as it is not an operator but an alias - I do believe this explanation should also land in the Javadoc.

Thanks!

[mminella]

Thank you for this contribution. Unfortunately, as a project stewarded by Broadcom, we are unable to accept contributions from Russian sources due to Broadcom export policy at this time. Thanks for your continued use of Spring.

[LashaDev]

Thank you for this contribution. Unfortunately, as a project stewarded by Broadcom, we are unable to accept contributions from Russian sources due to Broadcom export policy at this time. Thanks for your continued use of Spring.

@mminella
I'm sorry, I'm a little confused.
If anyone else (for instance not related to Russia) opens the same PR - do you accept it?
So you decided to decline this just based on the contributor ethnicity/nationality. Sounds weird. Ho do you find this even legal?
Is it still 'open source' reactor project? Or it is just free work for you/Broadcom/whatever?

Let's check Contributor Covenant Code of Conduct.

We as members, contributors, and leaders pledge to make participation in this project and our community a harassment-free experience for everyone, regardless of age, body size, visible or invisible disability, ethnicity, sex characteristics, gender identity and expression, level of experience, education, socio-economic status, nationality, personal appearance, race, caste, color, religion, or sexual identity and orientation.
We pledge to act and interact in ways that contribute to an open, welcoming, diverse, inclusive, and healthy community.

So, @mminella could you please point us to the policy you used? What do you mean by Russian sources?

Thank you.

[medeyko]

Thank you for this contribution. Unfortunately, as a project stewarded by Broadcom, we are unable to accept contributions from Russian sources due to Broadcom export policy at this time.

Technically for Broadcom this pull request is an import, not an export, so it's unlikely that the export policy applies in any case.

[akuleshov7]

Created a new PR #3921

[vlow]

This thread understandably steers towards political and emotional statements rather quickly and I'd like to add a few observations to nurture the more constructive parts of the discussion:

I know there is a raging war with unimaginable suffering and uncountable deaths. There is no denying or belittling this and no "but". Please keep in mind that it's highly unlikely the decission to close this PR was meant as a political statement by those who made it.

This decission has the smell of corporate legal and CYA all over it. It's highly probable that @mminella is just the bearer of news and chances of resolving this issue in discussion with him are slim.

I assume the obvious road block in merging this PR is the plain visibility of its authors whereabouts in his profile. I can't imagine any realistic way of preventing people with a certain nationality from OSS projects like Spring other than looking for obvious signs in there publicly available profile. So the factual consequence of the corporate decision to close PRs based on these criteria seems superficial at best.

That said, I also can't imagine just re-opening this contribution from another account would resolve any legal issue since I assume the copyright/intellectual property is obviously still @asakaev's which seems to be the actual issue. I'm no lawyer, but it's hard to imagine it would be so easy to resolve this issue.