#18962 MergeSorted and read() completion handling

akka-stream/src/main/scala/akka/stream/javadsl/SubSource.scala

@@ -15,6 +15,7 @@ import scala.annotation.unchecked.uncheckedVariance
 import scala.concurrent.Future
 import scala.concurrent.duration.FiniteDuration
 import akka.japi.Util
+import java.util.Comparator
 
 /**
  * A “stream of streams” sub-flow of data elements, e.g. produced by `groupBy`.
@@ -768,6 +769,24 @@ class SubSource[+Out, +Mat](delegate: scaladsl.SubFlow[Out, Mat, scaladsl.Source
   def interleave[T >: Out](that: Graph[SourceShape[T], _], segmentSize: Int): SubSource[T, Mat] =
     new SubSource(delegate.interleave(that, segmentSize))
 
+  /**
+   * Merge the given [[Source]] to this [[Source]], taking elements as they arrive from input streams,
+   * picking always the smallest of the available elements (waiting for one element from each side
+   * to be available). This means that possible contiguity of the input streams is not exploited to avoid
+   * waiting for elements, this merge will block when one of the inputs does not have more elements (and
+   * does not complete).
+   *
+   * '''Emits when''' all of the inputs have an element available
+   *
+   * '''Backpressures when''' downstream backpressures
+   *
+   * '''Completes when''' all upstreams complete
+   *
+   * '''Cancels when''' downstream cancels
+   */
+  def mergeSorted[U >: Out, M](that: Graph[SourceShape[U], M], comp: Comparator[U]): javadsl.SubSource[U, Mat] =
+    new SubSource(delegate.mergeSorted(that)(Ordering.comparatorToOrdering(comp)))
+
   /**
    * Combine the elements of current [[Flow]] and the given [[Source]] into a stream of tuples.
    *