feat: Add switchMap operator

docs/src/main/paradox/stream/operators/Source-or-Flow/switchMap.md

@@ -1,6 +1,6 @@
 # switchMap
 
-Transforms each input element into a `Source` of output elements that is then flattened into the output stream until a new input element is received.
+Transforms each input element into a `Source` of output elements that is then flattened into the output stream until a new input element is received (at which point the current (now previous) substream is cancelled and the new one is flattend into the output stream).
 
 @ref[Nesting and flattening operators](../index.md#nesting-and-flattening-operators)
 
@@ -11,8 +11,9 @@ Transforms each input element into a `Source` of output elements that is then fl
 ## Description
 
 Transforms each input element into a `Source` of output elements that is then flattened into the output stream until a 
-new input element is received at which point the current (now previous) substream is cancelled (which is why this
-operator is sometimes also called "flatMapLatest").
+new input element is received at which point the current (now previous) substream is cancelled while the `Source` resulting
+from the input element is subscribed to and flattened into the output stream. So effectively, only the "latest" `Source`
+is flattened into the output stream (which is why this  operator is sometimes also called "flatMapLatest").
 
 ## Reactive Streams semantics
 

stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Switch.scala

@@ -31,6 +31,7 @@ import pekko.stream.stage.GraphStage
 import pekko.stream.stage.GraphStageLogic
 import pekko.stream.stage.InHandler
 import pekko.stream.stage.OutHandler
+import pekko.util.OptionVal
 
 /**
  * INTERNAL API
@@ -45,35 +46,33 @@ import pekko.stream.stage.OutHandler
   override val shape = FlowShape(in, out)
 
   override def createLogic(enclosingAttributes: Attributes) =
-    new GraphStageLogic(shape) {
+    new GraphStageLogic(shape) with InHandler with OutHandler {
 
-      var source = Option.empty[SubSinkInlet[T]]
+      var source = OptionVal.none[SubSinkInlet[T]]
 
       override def preStart(): Unit = {
         pull(in)
         super.preStart()
       }
 
-      setHandler(in,
-        new InHandler {
-          override def onPush(): Unit = {
-            val source = grab(in)
-            setSource(source)
-            tryPull(in)
-          }
+      override def onPush(): Unit = {
+        val source = grab(in)
+        setSource(source)
+        tryPull(in)
+      }
 
-          override def onUpstreamFinish(): Unit = if (source.isEmpty) completeStage()
-        })
+      override def onUpstreamFinish(): Unit = if (source.isEmpty) completeStage()
 
-      setHandler(out,
-        new OutHandler {
-          override def onPull(): Unit = {
-            if (isAvailable(out)) tryPushOut()
-          }
-        })
+      override def onPull(): Unit = {
+        if (isAvailable(out)) tryPushOut()
+      }
+
+      setHandler(in, this)
+      setHandler(out, this)
 
       def tryPushOut(): Unit = {
-        source.foreach { src =>
+        if (source.isDefined) {
+          val src = source.get
           if (src.isAvailable) {
             push(out, src.grab())
             if (!src.isClosed) src.pull()
@@ -99,17 +98,21 @@ import pekko.stream.stage.OutHandler
           }
         })
         sinkIn.pull()
-        this.source = Some(sinkIn)
+        this.source = OptionVal.Some(sinkIn)
         val graph = Source.fromGraph(source).to(sinkIn.sink)
         subFusingMaterializer.materialize(graph, defaultAttributes = enclosingAttributes)
       }
 
       def removeCurrentSource(completeIfClosed: Boolean): Unit = {
-        source = None
+        source = OptionVal.none
         if (completeIfClosed && isClosed(in)) completeStage()
       }
 
-      private def cancelCurrentSource(): Unit = source.foreach(_.cancel())
+      private def cancelCurrentSource(): Unit = {
+        if (source.isDefined) {
+          source.get.cancel()
+        }
+      }
 
       override def postStop(): Unit = cancelCurrentSource()
 

stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala

@@ -2821,7 +2821,7 @@ trait FlowOps[+Out, +Mat] {
    * @since 1.2.0
    */
   def switchMap[T, M](f: Out => Graph[SourceShape[T], M]): Repr[T] =
-    map(f).via(new Switch[T, M])
+    map(f).via(new Switch[T, M].addAttributes(Attributes(SourceLocation.forLambda(f))))
 
   /**
    * If the first element has not passed through this operator before the provided timeout, the stream is failed