He-Pin opened a new pull request, #3114: URL: https://github.com/apache/pekko/pull/3114
### Motivation `SourceRefImpl.source` and `SinkRefImpl.sink()` create a new `SourceRefStageImpl` / `SinkRefStageImpl` instance on every access. If a user accidentally materializes the returned `Source`/`Sink` more than once, multiple stage instances compete for the same partner handshake, causing the second materialization to fail with a confusing error message (typically a handshake timeout or `IllegalStateException in state AwaitingSubscription`). Fixes #3106 ### Modification Add an `AtomicBoolean` guard per `SourceRefImpl` / `SinkRefImpl` instance. When the backing stage is materialized, the guard is atomically claimed via `compareAndSet(false, true)`. A second materialization attempt throws a clear `IllegalStateException`: ``` This SourceRef has already been materialized. SourceRef is single-use: calling .source and materializing it more than once is not supported. ``` The guard is passed as an optional `owner` parameter to `SourceRefStageImpl` / `SinkRefStageImpl` (defaulting to `null` for the initial stage created by `StreamRefs.sourceRef()` / `StreamRefs.sinkRef()`, which are always single-materialization by design). **Key design decisions:** - `AtomicBoolean` per instance (not `AtomicBooleanFieldUpdater`, which does not exist in the JDK; `AtomicIntegerFieldUpdater` from a Scala companion object fails due to JVM-level `private` access restrictions on case class body fields) - The `owner` constructor parameter defaults to `null`, so `StreamRefs.scala` call sites remain unchanged - `SourceRefImpl` / `SinkRefImpl` are `private[stream]` and `@InternalApi`, so the new field does not affect public binary compatibility ### Result Double materialization of `SourceRef` or `SinkRef` now fails fast with a descriptive error, instead of the previous opaque handshake failure. ### Tests ``` sbt "stream-tests / Test / testOnly org.apache.pekko.stream.scaladsl.StreamRefsSpec" # 25/25 passed ``` Two new tests added: - `A SourceRef must "not allow materializing multiple times"` — verifies `IllegalStateException` on second `.source.run()` - `A SinkRef must "not allow materializing multiple times"` — updated from previous silent-cancellation behavior to expect the new `IllegalStateException` ### References Fixes #3106 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
