feat: Add asJavaStream to Sink

Author: He-Pin

stream-tests/src/test/java/org/apache/pekko/stream/io/SinkAsJavaSourceTest.java

@@ -44,4 +44,12 @@ public void mustBeAbleToUseAsJavaStream() throws Exception {
     java.util.stream.Stream<Integer> javaStream = Source.from(list).runWith(streamSink, system);
     assertEquals(list, javaStream.collect(Collectors.toList()));
   }
+
+  @Test
+  public void mustBeAbleToUseAsJavaStreamOnSink() throws Exception {
+    final List<Integer> list = Arrays.asList(1, 2, 3);
+    java.util.stream.Stream<Integer> javaStream =
+        Source.from(list).runWith(Sink.asJavaStream(), system);
+    assertEquals(list, javaStream.collect(Collectors.toList()));
+  }
 }

stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/SinkAsJavaStreamSpec.scala

@@ -32,6 +32,9 @@ class SinkAsJavaStreamSpec extends StreamSpec(UnboundedMailboxConfig) {
     "work in happy case" in {
       val javaSource = Source(1 to 100).runWith(StreamConverters.asJavaStream())
       javaSource.count() should ===(100L)
+      //
+      Source(1 to 100).runWith(Sink.asJavaStream())
+        .count() should ===(100L)
     }
 
     "fail if parent stream is failed" in {

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
+   * configured through the ``pekko.stream.blocking-io-dispatcher``.
+   *
+   * @since 2.0.0
+   */
+  def asJavaStream[T](): Sink[T, java.util.stream.Stream[T]] = new Sink(scaladsl.StreamConverters.asJavaStream())
+
   /**
    * A `Sink` that will invoke the given procedure for each received element. The sink is materialized
    * into a [[java.util.concurrent.CompletionStage]] which will be completed with `Success` when reaching the

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
+   * configured through the ``pekko.stream.blocking-io-dispatcher``.
+   *
+   * @since 2.0.0
+   */
+  def asJavaStream[T](): Sink[T, java.util.stream.Stream[T]] = StreamConverters.asJavaStream()
+
   /**
    * A `Sink` that will consume the stream and discard the elements.
    */