Skip to content

Conversation

@pjfanning
Copy link
Member

@pjfanning pjfanning commented Nov 8, 2025

  • Currently users have to use JavaFlowSupport.Source and JavaFlowSupport.Sink instead
  • Still some doc changes to make
  • In particular, I need to create asJavaSubscriber.md and asJavaPublisher.md

@pjfanning pjfanning added this to the 2.0.0-M1 milestone Nov 8, 2025
@pjfanning pjfanning marked this pull request as draft November 8, 2025 23:43
override def createPublisher(elements: Long): Publisher[Int] = {
val sourceViaJavaFlowPublisher: JavaFlow.Publisher[Int] = Source(iterable(elements))
.runWith(JavaFlowSupport.Sink.asPublisher(fanout = false))
.runWith(Sink.asJavaPublisher(fanout = false))
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nice, much clear

@raboof
Copy link
Member

raboof commented Nov 9, 2025

I wonder if we shouldn't go a step further, and instead of adding a asJavaPublisher, actually changing the existing asPublisher to use the Java classes instead of the old org.reactivestreams ones

@pjfanning
Copy link
Member Author

I wonder if we shouldn't go a step further, and instead of adding a asJavaPublisher, actually changing the existing asPublisher to use the Java classes instead of the old org.reactivestreams ones

There is #1977. I would prefer to treat that as a separate issue. In that issue, @He-Pin points out that some libs still prefer reactivestreams lib (due to inertia or continuing desire to support Java 8). I have little issue with supporting both.

It just seems like we shouldn't treat the Java Flow API as a 2nd class citizen.

@raboof
Copy link
Member

raboof commented Nov 9, 2025

It just seems like we shouldn't treat the Java Flow API as a 2nd class citizen.

Much agreed. Perhaps we could make asPublisher the Java Flow one and rename the Reactive Streams one to something like asReactiveStreamsPublisher?

@mdedetrich
Copy link
Contributor

It just seems like we shouldn't treat the Java Flow API as a 2nd class citizen.

Much agreed. Perhaps we could make asPublisher the Java Flow one and rename the Reactive Streams one to something like asReactiveStreamsPublisher?

Another alternative would be to have static methods i.e. ReactiveStreams.asPublisher(source) vs JavaFlow.asPublisher(source) where ReactiveStreams would be a Scala object and JavaFlow would be a Java class with asPublisher being static methods.

That way the method name is the same, but you import a different object depending on if you want Java's flow or Reactive Streams

@pjfanning
Copy link
Member Author

@mdedetrich @raboof Is it just the asJavaSubscriber and asJavaProducer that is causing concerns?

If I separate out those changes but keep the moves of the fromPublisher and fromSubscriber methods, is that ok? Those methods can be overloaded because they don't hit issues with generics and type erasure.

@mdedetrich
Copy link
Contributor

@mdedetrich @raboof Is it just the asJavaSubscriber and asJavaProducer that is causing concerns?

If I separate out those changes but keep the moves of the fromPublisher and fromSubscriber methods, is that ok? Those methods can be overloaded because they don't hit issues with generics and type erasure.

For me the design is cleaner to have asSubscriber/asPublisher and fromSubscriber/fromPublisher method names reused for both reactive streams and Java flow and just have different FQN (including the object), i.e. pekko.stream.JavaFlow.asSubscriber vs pekko.stream.ReactiveStreams.asSubscriber.

If the design is breaking with Pekko 1.3.x you can create alias methods in Pekko 1.3.x that point to the new design in 2.0.0 and deprecate the older ones (Pekko 1.3.x hasn't been released yet).

@pjfanning
Copy link
Member Author

I am focusing on 2.x changes. I'm not going to rush a change to suit the 1.3.0 timeline. Pekko 2.x is built with Java 17 making it easier to support and test Java Flow API integration.

@pjfanning
Copy link
Member Author

We already have JavaFlowSupport so I guess the feedback is that some people prefer that to us moving the code to Source and Sink directly. The question then arises about us deprecating the org.reactivestreams code in Source and Sink and creating a ReactiveStreamsSupport class that mirrors the JavaFlowSupport.

A more radical reworking sees us deprecating JavaFlowSupport and creating a new class that has similar utility functions but possibly with less verbose naming and then to mirror this new class to create an org.reactivestreams equivalent.

@mdedetrich
Copy link
Contributor

mdedetrich commented Nov 9, 2025

I am focusing on 2.x changes. I'm not going to rush a change to suit the 1.3.0 timeline. Pekko 2.x is built with Java 17 making it easier to support and test Java Flow API integration.

I am aware, the point is not about rushing changes to suite a 1.3.x changeline but rather creating an ideal API design given that we have both reactive streams and java 9 flow

If that API design requires changing the current reactive-streams interopt then its possible to create the necessary changes (i.e. aliases/deprecations) in 1.3.0 branch, thats all

@mdedetrich
Copy link
Contributor

mdedetrich commented Nov 9, 2025

We already have JavaFlowSupport so I guess the feedback is that some people prefer that to us moving the code to Source and Sink directly.

If the classes/datastructure didn't have the exact same name for both reactive streams and java 9 flow I would have also preferred this option, but in both cases they are called Publishers/Subscribers which means that you would have source.asJava9Subscriber/source.asReactiveStreamsSubscriber which is a bit of a mouthful.

You could avoid this on the scaladsl side since you can use implicit classes/extension methods and you only import the support that you can are about (i.e. import java9FlowSupport vs import reactivestreams support) but that doesn't work for javadsl.

Another disadvantage of inlining the source is that its not modular, given that already have java 9 flow/reactive streams flow we could have yet another streaming interface in the future which we might want to implement. Hence another argument for having separate objects/classes with static methods that provide the conversion methods.

@pjfanning pjfanning modified the milestones: 2.0.0-M1, 2.0.0-M2 Nov 10, 2025
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

4 participants