You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pekko.apache.org by he...@apache.org on 2025/12/14 11:13:08 UTC
(pekko) branch main updated: chore: optimize Source#future and Source#futureSource (#2560)
This is an automated email from the ASF dual-hosted git repository.
hepin 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 01f2265fb2 chore: optimize Source#future and Source#futureSource (#2560)
01f2265fb2 is described below
commit 01f2265fb2d69e4e82924680740d1b0e3e212f10
Author: He-Pin(kerr) <he...@gmail.com>
AuthorDate: Sun Dec 14 19:13:03 2025 +0800
chore: optimize Source#future and Source#futureSource (#2560)
* chore: optimize Source#future and Source#futureSource
---------
Co-authored-by: PJ Fanning <pj...@users.noreply.github.com>
---
.../pekko/stream/impl/TraversalBuilderSpec.scala | 5 +-
.../apache/pekko/stream/scaladsl/SourceSpec.scala | 83 +++++++++++++++++++++-
.../org/apache/pekko/stream/scaladsl/Source.scala | 18 +++--
3 files changed, 99 insertions(+), 7 deletions(-)
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 ae0a154061..2af4b9b0a7 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
@@ -13,7 +13,7 @@
package org.apache.pekko.stream.impl
-import scala.concurrent.Future
+import scala.concurrent.Promise
import org.apache.pekko
import pekko.NotUsed
@@ -508,7 +508,8 @@ class TraversalBuilderSpec extends PekkoSpec {
}
"find Source.future via TraversalBuilder with getValuePresentedSource" in {
- val future = Future.successful("a")
+ val promise = Promise[String]()
+ val future = promise.future
TraversalBuilder.getValuePresentedSource(Source.future(future)).get.asInstanceOf[FutureSource[
String]].future should ===(
future)
diff --git a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/SourceSpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/SourceSpec.scala
index 72fd8af97c..9cef219a9b 100644
--- a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/SourceSpec.scala
+++ b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/SourceSpec.scala
@@ -556,8 +556,49 @@ class SourceSpec extends StreamSpec with DefaultTimeout {
}
}
- "Source.futureSource" must {
+ "Source.future" must {
+ "work as empty source when the future source completes with null" in {
+ val source = Source.future(Future.successful(null.asInstanceOf[String]))
+ val probe = source.runWith(TestSink[String]())
+
+ probe.request(1)
+ probe.expectComplete()
+ }
+
+ "work with a successful future" in {
+ val source = Source.future(Future.successful(42))
+ val probe = source.runWith(TestSink[Int]())
+
+ probe.request(1)
+ probe.expectNext(42)
+ probe.expectComplete()
+ }
+ "work with a failed future" in {
+ val ex = new RuntimeException("boom")
+ val source = Source.future(Future.failed(ex))
+ val probe = source.runWith(TestSink[Int]())
+
+ probe.request(1)
+ probe.expectError().getMessage should ===("boom")
+ }
+
+ "work with a delayed future" in {
+ val promise = scala.concurrent.Promise[Int]()
+ val source = Source.future(promise.future)
+ val probe = source.runWith(TestSink[Int]())
+
+ probe.request(1)
+ probe.expectNoMessage(500.millis)
+
+ promise.success(42)
+
+ probe.expectNext(42)
+ probe.expectComplete()
+ }
+ }
+
+ "Source.futureSource" must {
"not cancel substream twice" in {
val result = Source
.futureSource(pekko.pattern.after(2.seconds)(Future.successful(Source(1 to 2))))
@@ -567,6 +608,46 @@ class SourceSpec extends StreamSpec with DefaultTimeout {
Await.result(result, 4.seconds) shouldBe Done
}
+
+ "fail when the future completes with null" in {
+ val source = Source.futureSource(Future.successful(null.asInstanceOf[Source[Int, NotUsed]]))
+ val probe = source.runWith(TestSink[Int]())
+
+ probe.request(1)
+ probe.expectError().getMessage should include("futureSource completed with null")
+ }
+
+ "work with a successful future" in {
+ val source = Source.futureSource(Future.successful(Source(1 to 3)))
+ val probe = source.runWith(TestSink[Int]())
+
+ probe.request(3)
+ probe.expectNext(1, 2, 3)
+ probe.expectComplete()
+ }
+
+ "work with a failed future source" in {
+ val ex = new RuntimeException("boom")
+ val source = Source.futureSource(Future.failed(ex))
+ val probe = source.runWith(TestSink[Int]())
+
+ probe.request(1)
+ probe.expectError().getMessage should ===("boom")
+ }
+
+ "work with a delayed future source" in {
+ val promise = scala.concurrent.Promise[Source[Int, NotUsed]]()
+ val source = Source.futureSource(promise.future)
+ val probe = source.runWith(TestSink[Int]())
+
+ probe.request(3)
+ probe.expectNoMessage(500.millis)
+
+ promise.success(Source(1 to 3))
+
+ probe.expectNext(1, 2, 3)
+ probe.expectComplete()
+ }
}
"Source of sources" must {
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 1cc0b59eef..e6c867ebc9 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
@@ -589,8 +589,12 @@ object Source {
* Emits a single value when the given `Future` is successfully completed and then completes the stream.
* The stream fails if the `Future` is completed with a failure.
*/
- def future[T](futureElement: Future[T]): Source[T, NotUsed] =
- fromGraph(new FutureSource[T](futureElement))
+ def future[T](futureElement: Future[T]): Source[T, NotUsed] = futureElement.value match {
+ case None => fromGraph(new FutureSource[T](futureElement))
+ case Some(scala.util.Success(null)) => empty[T]
+ case Some(scala.util.Success(elem)) => single(elem)
+ case Some(scala.util.Failure(ex)) => failed[T](ex)
+ }
/**
* Never emits any elements, never completes and never fails.
@@ -612,8 +616,14 @@ object Source {
* Turn a `Future[Source]` into a source that will emit the values of the source when the future completes successfully.
* If the `Future` is completed with a failure the stream is failed.
*/
- def futureSource[T, M](futureSource: Future[Source[T, M]]): Source[T, Future[M]] =
- fromGraph(new FutureFlattenSource(futureSource))
+ def futureSource[T, M](futureSource: Future[Source[T, M]]): Source[T, Future[M]] = futureSource.value match {
+ case None => fromGraph(new FutureFlattenSource(futureSource))
+ case Some(scala.util.Success(null)) =>
+ val exception = new NullPointerException("futureSource completed with null")
+ Source.failed(exception).mapMaterializedValue(_ => Future.failed[M](exception))
+ case Some(scala.util.Success(source)) => source.mapMaterializedValue(Future.successful)
+ case Some(scala.util.Failure(ex)) => Source.failed[T](ex).mapMaterializedValue(_ => Future.failed[M](ex))
+ }
/**
* Defers invoking the `create` function to create a single element until there is downstream demand.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org