optimize: extend internalConcat dispatch for value-presented sources

[He-Pin]

⚠️ Stacked on #2977 — do not merge until #2977 lands. The current diff includes #2977's commits; once #2977 merges, this PR's diff will narrow to its own commit. Drafted to make the dependency explicit.

Motivation

FlowOps#internalConcat previously had only one fast-path: a SingleSource on the right-hand side of concat / concatLazy was rerouted through the lightweight SingleConcat stage instead of the general two-port Concat[U](2, detached) fan-in graph (which materializes the whole substream plus a detacher buffer). All other value-presented sources (IterableSource, IteratorSource, RangeSource, RepeatSource, JavaStreamSource, FutureSource, FailedSource) still took the heavy concatGraph path even though their data is already in memory or trivially producible — the fan-in machinery and substream materialization were pure overhead. Heavy concat users (pekko-http and others) carry that cost on every materialization.

This mirrors the optimization just shipped for FlattenConcat / FlattenMerge in #2977, applied to the concat operator chain.

Modification

Add four small specialized GraphStage[FlowShape[E, E]] siblings of SingleConcat, each passing through elements while upstream is alive and draining its captured value-presented payload on onUpstreamFinish:

New stage	Handles	Drain behavior
IterableConcat[E](createIterator)	IterableSource, IteratorSource, RangeSource, JavaStreamSource	emitMultiple(out, createIterator(), () => completeStage())
RepeatConcat[E](elem)	RepeatSource	swap OutHandler so each onPull pushes elem
FailedConcat[E](failure)	FailedSource	failStage(failure)
FutureConcat[E](future)	FutureSource	If completed: emit/fail. If pending: swap OutHandler to a no-op (avoid pulling closed in), register async callback that emits/fails when resolved.

internalConcat is extended to dispatch via TraversalBuilder.getValuePresentedSource and pattern-match the eight value-presented source types (existing SingleSource path is preserved). The detached flag is irrelevant for these stages — the right-hand data is already present, so the one-element pre-fetch buffer that detached=true provides has nothing to fetch (matching SingleConcat's precedent).

Note: Source.future(Future.successful(x)) is itself optimized upstream into SingleSource, so completed futures dispatch via SingleConcat. Same for Source.future(Future.failed(ex)) → FailedSource → FailedConcat. Only genuinely-pending futures land on FutureConcat.

Result

For the eight value-presented source types, concat and concatLazy no longer pay for substream materialization or the two-port fan-in graph. Observable behavior is unchanged for all other sources, which still take the existing concatGraph path.

Tests

Eleven directional tests added to AbstractFlowConcatSpec (cover both concat and concatLazy paths):

• optimize iterable concat / range concat / iterator concat / java-stream concat — assert IterableConcat in the traversal builder, output correct.
• optimize repeat concat — assert RepeatConcat(0), bounded with .take(6).
• optimize failed concat — assert FailedConcat, error propagated.
• optimize completed-future concat — confirms it routes through SingleConcat (upstream optimization).
• optimize pending-future concat — uses an unresolved Promise, asserts FutureConcat and correct delivery after resolution.
• optimize failed-future concat — confirms it routes through FailedConcat (upstream optimization).
• avoid downstream substream materialization for value-presented sources — counter-based assertion of zero substream materialization.

Verification commands:

sbt "stream-tests/testOnly *FlowConcatSpec *FlowConcatLazySpec *FlowConcatAllSpec *FlowConcatAllLazySpec *GraphConcatSpec"
sbt "stream/mimaReportBinaryIssues"
sbt "stream-tests/test"
sbt scalafmtAll headerCreateAll

All *ConcatSpec (96 tests) pass. MiMa clean (all new stages are private[pekko] / InternalApi). Full stream-tests/test (3027 tests) passes; the only failure was an unrelated BoundedSourceQueueSpec flake that does not use concat and passes in isolation.

References

Stacked on #2977.