Add Flow.lazyInitAsync operator for lazy flow materialization

[He-Pin]

Motivation

Currently, pekko provides Sink.lazyInit for lazily creating a sink based on the first element, and Source.lazySource/Source.lazyFutureSource for lazy source creation. However, there is no equivalent operator for Flow — the ability to lazily create a flow based on the first element that passes through it.

This is useful when:

• The flow's behavior depends on properties of the first element (e.g., detecting format/encoding from the first message)
• Expensive resources should only be initialized when data actually starts flowing
• The flow factory is asynchronous (e.g., needs to query a service before deciding how to process elements)

Proposed Enhancement

Add a Flow.lazyInitAsync operator that:

1. Accepts a factory function (firstElement: T) => Future[Flow[T, U, M]]
2. On the first element, invokes the factory asynchronously
3. Materializes the returned flow and feeds all elements (including the first) through it
4. Returns the materialized value of the lazily created flow

References

This operator was implemented in the Akka.NET project: akkadotnet/akka.net#5476

[He-Pin]

Closing due to factual errors in the description:

1. Flow.lazyFlow and Flow.lazyFutureFlow already exist (scaladsl Flow.scala:726,754, javadsl Flow.scala:302,333). These accept () => Flow[...] and () => Future[Flow[...]] factories respectively.
2. Sink.lazyInit does not exist — the correct operator names are Sink.lazySink and Sink.lazyFutureSink.

The core idea — a flow factory driven by the first element ((firstElement: T) => Future[Flow[T, U, M]]) — is genuinely distinct from existing lazyFutureFlow (which takes a no-arg factory). If this capability is needed, please open a new issue that:

• Acknowledges the existing lazyFlow/lazyFutureFlow operators
• Uses correct operator names (Sink.lazySink, not Sink.lazyInit)
• Follows Pekko naming conventions (e.g., Flow.lazyFutureFlowWith rather than lazyInitAsync)
• Justifies the first-element-driven variant vs. using lazyFutureFlow with external state or prefixAndTail(1).flatMapConcat(...)