@@ -35,16 +35,9 @@ import pekko.stream.Attributes.{ InputBuffer, LogLevels }
3535import pekko .stream .Attributes .SourceLocation
3636import pekko .stream .OverflowStrategies ._
3737import pekko .stream .Supervision .Decider
38- import pekko .stream .impl .{
39- Buffer => BufferImpl ,
40- ContextPropagation ,
41- FailedSource ,
42- JavaStreamSource ,
43- ReactiveStreamsCompliance ,
44- TraversalBuilder
45- }
38+ import pekko .stream .impl .{ Buffer => BufferImpl , ContextPropagation , ReactiveStreamsCompliance , TraversalBuilder }
4639import pekko .stream .impl .Stages .DefaultAttributes
47- import pekko .stream .impl .fusing .GraphStages .{ FutureSource , SimpleLinearGraphStage , SingleSource }
40+ import pekko .stream .impl .fusing .GraphStages .SimpleLinearGraphStage
4841import pekko .stream .scaladsl .{
4942 DelayStrategy ,
5043 Source ,
@@ -2183,36 +2176,19 @@ private[pekko] object TakeWithin {
21832176 override def onPull (): Unit = pull(in)
21842177
21852178 @ nowarn(" msg=Any" )
2186- @ tailrec
21872179 def onFailure (ex : Throwable ): Unit = {
21882180 import Collect .NotApplied
21892181 if (maximumRetries < 0 || attempt < maximumRetries) {
21902182 pf.applyOrElse(ex, NotApplied ) match {
21912183 case _ : NotApplied .type => failStage(ex)
21922184 case source : Graph [SourceShape [T ] @ unchecked, M @ unchecked] if TraversalBuilder .isEmptySource(source) =>
21932185 completeStage()
2194- case source : Graph [SourceShape [T ] @ unchecked, M @ unchecked] =>
2195- TraversalBuilder .getValuePresentedSource(source) match {
2196- case OptionVal .Some (graph) => graph match {
2197- case singleSource : SingleSource [T @ unchecked] => emit(out, singleSource.elem, () => completeStage())
2198- case failed : FailedSource [T @ unchecked] => onFailure(failed.failure)
2199- case futureSource : FutureSource [T @ unchecked] => futureSource.future.value match {
2200- case Some (Success (elem)) => emit(out, elem, () => completeStage())
2201- case Some (Failure (ex)) => onFailure(ex)
2202- case None =>
2203- switchTo(source)
2204- attempt += 1
2205- }
2206- case iterableSource : IterableSource [T @ unchecked] =>
2207- emitMultiple(out, iterableSource.elements, () => completeStage())
2208- case javaStreamSource : JavaStreamSource [T @ unchecked, _] =>
2209- emitMultiple(out, javaStreamSource.open().spliterator(), () => completeStage())
2210- case _ =>
2211- switchTo(source)
2212- attempt += 1
2213- }
2186+ case other : Graph [SourceShape [T ] @ unchecked, M @ unchecked] =>
2187+ TraversalBuilder .getSingleSource(other) match {
2188+ case OptionVal .Some (singleSource) =>
2189+ emit(out, singleSource.elem.asInstanceOf [T ], () => completeStage())
22142190 case _ =>
2215- switchTo(source )
2191+ switchTo(other )
22162192 attempt += 1
22172193 }
22182194 case _ => throw new IllegalStateException () // won't happen, compiler exhaustiveness check pleaser
0 commit comments