System wide materializer (#27491)

Introduces a materializer started through an extension and then an implicit
conversion for Scala turning an implicitly available ActorSystem into a
materializer. The Java APIs has been ammended with run-methods accepting
an ActorSystem.

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

@@ -28,6 +28,7 @@ import java.util.concurrent.CompletionStage
 import java.util.concurrent.CompletableFuture
 import java.util.function.{ BiFunction, Supplier }
 
+import akka.actor.ClassicActorSystemProvider
 import akka.util.unused
 import com.github.ghik.silencer.silent
 
@@ -566,6 +567,20 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[
   /**
    * Materializes this Source, immediately returning (1) its materialized value, and (2) a new Source
    * that can be used to consume elements from the newly materialized Source.
+   *
+   * Note that the `ActorSystem` can be used as the `systemProvider` parameter.
+   */
+  def preMaterialize(systemProvider: ClassicActorSystemProvider)
+      : Pair[Mat @uncheckedVariance, Source[Out @uncheckedVariance, NotUsed]] = {
+    val (mat, src) = delegate.preMaterialize()(SystemMaterializer(systemProvider.classicSystem).materializer)
+    Pair(mat, new Source(src))
+  }
+
+  /**
+   * Materializes this Source, immediately returning (1) its materialized value, and (2) a new Source
+   * that can be used to consume elements from the newly materialized Source.
+   *
+   * Prefer the method taking an `ActorSystem` unless you have special requirements.
    */
   def preMaterialize(
       materializer: Materializer): Pair[Mat @uncheckedVariance, Source[Out @uncheckedVariance, NotUsed]] = {
@@ -662,6 +677,17 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[
   /**
    * Connect this `Source` to a `Sink` and run it. The returned value is the materialized value
    * of the `Sink`, e.g. the `Publisher` of a `Sink.asPublisher`.
+   *
+   * Note that the classic or typed `ActorSystem` can be used as the `systemProvider` parameter.
+   */
+  def runWith[M](sink: Graph[SinkShape[Out], M], systemProvider: ClassicActorSystemProvider): M =
+    delegate.runWith(sink)(SystemMaterializer(systemProvider.classicSystem).materializer)
+
+  /**
+   * Connect this `Source` to a `Sink` and run it. The returned value is the materialized value
+   * of the `Sink`, e.g. the `Publisher` of a `Sink.asPublisher`.
+   *
+   * Prefer the method taking an `ActorSystem` unless you have special requirements
    */
   def runWith[M](sink: Graph[SinkShape[Out], M], materializer: Materializer): M =
     delegate.runWith(sink)(materializer)
@@ -673,6 +699,24 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[
    * The returned [[java.util.concurrent.CompletionStage]] will be completed with value of the final
    * function evaluation when the input stream ends, or completed with `Failure`
    * if there is a failure is signaled in the stream.
+   *
+   * Note that the classic or typed `ActorSystem` can be used as the `systemProvider` parameter.
+   */
+  def runFold[U](
+      zero: U,
+      f: function.Function2[U, Out, U],
+      systemProvider: ClassicActorSystemProvider): CompletionStage[U] =
+    runWith(Sink.fold(zero, f), systemProvider)
+
+  /**
+   * Shortcut for running this `Source` with a fold function.
+   * The given function is invoked for every received element, giving it its previous
+   * output (or the given `zero` value) and the element as input.
+   * The returned [[java.util.concurrent.CompletionStage]] will be completed with value of the final
+   * function evaluation when the input stream ends, or completed with `Failure`
+   * if there is a failure is signaled in the stream.
+   *
+   * Prefer the method taking an ActorSystem unless you have special requirements.
    */
   def runFold[U](zero: U, f: function.Function2[U, Out, U], materializer: Materializer): CompletionStage[U] =
     runWith(Sink.fold(zero, f), materializer)
@@ -684,6 +728,23 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[
    * The returned [[java.util.concurrent.CompletionStage]] will be completed with value of the final
    * function evaluation when the input stream ends, or completed with `Failure`
    * if there is a failure is signaled in the stream.
+   *
+   * Note that the classic or typed `ActorSystem` can be used as the `systemProvider` parameter.
+   */
+  def runFoldAsync[U](
+      zero: U,
+      f: function.Function2[U, Out, CompletionStage[U]],
+      systemProvider: ClassicActorSystemProvider): CompletionStage[U] = runWith(Sink.foldAsync(zero, f), systemProvider)
+
+  /**
+   * Shortcut for running this `Source` with an asynchronous fold function.
+   * The given function is invoked for every received element, giving it its previous
+   * output (or the given `zero` value) and the element as input.
+   * The returned [[java.util.concurrent.CompletionStage]] will be completed with value of the final
+   * function evaluation when the input stream ends, or completed with `Failure`
+   * if there is a failure is signaled in the stream.
+   *
+   * Prefer the method taking an `ActorSystem` unless you have special requirements
    */
   def runFoldAsync[U](
       zero: U,
@@ -702,6 +763,28 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[
    * the reduce operator will fail its downstream with a [[NoSuchElementException]],
    * which is semantically in-line with that Scala's standard library collections
    * do in such situations.
+   *
+   * Note that the classic or typed `ActorSystem` can be used as the `systemProvider` parameter.
+   */
+  def runReduce(
+      f: function.Function2[Out, Out, Out],
+      systemProvider: ClassicActorSystemProvider): CompletionStage[Out] =
+    runWith(Sink.reduce(f), systemProvider.classicSystem)
+
+  /**
+   * Shortcut for running this `Source` with a reduce function.
+   * The given function is invoked for every received element, giving it its previous
+   * output (from the second ones) an the element as input.
+   * The returned [[java.util.concurrent.CompletionStage]] will be completed with value of the final
+   * function evaluation when the input stream ends, or completed with `Failure`
+   * if there is a failure is signaled in the stream.
+   *
+   * If the stream is empty (i.e. completes before signalling any elements),
+   * the reduce operator will fail its downstream with a [[NoSuchElementException]],
+   * which is semantically in-line with that Scala's standard library collections
+   * do in such situations.
+   *
+   * Prefer the method taking an `ActorSystem` unless you have special requirements
    */
   def runReduce(f: function.Function2[Out, Out, Out], materializer: Materializer): CompletionStage[Out] =
     runWith(Sink.reduce(f), materializer)
@@ -1387,6 +1470,20 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[
    * The returned [[java.util.concurrent.CompletionStage]] will be completed normally when reaching the
    * normal end of the stream, or completed exceptionally if there is a failure is signaled in
    * the stream.
+   *
+   * Note that the classic or typed `ActorSystem` can be used as the `systemProvider` parameter.
+   */
+  def runForeach(f: function.Procedure[Out], systemProvider: ClassicActorSystemProvider): CompletionStage[Done] =
+    runWith(Sink.foreach(f), systemProvider)
+
+  /**
+   * Shortcut for running this `Source` with a foreach procedure. The given procedure is invoked
+   * for each received element.
+   * The returned [[java.util.concurrent.CompletionStage]] will be completed normally when reaching the
+   * normal end of the stream, or completed exceptionally if there is a failure is signaled in
+   * the stream.
+   *
+   * Prefer the method taking an `ActorSystem` unless you have special requirements
    */
   def runForeach(f: function.Procedure[Out], materializer: Materializer): CompletionStage[Done] =
     runWith(Sink.foreach(f), materializer)