SourceRef.source and SinkRef.sink() create new instances on each access

[He-Pin]

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.

Current behavior

SourceRefImpl.scala:32-33:

def source: Source[T, NotUsed] =
  Source.fromGraph(new SourceRefStageImpl(OptionVal.Some(initialPartnerRef)))

SinkRefImpl.scala:33-34:

override def sink(): Sink[In, NotUsed] =
  Sink.fromGraph(new SinkRefStageImpl[In](OptionVal.Some(initialPartnerRef)))

Proposed fix

Option A: Cache with lazy val (requires changing from case class to class):

private[stream] final class SourceRefImpl[T](initialPartnerRef: ActorRef) extends SourceRef[T] {
  lazy val source: Source[T, NotUsed] =
    Source.fromGraph(new SourceRefStageImpl(OptionVal.Some(initialPartnerRef))).mapMaterializedValue(_ => NotUsed)
  // ... equals/hashCode/toString to preserve case class semantics
}

Option B: Add a clear error message when double-materialization is attempted, rather than a confusing handshake failure.