add Source.apply that is optimised for Seq (#2562)

* add Source.apply that is optimised for Seq

* Update TraversalBuilderSpec.scala

* Update stream/src/main/scala/org/apache/pekko/stream/scaladsl/Source.scala

Co-authored-by: Copilot <[email protected]>

* Update DslFactoriesConsistencySpec.scala

---------

Co-authored-by: Copilot <[email protected]>

Author: pjfanning

stream-tests/src/test/scala/org/apache/pekko/stream/DslFactoriesConsistencySpec.scala

@@ -161,6 +161,12 @@ class DslFactoriesConsistencySpec extends AnyWordSpec with Matchers {
         _ == "apply",
         _ == 1,
         _ == List(classOf[pekko.stream.impl.SourceModule[_, _]])),
+      // no Java equivalent for this Scala only convenience method
+      Ignore(
+        _ == pekko.stream.scaladsl.Source.getClass,
+        _ == "apply",
+        _ == 1,
+        _ == List(classOf[scala.collection.immutable.Seq[_]])),
       // corresponding matches on java side would need to have Function23
       Ignore(_ == pekko.stream.scaladsl.Source.getClass, _ == "apply", _ == 24, _ => true),
       Ignore(_ == pekko.stream.scaladsl.Flow.getClass, _ == "apply", _ == 24, _ => true),

stream-tests/src/test/scala/org/apache/pekko/stream/impl/TraversalBuilderSpec.scala

@@ -521,7 +521,20 @@ class TraversalBuilderSpec extends PekkoSpec {
   }
 
   "find Source.iterable via TraversalBuilder with getValuePresentedSource" in {
-    val iterable = List("a")
+    val iterable = Set("a", "b", "c")
+    TraversalBuilder.getValuePresentedSource(Source(iterable)).get.asInstanceOf[IterableSource[
+      String]].elements should ===(
+      iterable)
+    val iterableSource = new IterableSource(iterable)
+    TraversalBuilder.getValuePresentedSource(iterableSource) should be(OptionVal.Some(iterableSource))
+
+    TraversalBuilder.getValuePresentedSource(Source(iterable).async) should be(OptionVal.None)
+    TraversalBuilder.getValuePresentedSource(Source(iterable).mapMaterializedValue(_ => "Mat")) should be(
+      OptionVal.None)
+  }
+
+  "find Source.iterable via TraversalBuilder with getValuePresentedSource (Seq input)" in {
+    val iterable = Seq("a", "b", "c")
     TraversalBuilder.getValuePresentedSource(Source(iterable)).get.asInstanceOf[IterableSource[
       String]].elements should ===(
       iterable)

stream/src/main/scala/org/apache/pekko/stream/scaladsl/Source.scala

@@ -395,14 +395,16 @@ object Source {
 
   /**
    * Helper to create [[Source]] from `Iterable`.
-   * Example usage: `Source(Seq(1,2,3))`
+   * Example usage: `Source(Set(1,2,3))`
    *
    * Starts a new `Source` from the given `Iterable`. This is like starting from an
    * Iterator, but every Subscriber directly attached to the Publisher of this
    * stream will see an individual flow of elements (always starting from the
    * beginning) regardless of when they subscribed.
+   * @see [[apply(immutable.Seq)]]
    */
   def apply[T](iterable: immutable.Iterable[T]): Source[T, NotUsed] = {
+    // unknown size is -1
     (iterable.knownSize: @switch) match {
       case 0 => empty
       case 1 => single(iterable.head)
@@ -411,6 +413,26 @@ object Source {
     }
   }
 
+  /**
+   * Helper to create [[Source]] from `Seq`.
+   * Example usage: `Source(Seq(1,2,3))`
+   *
+   * Starts a new `Source` from the given `Seq`. This is like starting from an
+   * Iterator, but every Subscriber directly attached to the Publisher of this
+   * stream will see an individual flow of elements (always starting from the
+   * beginning) regardless of when they subscribed.
+   * @see [[apply(immutable.Iterable)]]
+   * @since 2.0.0
+   */
+  def apply[T](seq: immutable.Seq[T]): Source[T, NotUsed] = {
+    seq match {
+      case immutable.Seq()                   => empty[T]
+      case immutable.Seq(elem: T @unchecked) => single(elem)
+      case _                                 =>
+        fromGraph(new IterableSource[T](seq)).withAttributes(DefaultAttributes.iterableSource)
+    }
+  }
+
   /**
    * Creates a `Source` from an array, if the array is empty, the stream is completed immediately,
    * otherwise, every element of the array will be emitted sequentially.