You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@pekko.apache.org by "He-Pin (via GitHub)" <gi...@apache.org> on 2025/09/06 09:09:59 UTC
[PR] feat: Add asJavaStream to Sink [pekko]
He-Pin opened a new pull request, #2156:
URL: https://github.com/apache/pekko/pull/2156
Motivation:
Make code much easy to find for daily job.
refs: https://smallrye.io/smallrye-mutiny/latest/guides/reactive-to-imperative/#iterating-over-multis-items
which has a `asStream` on Mutiny.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add asJavaStream to Sink [pekko]
Posted by "mdedetrich (via GitHub)" <gi...@apache.org>.
mdedetrich commented on PR #2156:
URL: https://github.com/apache/pekko/pull/2156#issuecomment-3261769621
> 1. There's already a declaration here.
Yes the PR exposes Java 1.8 Stream in a `Sink` so it should be fine, but the same pitfalls with the other stream converter methods also needs to be documented. I'll check in more detail when I have access to laptop.
> 2. Java 25 is coming soon, and managed blocking is a thing of the past.
Irrelevant, not everyone is running on JVM 25 and people run Pekko in other environments.
> 3. The API needs to be kept simple and reasonable. If you think this is wrong, why does smallrye-mutiny have an `asStream` method? and `reactor.core.publisher.Flux#toStream()`
I don't know how mutiny works. All I am saying is that a core design of Pekko Streams specifically is that it's built on the assumption that everything is purely asynchronous which is why as I explained before that materialized values/stream elements can never block and since we can't enforce that users will always 100% of the time use `VirtualThread` we have to deal with this.
Unfortunately this was the issue with Java not having a core async/concurrency primitive when it was releases (in contrast to Kotkin which has coroutines and Go which has channels) and it takes 30 years to release an officially supported one (although you can argue that `CompletionStage` is that primitive)
That actually opens up another set of questions, we may be able to simplify the current stream converters API by wrapping the results in a `CompletionStage`. That way the API is easier to work with (the current design of it being in a `Sink` is a pita to work with) and we may be able to avoid these deadlock issues.
I am not against have another API which is designed for `VirtualThread` and is even simpler, it would just have to assert that it's being run on a `VirtualThread` and crash with a helpful error message if it's not
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add asJavaStream to Sink [pekko]
Posted by "mdedetrich (via GitHub)" <gi...@apache.org>.
mdedetrich commented on PR #2156:
URL: https://github.com/apache/pekko/pull/2156#issuecomment-3261796596
> Kotlin's coroutines are fake, even worse than Java 21's virtual threads. I added this API quite cautiously. You see, I didn't add toOutputStream to Sink. If other libraries had these features, a new user might see that Pekko Stream doesn't have this feature and choose another library.
I don't know what you mean by fake, but coroutines are by far the best implementation of an async concurrent primitive out of any of the JVM languages. Its performance is far higher than Loom as coroutines just compile do jump statements in JVM bytecode, it supports cancellation of resources and to boot it's also supported on all other kotlin platforms (native/js) with zero performance overhead.
Loom on the hand only works on newer JVMs and is slower, the only good thing about it is that it uses the same programming model as threads but that is also what makes it slow (i.e. it needs to maintain stack traces).
Akka/Pekko actors are ironically faster than Loom as a concurrency primitive as you only pass around pure values.
>
> It seems you're planning to build a Kotlin-specific DSL for Pekko. While that's great, it also presents significant challenges, as it will have to compete with Kotlin's Flow API. It's a real challenge to have something that others don't have, and to be superior to others.
I am planning to build a kotlin but what I am saying is not specifically because of this, it's an issue with Java as a language.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add asJavaStream to Sink [pekko]
Posted by "mdedetrich (via GitHub)" <gi...@apache.org>.
mdedetrich commented on PR #2156:
URL: https://github.com/apache/pekko/pull/2156#issuecomment-3261945680
Oh and on yet another note, those benchmarks at https://github.com/LesnyRumcajs/grpc_bench/discussions/441 show akka being slightly faster than pekko, which means there are more performance improvements on the table for pekko. It might even be possible to make akka/pekko beat vert.x with new JVM capabilities (although we would have to use multi jar for that)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add asJavaStream to Sink [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #2156:
URL: https://github.com/apache/pekko/pull/2156#issuecomment-3261742487
1. There's already a declaration here.
2. Java 25 is coming soon, and managed blocking is a thing of the past.
3. The API needs to be kept simple and reasonable. If you think this is wrong, why does smallrye-mutiny have an `asStream` method?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add asJavaStream to Sink [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #2156:
URL: https://github.com/apache/pekko/pull/2156#issuecomment-3261810408
1. Is that true? then why it's not the 1st of the JVM in https://github.com/LesnyRumcajs/grpc_bench/discussions/441 ?
2. it will require a Dispatchers.io to run the blocking code, let kotlin to handle old java file and Socket, its slow.
3. what's the color your function problem is a no for me, that's why loom is much better than kotlin one.
4. Akka/Pekko will be slow if it's been blocked than loom, the only way is running the dispatcher with Loom.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add asJavaStream to Sink [pekko]
Posted by "raboof (via GitHub)" <gi...@apache.org>.
raboof commented on PR #2156:
URL: https://github.com/apache/pekko/pull/2156#issuecomment-3261727713
> > The warning there (which is also in your javadoc/scaladoc) is that this will use blocking operations. That could be a reason to discourage this API, and putting them in the separate `StreamConverters` might be a good way to do that. So from that perspective, perhaps having them in `StreamConverters` (and thus harder to find) might be a good thing?
>
> I would be against removing these blocking methods, as I have in the past used the frequently with other API's that only work with `InputStream`/`OutputStream` which is very common in Java land.
I'm definitely not suggesting removing the blocking methods, just 'not moving them to Sink'. I guess I should have said "keeping them in the separate `StreamConverters`" rather than "putting them in the separate `StreamConverters`" ;)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add asJavaStream to Sink [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #2156:
URL: https://github.com/apache/pekko/pull/2156#issuecomment-3261692700
I think the `asStream` can be non-blocking, only triggering a pull when there is a `hasNext` request, and letting the poller who consumes the Java stream wait on the condition, the current is a bit complex but still never block the queue, but the poller, so this method should be safe.
Does this method live outside before Java 8?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add asJavaStream to Sink [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #2156:
URL: https://github.com/apache/pekko/pull/2156#issuecomment-3449700733
https://github.com/ReactiveX/RxJava/wiki/What's-different-in-3.0#blockingstream
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add asJavaStream to Sink [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #2156:
URL: https://github.com/apache/pekko/pull/2156#issuecomment-3262106946
Btw, there are already a `Source#fromJavaStream` so it's fair to have a `Sink#asJavaStream` too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add asJavaStream to Sink [pekko]
Posted by "mdedetrich (via GitHub)" <gi...@apache.org>.
mdedetrich commented on PR #2156:
URL: https://github.com/apache/pekko/pull/2156#issuecomment-3261704960
> This definitely makes it easier to find.
>
> This does beg the question whether we should move and deprecate all methods in https://pekko.apache.org/docs/pekko/current/stream/operators/index.html#additional-sink-and-source-converters .
>
> The warning there (which is also in your javadoc/scaladoc) is that this will use blocking operations. That could be a reason to discourage this API, and putting them in the separate `StreamConverters` might be a good way to do that. So from that perspective, perhaps having them in `StreamConverters` (and thus harder to find) might be a good thing?
I would be against removing these blocking methods, as I have in the past used the frequently with other API's that only work with `InputStream`/`OutputStream` which is very common in Java land.
There is nothing you can do about these methods being blocking as the underlying `InputStream`/`OutputStream` are also blocking, it is what it is. There are the annoying gotcha's, i.e. its easy to make deadlocks because actor stream materialization cannot be blocking but this is the price we have to pay when working with Java API's
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add asJavaStream to Sink [pekko]
Posted by "mdedetrich (via GitHub)" <gi...@apache.org>.
mdedetrich commented on PR #2156:
URL: https://github.com/apache/pekko/pull/2156#issuecomment-3265934670
> The current implementation uses `Await.result` in `Iterator.hasNext`. It might be possible to create an implementation that avoids blocking - or at least avoids it as long as the user doesn't explicitly trigger operations that are blocking 'by nature'. I think we should do that before adding it to `Sink`.
I would go further and say that `Await.result` should never be used in production code with the only real exception I can think of as being shut down of the main application.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add asJavaStream to Sink [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #2156:
URL: https://github.com/apache/pekko/pull/2156#issuecomment-3261697973
@raboof As a Java developer, I think that's bad to put them on StreamConverters, users need to remember too many, today, LLM can only handle less mcp tools and mcp servers, so does humans.
I think Pekko should make code smooth brain just as ZIO and Spring :), which actually will cover 80% the daily usage.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add asJavaStream to Sink [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #2156:
URL: https://github.com/apache/pekko/pull/2156#issuecomment-3261784640
Kotlin's coroutines are fake, even worse than Java 21's virtual threads. I added this API quite cautiously. You see, I didn't add toOutputStream to Sink. If other libraries had these features, a new user might see that Pekko Stream doesn't have this feature and choose another library.
It seems you're planning to build a Kotlin-specific DSL for Pekko. While that's great, it also presents significant challenges, as it will have to compete with Kotlin's Flow API. It's a real challenge to have something that others don't have, and to be superior to others.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add asJavaStream to Sink [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #2156:
URL: https://github.com/apache/pekko/pull/2156#issuecomment-3261738859
When you are running inside a Virtual thread, and then you are blocked by, let's say, a Lock, and then the virtual thread is parked and unmounted, so the carrier thread can execute other things, has the virtual thread been blocked？ yes, it cann't make any progress until it is unparked, and no, the carrier thread is still running, just some memory copying.
You can try this with Pekko 1.2, set the max thread count to `1` and turn on `virualization = on`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add asJavaStream to Sink [pekko]
Posted by "raboof (via GitHub)" <gi...@apache.org>.
raboof commented on PR #2156:
URL: https://github.com/apache/pekko/pull/2156#issuecomment-3265929548
The current implementation uses `Await.result` in `Iterator.hasNext`. It might be possible to create an implementation that avoids blocking - or at least avoids it as long as the user doesn't explicitly trigger operations that are blocking 'by nature'. I think we should do that before adding it to `Sink`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add asJavaStream to Sink [pekko]
Posted by "mdedetrich (via GitHub)" <gi...@apache.org>.
mdedetrich commented on PR #2156:
URL: https://github.com/apache/pekko/pull/2156#issuecomment-3261740509
> When you are running inside a Virtual thread, and then you are blocked by, let's say, a Lock, and then the virtual thread is parked and unmounted, so the carrier thread can execute other things, has the virtual thread been blocked？ yes, it cann't make any progress until it is unparked, and no, the carrier thread is still running, just some memory copying.
>
> You can try this with Pekko 1.2, set the max thread count to `1` and turn on `virualization = on`.
I am aware that Virtual Threads solve the issues, the point is we don't know and we cannot force people to always use VirtualThreads so the API cannot be designed assuming that people will always use a `VirtualThread`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add asJavaStream to Sink [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #2156:
URL: https://github.com/apache/pekko/pull/2156#issuecomment-3261796658
If we have an `AsyncStream/AsyncQueue`, then the problem can be simpler, which pulls always return a Future.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add asJavaStream to Sink [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #2156:
URL: https://github.com/apache/pekko/pull/2156#issuecomment-3261960944
@mdedetrich maybe a new run will show some different.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add asJavaStream to Sink [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #2156:
URL: https://github.com/apache/pekko/pull/2156#issuecomment-3265971607
@mdedetrich The Materializer itself is using the `Await.result/ready`:)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add asJavaStream to Sink [pekko]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on code in PR #2156:
URL: https://github.com/apache/pekko/pull/2156#discussion_r2326706242
##########
stream/src/main/scala/org/apache/pekko/stream/javadsl/Sink.scala:
##########
@@ -212,6 +212,21 @@ object Sink {
def asPublisher[T](fanout: AsPublisher): Sink[T, Publisher[T]] =
new Sink(scaladsl.Sink.asPublisher(fanout == AsPublisher.WITH_FANOUT))
+ /**
+ * Creates a sink which materializes into Java 8 ``Stream`` that can be run to trigger demand through the sink.
+ * Elements emitted through the stream will be available for reading through the Java 8 ``Stream``.
+ *
+ * The Java 8 ``Stream`` will be ended when the stream flowing into this ``Sink`` completes, and closing the Java
+ * ``Stream`` will cancel the inflow of this ``Sink``.
+ *
+ * Java 8 ``Stream`` throws exception in case reactive stream failed.
+ *
+ * Be aware that Java ``Stream`` blocks current thread while waiting on next element from downstream.
+ * As it is interacting wit blocking API the implementation runs on a separate dispatcher
+ * configured through the ``pekko.stream.blocking-io-dispatcher``.
+ */
Review Comment:
add `@since 2.0.0`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add asJavaStream to Sink [pekko]
Posted by "mdedetrich (via GitHub)" <gi...@apache.org>.
mdedetrich commented on PR #2156:
URL: https://github.com/apache/pekko/pull/2156#issuecomment-3261738637
> @mdedetrich You can give it a try, this code will work ok if the event loop is a virtual thread; otherwise, dead deadlock. [netty/netty#15405](https://github.com/netty/netty/discussions/15405)
I know that it won't block if you use a `VirtualThread` but we cannot force people to use `VirtualThread` so if it blocks on standard usage (which it does) then its as unsafe as the other stream converters and for this reason you need to follow the same pattern, i.e. only give the reference to a Java 1.8 Stream as a final value in a `Sink` otherwise people will almost always deadlock themselves when using the API.
There is a reason why the current StreamCoverters are designed the way they are.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add asJavaStream to Sink [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #2156:
URL: https://github.com/apache/pekko/pull/2156#issuecomment-3261814692
We are using Kotlin at work, some part of our server and our ios and android sharing the same codebase, but on server side, mostly using Java 21 nowadays.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add asJavaStream to Sink [pekko]
Posted by "raboof (via GitHub)" <gi...@apache.org>.
raboof commented on PR #2156:
URL: https://github.com/apache/pekko/pull/2156#issuecomment-3261703293
> I think the `asStream` can be non-blocking, only triggering a pull when there is a `hasNext` request, and letting the poller who consumes the Java stream wait on the condition, the current is a bit complex but still never block the queue, but the poller, so this method should be safe.
Gotcha, so `asInputStream` and `asOutputStream` are dangerous but `asJavaStream` is safe? That might be a good reason to expose it directly on `Sink`.
I see the original `asJavaStream` already had the warning "Be aware that Java Stream blocks current thread while waiting on next element from downstream.". That sounds confusing to me. Is it waiting for an element from upstream, or demand from downstream?
I see you have added "As it is interacting wit blocking API the implementation runs on a separate dispatcher configured through the ``pekko.stream.blocking-io-dispatcher``." to the javadoc/scaladoc, but which implementation is that?
> I think Pekko should make code smooth brain just as ZIO and Spring :), which actually will cover 80% the daily usage.
I agree, we should be making things "as simple as possible, but not simpler" :smile:
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add asJavaStream to Sink [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #2156:
URL: https://github.com/apache/pekko/pull/2156#issuecomment-3261715949
We are currently using Spring-Ai and MCP SDK at work, which are totally built the the reactor-core. I always wondered, why can't it be Pekko-stream.
And nowadays, the virtual thread makes direct style programming possible, see how ox-stream and ox-actor are being implemented, yes, the blocking should be managed, but may not be a problem when you are running the actor/stream with virtual thread.
There is a Jox library being implemented with pure java, so I think Pekko Stream should be quite simple and friendly for Java developers too.
<img width="1213" height="507" alt="image" src="https://github.com/user-attachments/assets/d0ffe44f-46d2-4608-a3c5-39e63f037c92" />
This code is from a system where we leverage pekko stream to do translation for Taobao English, you can see the long
```
withAttributes(
ActorAttributes.withSupervisionStrategy(
(org.apache.pekko.japi.function.Function<Throwable, Supervision.Directive>) param -> {
log.error("AsyncOvsTranslator error", param);
return (Supervision.Directive) Supervision.resume();
}))
```
In Reactor-core will just be an `onErrorContinue`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add asJavaStream to Sink [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #2156:
URL: https://github.com/apache/pekko/pull/2156#issuecomment-3261732582
@mdedetrich You can give it a try, this code will work ok if the event loop is a virtual thread; otherwise, dead deadlock.
https://github.com/netty/netty/discussions/15405
At work, Ajdk( a JDK backed on Dragonwell) 21 backported and fixed many issues around Loom, and now we can safely use Loom. I do encounter some deadlocks when using Loom, but now it's all fixed, I think that's why OX is starting direct style.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add asJavaStream to Sink [pekko]
Posted by "mdedetrich (via GitHub)" <gi...@apache.org>.
mdedetrich commented on PR #2156:
URL: https://github.com/apache/pekko/pull/2156#issuecomment-3261920824
> 1. Is that true? then why it's not the 1st of the JVM in [2024-04-13 i9-13900KF (32) @ 5.500GHz, 64G RAM LesnyRumcajs/grpc_bench#441](https://github.com/LesnyRumcajs/grpc_bench/discussions/441) ?
I don't know, I would have to look into it. I am saying that the co-routine compiler transformations are well known and understood and they just translate to raw JVM bytecode statements `jsr`/`goto`, you can't really beat that.
Even though its not really used, scala had its own coroutine plugin as part of the compiler and it also did the same.
As you said, coroutines are slower if they have to interact with blocking Java code and kotlin-grpc might be doing that. On the other hand, if you have a purely async non blocking implementation like akka/pekko then it should be as fast, if not faster and actually your benchmarks show that akka/pekko grpc is just as fast as java vertx and the fact that java vertx is not meaningfully faster than akka/pekko grpc actually shows how Loom is not that fast as Loom has full runtime support in the JVM where as pekko/akka does not, as it implements actors as a library. If JVM implemented actors natively, then pekko/actor would be much faster, moerso than loom. There are lots of tricks that you can use to make actors very fast which Erlang does but cannot be done with akka/pekko as we don't have full JVM support (we however might be able to make actors even faster with newer versions of JVM as they are giving us more tools).
> 2. it will require a Dispatchers.io to run the blocking code, let kotlin to handle old java file and Socket, its slow.
Sure if you need to interact with old blocking Java code it will be slower, the solution here is to not write/use blocking code. Thats why akka/pekko itself was written completely with async in mind and also why its as fast as java vertx 😉
> 3. what's the color your function problem is a no for me; that's why Loom is much better than kotlin one.
True, I am just saying there are tradeoffs. If you don't want to deal with the color function problem then you can use loom, I am just saying not everyone is in that boat.
> 4. Akka/Pekko will be slow if it's been blocked than loom, the only way is running the dispatcher with Loom.
>
> We are using Kotlin at work, some part of our server and our ios and android sharing the same codebase, but on server side, mostly using Java 21 nowadays.
We are going on a bit of a tangent here though, my point is that not everyone who uses akka/pekko is using loom and akka/pekko is not a loom only API design. As I said, I don't have issues with making an API specific for loom users if it makes their life simpler from an API perspective, its just that we need to check for that because if you don't happen to be using loom then the code will just deadlock and thats a terrible user experience.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add asJavaStream to Sink [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #2156:
URL: https://github.com/apache/pekko/pull/2156#issuecomment-3265981387
I did not see much deadlock, because of the blocking method on Reactor-core/Rxjava which have a larger java users base.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add asJavaStream to Sink [pekko]
Posted by "mdedetrich (via GitHub)" <gi...@apache.org>.
mdedetrich commented on PR #2156:
URL: https://github.com/apache/pekko/pull/2156#issuecomment-3265970623
Actually now that I think about it, if its not possible to implement this function without blocking (i.e. `Await.result`) it might make sense to do what I asked before and only have this method available for loom/`VirtualThread` users, otherwise its way too easy to deadlock
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add asJavaStream to Sink [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #2156:
URL: https://github.com/apache/pekko/pull/2156#issuecomment-3265969860
But reactor-core has that on Flux and no problem.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add asJavaStream to Sink [pekko]
Posted by "mdedetrich (via GitHub)" <gi...@apache.org>.
mdedetrich commented on PR #2156:
URL: https://github.com/apache/pekko/pull/2156#issuecomment-3261720137
> We are currently using Spring-Ai and MCP SDK at work, which are totally built the the reactor-core. I always wondered, why can't it be Pekko-stream.
>
> And nowadays, the virtual thread makes direct style programming possible, see how ox-stream and ox-actor are being implemented, yes, the blocking should be managed, but may not be a problem when you are running the actor/stream with virtual thread.
>
> There is a Jox library being implemented with pure java, so I think Pekko Stream should be quite simple and friendly for Java developers too. <img alt="image" width="1213" height="507" src="https://private-user-images.githubusercontent.com/501740/486385462-d0ffe44f-46d2-4608-a3c5-39e63f037c92.png?jwt=eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpc3MiOiJnaXRodWIuY29tIiwiYXVkIjoicmF3LmdpdGh1YnVzZXJjb250ZW50LmNvbSIsImtleSI6ImtleTUiLCJleHAiOjE3NTcxNTI0MzEsIm5iZiI6MTc1NzE1MjEzMSwicGF0aCI6Ii81MDE3NDAvNDg2Mzg1NDYyLWQwZmZlNDRmLTQ2ZDItNDYwOC1hM2M1LTM5ZTYzZjAzN2M5Mi5wbmc_WC1BbXotQWxnb3JpdGhtPUFXUzQtSE1BQy1TSEEyNTYmWC1BbXotQ3JlZGVudGlhbD1BS0lBVkNPRFlMU0E1M1BRSzRaQSUyRjIwMjUwOTA2JTJGdXMtZWFzdC0xJTJGczMlMkZhd3M0X3JlcXVlc3QmWC1BbXotRGF0ZT0yMDI1MDkwNlQwOTQ4NTFaJlgtQW16LUV4cGlyZXM9MzAwJlgtQW16LVNpZ25hdHVyZT1jNTkyOWIzMmQzNTE5ODZjMmMzM2M2Yzg0ODQxZGFmYjExN2YzN2M3OWE4ZmFiNWMyYzMzZWU3ODZjNTA1YjczJlgtQW16LVNpZ25lZEhlYWRlcnM9aG9zdCJ9.eqERrPxqxHH6ATU3RhKPWbFPLey4h7bsvPfyEQ2kJIQ"> This code is from a system
where we leverage pekko stream to do translation for Taobao English, you can see the long
>
> ```
> withAttributes(
> ActorAttributes.withSupervisionStrategy(
> (org.apache.pekko.japi.function.Function<Throwable, Supervision.Directive>) param -> {
> log.error("AsyncOvsTranslator error", param);
> return (Supervision.Directive) Supervision.resume();
> }))
> ```
>
> In Reactor-core will just be an `onErrorContinue`.
Whether you use loom or not is not relevant here, the question is does the new Java 1.8 stream block the thread while its waiting for a new element? If it does (which I do think is the case as I see no evidence from the API that its properly async), then its just as dangerous as the `InputStream`/`OutputStream` converters.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add asJavaStream to Sink [pekko]
Posted by "mdedetrich (via GitHub)" <gi...@apache.org>.
mdedetrich commented on PR #2156:
URL: https://github.com/apache/pekko/pull/2156#issuecomment-3261708862
> I see the original `asJavaStream` already had the warning "Be aware that Java Stream blocks current thread while waiting on next element from downstream.". That sounds confusing to me. Is it waiting for an element from upstream, or demand from downstream?
The Java Stream itself blocks while its waiting for the next element (i.e. it stalls the current thread) which means its easy to make a deadlock if you expose the `InputStream`/`OutputStream` as a materialized element, as the materialized element needs to evaluate immediately (i.e. no blocking) before the stream starts but the pekko stream cannot start because the `InputStream` blocks until a receives the element (which it will never receive as the pekko stream hasn't started yet), hence the deadlock.
I experienced this myself, not sure what the easiest way to deal with this is from an API don't shoot yourself in the foot perspective.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add asJavaStream to Sink [pekko]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2156:
URL: https://github.com/apache/pekko/pull/2156#discussion_r2326923278
##########
stream/src/main/scala/org/apache/pekko/stream/javadsl/Sink.scala:
##########
@@ -212,6 +212,23 @@ object Sink {
def asPublisher[T](fanout: AsPublisher): Sink[T, Publisher[T]] =
new Sink(scaladsl.Sink.asPublisher(fanout == AsPublisher.WITH_FANOUT))
+ /**
+ * Creates a sink which materializes into Java 8 ``Stream`` that can be run to trigger demand through the sink.
+ * Elements emitted through the stream will be available for reading through the Java 8 ``Stream``.
+ *
+ * The Java 8 ``Stream`` will be ended when the stream flowing into this ``Sink`` completes, and closing the Java
+ * ``Stream`` will cancel the inflow of this ``Sink``.
+ *
+ * Java 8 ``Stream`` throws exception in case reactive stream failed.
+ *
+ * Be aware that Java ``Stream`` blocks current thread while waiting on next element from downstream.
+ * As it is interacting wit blocking API the implementation runs on a separate dispatcher
Review Comment:
There's a typo in the comment: 'wit' should be 'with'.
```suggestion
* As it is interacting with blocking API the implementation runs on a separate dispatcher
```
##########
stream/src/main/scala/org/apache/pekko/stream/scaladsl/Sink.scala:
##########
@@ -292,6 +292,23 @@ object Sink {
if (fanout) new FanoutPublisherSink[T](DefaultAttributes.fanoutPublisherSink, shape("FanoutPublisherSink"))
else new PublisherSink[T](DefaultAttributes.publisherSink, shape("PublisherSink")))
+ /**
+ * Creates a sink which materializes into Java 8 ``Stream`` that can be run to trigger demand through the sink.
+ * Elements emitted through the stream will be available for reading through the Java 8 ``Stream``.
+ *
+ * The Java 8 ``Stream`` will be ended when the stream flowing into this ``Sink`` completes, and closing the Java
+ * ``Stream`` will cancel the inflow of this ``Sink``.
+ *
+ * If the Java 8 ``Stream`` throws exception the Pekko stream is cancelled.
+ *
+ * Be aware that Java ``Stream`` blocks current thread while waiting on next element from downstream.
+ * As it is interacting wit blocking API the implementation runs on a separate dispatcher
Review Comment:
There's a typo in the comment: 'wit' should be 'with'.
```suggestion
* As it is interacting with blocking API the implementation runs on a separate dispatcher
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org