Fix uncurried Pekko Stream ops in javadsl

[mdedetrich]

So I discovered this when using the javadsl within a kotlin project but it appears that many of the stream functions in javadsl are using their curried equivalents (i.e. limitWeighted(n: Long)(costFn: function.Function[Out, java.lang.Long]) vs limitWeighted(n: Long, costFn: function.Function[Out, java.lang.Long]),

This appears to be a massive oversight, as not only is this currently entirely inconsistent in the codebase (see fold vs foldWhile in Flow javadsl or fold in Source vs fold in Flow javadsl) but Java has no real concept curried functions and so applying curried functions written in Scala is not idiomatic. For Java the standard practice is to just have a single parameter list with the function as the last arg (and Kotlin also expects this).

There is also the case of watchTermination which had a zero arg curried implementation i.e. def watchTermination[M]()(matF: function.Function2[Mat, CompletionStage[Done], M]). I don't know if the zero arg was doing anything meaningful (afaik curried function application is strict so its not suspending the future unless you partially apply it), in any case I added a def watchTermination[M](matF: function.Function2[Mat, CompletionStage[Done], M]) variant so check if this is indeed correct and I am not messing a behaviour (if the future is meant to suspended in a thunk then we should just use java function).

@He-Pin @jrudolph @raboof Maybe you want to have a look at this PR since you are more intimate with the codebase, particularly wrt watchTermination and if its arg needs to be suspended or not

[He-Pin]

When using it from Java, it works, does it cause problems when using it from Kotlin? IDEA/Comping error? Does that affect the kotlin's trailing lambda?

I think they did this maybe that can help usage the javadsl from Scala ? I don't know.

[mdedetrich]

So TIL, curried functions written in Scala appear as single parameter lists within Java so there isn't any actual problem its just that it really messes syntax highlighting/code completion in cross Scala/Java projects. Updating PR so that we only have a single version of the function without any deprecation's aside from watchTermination (since after erasure both versions are exactly the same).

[mdedetrich]

When using it from Java, it works, does it cause problems when using it from Kotlin? IDEA/Comping error? Does that affect the kotlin's trailing lambda?

So its not a compile issue (since both versions are the same after erasure and the curried version is compiled into a single parameter list) but it does mess with code completion/syntax highlighting in IDE.

In any case I think this change is valuable since it makes it a lot more clear

[jxnu-liguobin]

I agree with this. It's better to make changes as soon as possible.

[mdedetrich]

Pull request is ready, since MiMa has passed we have verification that generated bytecode is exactly the same. Rather this feature now improves quality of life for users (i.e. reading code, reading snippets taken from source code, generated docs from source, correctness in syntax highlighting in cross Scala projects etc etc)

[pjfanning]

How do implicit args work when calling from Java? Is it possible that this constructor is not meant to be used directly by Java users, that is for use by internal Pekko code?

[mdedetrich]

implicit args also get flattened to flat parameter lists since under the hood they are also implemented via currying (with only Scala having the ability to pass those arguments implicitly).

Thats what I figured out, any type of currying (whether its done directly or via implicits) is just translated to a standard flattened parameter list for JVM bytecode.

[pjfanning]

But do you think that this constructor is to meant to be called by external Java users? It looks like an internal Pekko use only constructor.

[mdedetrich]

On the surface you seem to be right although it is referred to in multiple places in the docs.

In any case even if the constructor is only meant for internal Pekko usage, this change is harmless.

[He-Pin]

This change is fine, but it does break some binary compatibility, I'm using Javadsl only in Java, so I think that should also apply to others.

[mdedetrich]

If it breaks binary compatibility MiMa would detect it.

In Java its just called as a one arg method

[He-Pin]

Does this affect the library which compiles with pekko 1.0.x but be used where only pekko .1.1x in classpath? eg a runtime NoSuchMethodException.

[pjfanning]

Does this affect the library which compiles with pekko 1.0.x but be used where only pekko .1.1x in classpath? eg a runtime NoSuchMethodException.

In theory no - because the Mima checks are happy that the change is binary compatible.

[mdedetrich]

For those that are concerned about MiMa, breaking binary compatible or proof that any combination of implicits/currying in Scala just gets converted to a flat parameter when generating JVM bytecode, with the current version of

final class Builder[+Mat]()(private[stream] implicit val delegate: scaladsl.GraphDSL.Builder[Mat])

This .java source file compiles (as you can see its just being called as a standard single arg Java method)

package org.apache.pekko.stream.javadsl;

public class Test {
    void test() {
        org.apache.pekko.stream.scaladsl.GraphDSL.Builder<Void> delegate = new org.apache.pekko.stream.scaladsl.GraphDSL.Builder<>();

        new org.apache.pekko.stream.javadsl.GraphDSL.Builder<>(delegate);
    }
}

And if you change final class Builder to

final class Builder[+Mat](private[stream] implicit val delegate: scaladsl.GraphDSL.Builder[Mat])

It also compiles without any changes to the .java source. As mentioned before if there was an issue MiMa would have picked this up anyways

[He-Pin]

lgtm , as a daily java developer, I think this is fine now.

[He-Pin]

btw, why you don't use the Kotlin flow in a Kotlin project:)

[mdedetrich]

btw, why you don't use the Kotlin flow in a Kotlin project:)

We are, I actually noticed this when using Pekko javadsl in a Kotlin project

[He-Pin]

I'm using pekko stream because kotlin flow lack some operators:)