This is an automated email from the ASF dual-hosted git repository.
fanningpj pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new a869875329 add Source.apply that is optimised for Seq (#2562)
a869875329 is described below
commit a8698753298132c86b2c55b6783928945b99badb
Author: PJ Fanning <[email protected]>
AuthorDate: Wed Dec 10 11:03:28 2025 +0100
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]>
---
.../pekko/stream/DslFactoriesConsistencySpec.scala | 6 ++++++
.../pekko/stream/impl/TraversalBuilderSpec.scala | 15 +++++++++++++-
.../org/apache/pekko/stream/scaladsl/Source.scala | 24 +++++++++++++++++++++-
3 files changed, 43 insertions(+), 2 deletions(-)
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/DslFactoriesConsistencySpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/DslFactoriesConsistencySpec.scala
index 8aa9aee2c2..ff480505ec 100644
---
a/stream-tests/src/test/scala/org/apache/pekko/stream/DslFactoriesConsistencySpec.scala
+++
b/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),
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/impl/TraversalBuilderSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/impl/TraversalBuilderSpec.scala
index 36e91364fe..ae0a154061 100644
---
a/stream-tests/src/test/scala/org/apache/pekko/stream/impl/TraversalBuilderSpec.scala
+++
b/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)
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Source.scala
b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Source.scala
index c08b7de18a..1cc0b59eef 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Source.scala
+++ b/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.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
