Skip to content
Open
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 14 additions & 6 deletions akka-stream/src/main/scala/akka/stream/scaladsl/Source.scala
Original file line number Diff line number Diff line change
Expand Up @@ -5,17 +5,14 @@
package akka.stream.scaladsl

import java.util.concurrent.CompletionStage

import scala.annotation.nowarn
import scala.annotation.tailrec
import scala.annotation.unchecked.uncheckedVariance
import scala.collection.immutable
import scala.jdk.FutureConverters._
import scala.concurrent.{ Future, Promise }
import scala.concurrent.duration.FiniteDuration

import org.reactivestreams.{ Publisher, Subscriber }

import akka.{ Done, NotUsed }
import akka.actor.{ ActorRef, Cancellable }
import akka.annotation.InternalApi
Expand All @@ -27,6 +24,9 @@ import akka.stream.impl.fusing.GraphStages._
import akka.stream.stage.GraphStageWithMaterializedValue
import akka.util.ConstantFun

import scala.util.Failure
import scala.util.Success

/**
* A `Source` is a set of stream processing steps that has one open output. It can comprise
* any number of internal sources and transformations that are wired together, or it can be
Expand Down Expand Up @@ -505,7 +505,11 @@ object Source {
* 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))
futureElement.value match {
case Some(Success(value)) => single(value)
case Some(Failure(cause)) => failed(cause)
case None => fromGraph(new FutureSource[T](futureElement))
}

/**
* Never emits any elements, never completes and never fails.
Expand All @@ -527,8 +531,12 @@ 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 Some(Success(source)) => source.mapMaterializedValue(Future.successful)
case _ => fromGraph(new FutureFlattenSource(futureSource))
}
}

/**
* Defers invoking the `create` function to create a single element until there is downstream demand.
Expand Down
Loading