Add Flow#collectNonNull operator

[He-Pin]

This operator is been used in Spring ai, and I think it's actually useful than a map to Optional/Option and a then filter/collect

[He-Pin]

@raboof @mdedetrich @pjfanning wdyt, If you think this is ok, I will work on this weekend.

[mdedetrich]

Not sure that this makes sense as null as an element is not allowed in streams as per the reactive specification. If you send null in a Stream then it terminates immediately. In other words stream elements always have to be not null unless your desire is to explicitly terminate it.

That's why for Kotlin use of pekko streams you have to use Java Optional and not ?.

[raboof]

I don't find the name very clear - I had to google it and found https://kotlinlang.org/api/core/kotlin-stdlib/kotlin.collections/map-not-null.html (I didn't find it in Spring AI).

Couldn't you achieve this with a single collect, as that allows a partial function?

[He-Pin]

https://projectreactor.io/docs/core/release/api/reactor/core/publisher/Flux.html#mapNotNull-java.util.function.Function-
That true, @raboof yes, but this is more for a Java api.

[He-Pin]

And the use case in Spring ai https://github.com/spring-projects/spring-ai/blob/main/spring-ai-client-chat/src/main/java/org/springframework/ai/chat/client/DefaultChatClient.java#L589

[He-Pin]

@mdedetrich This operator will make avoid something like Source.map(Optional.ofNullable(...)).filter(Optional::isPresent).map(Optional::get)

[raboof]

It's an interesting observation that https://pekko.apache.org/japi/pekko/snapshot/org/apache/pekko/stream/javadsl/Flow.html#collect(scala.PartialFunction) is present on the Java API, but uses scala.PartialFunction.

We do supply some helper methods to make it easier to create Scala partial functions from Java (https://pekko.apache.org/docs/pekko/current/stream/operators/Source-or-Flow/collect.html), but I see how this is a bit icky.

For 2.0.0, it might be interesting to revisit if we want to have a replacement for Java. "A regular Java function, and when it returns 'null' that means it is not defined for that input" might be an interesting option, if the reactive spec indeed specifies null can never be an 'element' value. Perhaps we should look at other places where we use partial functions in Java APIs to see if that would be an option.

It's interesting that Flux also has something called 'collect', but that is more like a 'Sink'.

[He-Pin]

I like your idea, maybe a collectNonNull when it returns null, just dropped, especially with Java 21.

[mdedetrich]

@mdedetrich This operator will make avoid something like Source.map(Optional.ofNullable(...)).filter(Optional::isPresent).map(Optional::get)

Yes but how is this even going to work? On a fundamental level, pekko streams can never have null as an element, see https://doc.akka.io/libraries/akka-core/current/stream/stream-flows-and-basics.html#illegal-stream-elements.

I don't think this is even possible

[He-Pin]

It's possible, the null will just be ignored for some operator, and trigger another pull(in) if the upstream is not completed and not been pulled yet. just as the mapAsync operator, where when you return a CompletionStage<T> but the value is null, then the null is just dropped and trigger another pull(in), you can check the scaladoc of mapAsync.

I think we can have a special operator, which just drop the null value, when working with Java 21 's patten matching, it works like a charm

[mdedetrich]

It's possible, the null will just be ignored for some operator, and trigger another pull(in) if the upstream is not completed and not been pulled yet. just as the mapAsync operator, where when you return a CompletionStage<T> but the value is null, then the null is just dropped and trigger another pull(in), you can check the scaladoc of mapAsync.

I think we can have a special operator, which just drop the null value, when working with Java 21 's patten matching, it works like a charm

The entire pekko-stream codebase is all designed with the assumption that null is never going to be any element. This is not something thats easy to change, and I personally wouldn't accept such an operator until pekko streams in general works with null elements, which could be something for 2.0.0 but this is a massive change

There are way too many edge cases and surprising behaviour otherwise, and I have personally spent days/weeks debugging streams due to a single stranded null element being sent somewhere. The way that null behaves also differs completely depending on which streaming operator.

Illegal means illegal, the entire codebase is based on the assumption that it should never exist.

[mdedetrich]

And don't get me wrong, I would love pekko streams to work with null (it would make it much more ergonomic for kotlin users who idiomatically don't use Java Optional) but making null work with pekko streams is HUGE change