You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pekko.apache.org by he...@apache.org on 2025/03/08 20:32:30 UTC
(pekko) branch main updated: feat: Add switchMap operator (#1787)
This is an automated email from the ASF dual-hosted git repository.
hepin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new 57af84a56a feat: Add switchMap operator (#1787)
57af84a56a is described below
commit 57af84a56a1a17a2d9285e27e2edd303be0cc4f5
Author: Martin Hansen <co...@martin-hansen.de>
AuthorDate: Sat Mar 8 21:32:24 2025 +0100
feat: Add switchMap operator (#1787)
* feat: switchMap
* chore: minor code review issues
- adjust import style
- adjust header
* chore: fix header check
(add missing newline)
* chore: fix java formatting
* docs: fix operator docs
* docs: improve SwitchMap description
* chore: various review issues
* docs: fix typo
* docs: fix another typo
* chore: more review changes
---
.../stream/operators/Source-or-Flow/switchMap.md | 30 +++
docs/src/main/paradox/stream/operators/index.md | 2 +
.../org/apache/pekko/stream/javadsl/FlowTest.java | 26 ++
.../apache/pekko/stream/javadsl/SourceTest.java | 24 ++
.../pekko/stream/scaladsl/FlowSwitchSpec.scala | 299 +++++++++++++++++++++
.../org/apache/pekko/stream/impl/Stages.scala | 1 +
.../apache/pekko/stream/impl/fusing/Switch.scala | 124 +++++++++
.../org/apache/pekko/stream/javadsl/Flow.scala | 19 ++
.../org/apache/pekko/stream/javadsl/Source.scala | 19 ++
.../org/apache/pekko/stream/javadsl/SubFlow.scala | 19 ++
.../apache/pekko/stream/javadsl/SubSource.scala | 19 ++
.../org/apache/pekko/stream/scaladsl/Flow.scala | 19 ++
12 files changed, 601 insertions(+)
diff --git a/docs/src/main/paradox/stream/operators/Source-or-Flow/switchMap.md b/docs/src/main/paradox/stream/operators/Source-or-Flow/switchMap.md
new file mode 100644
index 0000000000..554a2fd8d5
--- /dev/null
+++ b/docs/src/main/paradox/stream/operators/Source-or-Flow/switchMap.md
@@ -0,0 +1,30 @@
+# 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 (at which point the current (now previous) substream is cancelled and the new one is flattened into the output stream).
+
+@ref[Nesting and flattening operators](../index.md#nesting-and-flattening-operators)
+
+## Signature
+
+@apidoc[Flow.switchMap](Flow) { scala="#switchMap[T,M](f:Out=%3Eorg.apache.pekko.stream.Graph[org.apache.pekko.stream.SourceShape[T],M]):FlowOps.this.Repr[T]" java="#switchMap(org.apache.pekko.japi.function.Function)" }
+
+## 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 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
+
+@@@div { .callout }
+
+**emits** when the current substream has an element available
+
+**backpressures** never
+
+**completes** upstream completes and the current substream completes
+
+@@@
+
+
diff --git a/docs/src/main/paradox/stream/operators/index.md b/docs/src/main/paradox/stream/operators/index.md
index 643d2caa2b..8127e569d6 100644
--- a/docs/src/main/paradox/stream/operators/index.md
+++ b/docs/src/main/paradox/stream/operators/index.md
@@ -262,6 +262,7 @@ See the @ref:[Substreams](../stream-substream.md) page for more detail and code
|Source/Flow|<a name="prefixandtail"></a>@ref[prefixAndTail](Source-or-Flow/prefixAndTail.md)|Take up to *n* elements from the stream (less than *n* only if the upstream completes before emitting *n* elements) and returns a pair containing a strict sequence of the taken element and a stream representing the remaining elements.|
|Source/Flow|<a name="splitafter"></a>@ref[splitAfter](Source-or-Flow/splitAfter.md)|End the current substream whenever a predicate returns `true`, starting a new substream for the next element.|
|Source/Flow|<a name="splitwhen"></a>@ref[splitWhen](Source-or-Flow/splitWhen.md)|Split off elements into a new substream whenever a predicate function return `true`.|
+|Source/Flow|<a name="switchmap"></a>@ref[switchMap](Source-or-Flow/switchMap.md)|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.|
## Time aware operators
@@ -591,6 +592,7 @@ For more background see the @ref[Error Handling in Streams](../stream-error.md)
* [splitWhen](Source-or-Flow/splitWhen.md)
* [statefulMap](Source-or-Flow/statefulMap.md)
* [statefulMapConcat](Source-or-Flow/statefulMapConcat.md)
+* [switchMap](Source-or-Flow/switchMap.md)
* [take](Source-or-Flow/take.md)
* [takeLast](Sink/takeLast.md)
* [takeWhile](Source-or-Flow/takeWhile.md)
diff --git a/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/FlowTest.java b/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/FlowTest.java
index 7ea1e7d9ff..1b6b400f85 100644
--- a/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/FlowTest.java
+++ b/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/FlowTest.java
@@ -787,6 +787,32 @@ public class FlowTest extends StreamTest {
assertEquals(expected, set);
}
+ @Test
+ public void mustBeAbleToUseSwitchMap() throws Exception {
+ final TestKit probe = new TestKit(system);
+ final Iterable<Integer> mainInputs = Arrays.asList(-1, 0, 1);
+ final Iterable<Integer> substreamInputs = Arrays.asList(10, 11, 12, 13, 14, 15, 16, 17, 18, 19);
+
+ final Flow<Integer, Integer, NotUsed> flow =
+ Flow.<Integer>create()
+ .switchMap(
+ new Function<Integer, Source<Integer, NotUsed>>() {
+ @Override
+ public Source<Integer, NotUsed> apply(Integer param) throws Exception {
+ return param > 0
+ ? Source.fromIterator(substreamInputs::iterator)
+ : Source.never();
+ }
+ });
+
+ CompletionStage<List<Integer>> future =
+ Source.from(mainInputs).via(flow).runWith(Sink.seq(), system);
+
+ List<Integer> result = future.toCompletableFuture().get(3, TimeUnit.SECONDS);
+
+ assertEquals(substreamInputs, result);
+ }
+
@Test
public void mustBeAbleToUseBuffer() throws Exception {
final TestKit probe = new TestKit(system);
diff --git a/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/SourceTest.java b/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/SourceTest.java
index 9be7dab422..86462f7b80 100644
--- a/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/SourceTest.java
+++ b/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/SourceTest.java
@@ -509,6 +509,30 @@ public class SourceTest extends StreamTest {
assertEquals(expected, set);
}
+ @Test
+ public void mustBeAbleToUseSwitchMap() throws Exception {
+ final TestKit probe = new TestKit(system);
+ final Iterable<Integer> mainInputs = Arrays.asList(-1, 0, 1);
+ final Iterable<Integer> substreamInputs = Arrays.asList(10, 11, 12, 13, 14, 15, 16, 17, 18, 19);
+
+ CompletionStage<List<Integer>> future =
+ Source.from(mainInputs)
+ .switchMap(
+ new Function<Integer, Source<Integer, NotUsed>>() {
+ @Override
+ public Source<Integer, NotUsed> apply(Integer param) throws Exception {
+ return param > 0
+ ? Source.fromIterator(substreamInputs::iterator)
+ : Source.never();
+ }
+ })
+ .runWith(Sink.seq(), system);
+
+ List<Integer> result = future.toCompletableFuture().get(3, TimeUnit.SECONDS);
+
+ assertEquals(substreamInputs, result);
+ }
+
@Test
public void mustBeAbleToUseBuffer() throws Exception {
final TestKit probe = new TestKit(system);
diff --git a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowSwitchSpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowSwitchSpec.scala
new file mode 100644
index 0000000000..cc21ee2ea4
--- /dev/null
+++ b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowSwitchSpec.scala
@@ -0,0 +1,299 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2014-2022 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.stream.scaladsl
+
+import org.apache.pekko
+import pekko.NotUsed
+import pekko.stream.Attributes
+import pekko.stream.Outlet
+import pekko.stream.SourceShape
+import pekko.stream.stage.GraphStage
+import pekko.stream.stage.GraphStageLogic
+import pekko.stream.stage.OutHandler
+import pekko.stream.testkit.StreamSpec
+import pekko.stream.testkit.TestPublisher
+import pekko.stream.testkit.Utils.TE
+import pekko.stream.testkit.scaladsl.TestSink
+import pekko.stream.testkit.scaladsl.TestSource
+
+import org.scalatest.exceptions.TestFailedException
+
+import scala.concurrent.Await
+import scala.concurrent.duration.DurationInt
+
+class FlowSwitchSpec extends StreamSpec {
+
+ import FlowSwitchSpec._
+ import system.dispatcher
+
+ "A Switch / switchMap operator" must {
+
+ "wait for the last substream to complete" in {
+ Source(1 to 3)
+ .switchMap(x => Source(1 to 3).map(x -> _))
+ .runWith(Sink.seq)
+ .map(_.takeRight(3))
+ .futureValue should ===(Seq(3 -> 1, 3 -> 2, 3 -> 3))
+ }
+
+ "not wait for the previous / non-last substream to complete" in {
+ val lastValues = List(1, 2)
+ Source(List(Source.never, Source(lastValues)))
+ .switchMap(identity)
+ .runWith(Sink.seq)
+ .futureValue should ===(lastValues)
+ }
+
+ // guards against bugs that may occur if in the future, someone applies certain Source.single-specific
+ // optimizations
+ "not behave differently for substreams Source.single(x) and Source(List(x))" in {
+ def via(createSubStream: Int => Source[Int, NotUsed]): Seq[Int] =
+ Source(0 to 3).switchMap(createSubStream).runWith(Sink.seq).futureValue
+
+ via(Source.single) should ===(via(x => Source(List(x))))
+ }
+
+ "complete if upstream completes after last substream" in {
+ val (mainPub, probe) =
+ TestSource.probe[Source[Int, NotUsed]].switchMap(identity).toMat(TestSink.probe)(Keep.both).run()
+
+ def nextSubstream(): TestPublisher.Probe[Int] = {
+ val p = TestPublisher.probe[Int]()
+ mainPub.sendNext(Source.fromPublisher(p))
+ p
+ }
+
+ probe.request(1)
+
+ nextSubstream().sendComplete()
+ nextSubstream().sendComplete()
+ probe.expectNoMessage()
+ mainPub.sendComplete()
+ probe.expectComplete()
+ }
+
+ // copied (with modifications) from org.apache.pekko.stream.scaladsl.FlowFlattenMergeSpec
+ "propagate early failure from main stream" in {
+ val ex = new Exception("buh")
+ intercept[TestFailedException] {
+ Source.failed(ex).switchMap((i: Int) => Source.single(i)).runWith(Sink.head).futureValue
+ }.cause.get should ===(ex)
+ }
+
+ // copied (with modifications) from org.apache.pekko.stream.scaladsl.FlowFlattenMergeSpec
+ "propagate late failure from main stream" in {
+ val ex = new Exception("buh")
+ intercept[TestFailedException] {
+ (Source(List(Source.never, Source.never)) ++ Source.failed(ex)).switchMap(identity).runWith(
+ Sink.head).futureValue
+ }.cause.get should ===(ex)
+ }
+
+ // copied (with modifications) from org.apache.pekko.stream.scaladsl.FlowFlattenMergeSpec
+ "propagate failure from map function" in {
+ val ex = new Exception("buh")
+ intercept[TestFailedException] {
+ Source(1 to 3).switchMap(i => if (i == 3) throw ex else Source.never).runWith(Sink.head).futureValue
+ }.cause.get should ===(ex)
+ }
+
+ // copied (with modifications) from org.apache.pekko.stream.scaladsl.FlowFlattenMergeSpec
+ "bubble up substream exceptions" in {
+ val ex = new Exception("buh")
+ intercept[TestFailedException] {
+ Source(List(Source.never, Source.never, Source.failed(ex))).switchMap(identity).runWith(Sink.head).futureValue
+ }.cause.get should ===(ex)
+ }
+
+ // copied (with modifications) from org.apache.pekko.stream.scaladsl.FlowFlattenMergeSpec
+ "bubble up substream materialization exception" in {
+ val matFail = TE("fail!")
+ object FailingInnerMat extends GraphStage[SourceShape[String]] {
+ val out = Outlet[String]("out")
+ val shape = SourceShape(out)
+
+ override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = new GraphStageLogic(shape) {
+ if ("confuse IntellIJ dead code checker".length > 2) {
+ throw matFail
+ }
+ }
+ }
+
+ val result = Source.single(()).switchMap(_ => Source.fromGraph(FailingInnerMat)).runWith(Sink.ignore)
+
+ result.failed.futureValue should ===(matFail)
+
+ }
+
+ // copied (with modifications) from org.apache.pekko.stream.scaladsl.FlowFlattenMergeSpec
+ "cancel substream when failing from main stream" in {
+ val (mainPub, probe) =
+ TestSource.probe[Source[Int, NotUsed]].switchMap(identity).toMat(TestSink.probe)(Keep.both).run()
+ val subPub = TestPublisher.probe[Int]()
+
+ probe.request(1)
+ mainPub.sendNext(Source.fromPublisher(subPub))
+ subPub.expectRequest()
+
+ val ex = new Exception("buh")
+ mainPub.sendError(ex)
+ subPub.expectCancellation()
+ }
+
+ "cancel previous substream when failing map function" in {
+ val p = TestPublisher.probe[Int]()
+ val ex = new Exception("buh")
+ Source.tick(0.millis, 100.millis, 1).scan(1)(_ + _)
+ .switchMap {
+ case 1 =>
+ Source.fromPublisher(p)
+ case 2 =>
+ throw ex
+ }
+ .toMat(Sink.seq)(Keep.right)
+ .run()
+ p.expectCancellation()
+ }
+
+ // copied (with modifications) from org.apache.pekko.stream.scaladsl.FlowFlattenMergeSpec
+ "cancel substream when being cancelled" in {
+ val p = TestPublisher.probe[Int]()
+ val sink = (Source(List(Source.fromPublisher(p))) ++ Source.never)
+ .switchMap(identity)
+ .runWith(TestSink.probe)
+ sink.request(1)
+ p.expectRequest()
+ sink.cancel()
+ p.expectCancellation()
+ }
+
+ "cancel previous substream once a next substream arrives" in {
+ val (mainPub, probe) =
+ TestSource.probe[Source[Int, NotUsed]].switchMap(identity).toMat(TestSink.probe)(Keep.both).run()
+
+ def nextSubstream(): TestPublisher.Probe[Int] = {
+ val p = TestPublisher.probe[Int]()
+ mainPub.sendNext(Source.fromPublisher(p))
+ p
+ }
+
+ probe.request(1)
+
+ val subPub1 = nextSubstream()
+ subPub1.ensureSubscription()
+ val subPub2 = nextSubstream()
+ subPub1.expectCancellation()
+ subPub2.ensureSubscription()
+ }
+
+ // copied (with modifications) from org.apache.pekko.stream.scaladsl.FlowFlattenMergeSpec
+ "propagate attributes to inner streams" in {
+ val f = Source
+ .single(attributesSource.addAttributes(Attributes.name("inner")))
+ .switchMap(identity)
+ .addAttributes(Attributes.name("outer"))
+ .runWith(Sink.head)
+
+ val attributes = Await.result(f, 3.seconds).attributeList
+ attributes should contain(Attributes.Name("inner"))
+ attributes should contain(Attributes.Name("outer"))
+ attributes.indexOf(Attributes.Name("inner")) < attributes.indexOf(Attributes.Name("outer")) should be(true)
+ }
+
+ "respect backpressure when emitting items from substream that is faster than downstream" in {
+ val (mainPub, probe) =
+ TestSource.probe[Source[Long, NotUsed]].switchMap(identity).toMat(TestSink.probe)(Keep.both).run()
+
+ def nextSubstream(): TestPublisher.Probe[Long] = {
+ val p = TestPublisher.probe[Long]()
+ mainPub.sendNext(Source.fromPublisher(p))
+ p
+ }
+
+ val subPub1 = nextSubstream()
+ probe.request(1)
+ val nRequests = subPub1.expectRequest()
+ nRequests should be >= 1L
+
+ for (x <- 1L to nRequests) {
+ subPub1.sendNext(x)
+ }
+
+ subPub1.expectNoMessage()
+
+ probe.expectNext(1L)
+ probe.expectNoMessage()
+
+ // end of backpressure:
+ probe.request(nRequests - 1)
+ for (x <- 2L to nRequests) {
+ probe.expectNext(x)
+ }
+
+ }
+
+ "never backpressure to upstream" in {
+ val (mainPub, probe) =
+ TestSource.probe[Source[Int, NotUsed]].switchMap(identity).toMat(TestSink.probe)(Keep.both).run()
+
+ for (_ <- 0 until 10) {
+ for (_ <- 0L until mainPub.expectRequest()) {
+ mainPub.sendNext(Source.single(-1))
+ }
+ for (_ <- 0L until mainPub.expectRequest()) {
+ mainPub.sendNext(Source.never)
+ }
+ }
+
+ val requests = mainPub.expectRequest()
+ mainPub.sendNext(Source(List(1, 2)))
+ probe.request(1)
+ probe.expectNext(1)
+ probe.expectNoMessage()
+ for (_ <- 0L until requests - 1) {
+ mainPub.sendNext(Source.never)
+ }
+ probe.expectNoMessage()
+ probe.request(1)
+ for (_ <- 0 until 10) {
+ for (_ <- 0L until mainPub.expectRequest()) {
+ mainPub.sendNext(Source.never)
+ }
+ }
+ probe.expectNoMessage()
+ mainPub.sendNext(Source.single(3))
+ probe.expectNext(3)
+ }
+ }
+
+}
+
+object FlowSwitchSpec {
+
+ // copied from org.apache.pekko.stream.scaladsl.FlowFlattenMergeSpec
+ private val attributesSource = Source.fromGraph(new GraphStage[SourceShape[Attributes]] {
+ val out = Outlet[Attributes]("AttributesSource.out")
+ override val shape: SourceShape[Attributes] = SourceShape(out)
+
+ override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
+ new GraphStageLogic(shape) with OutHandler {
+ override def onPull(): Unit = {
+ push(out, inheritedAttributes)
+ completeStage()
+ }
+
+ setHandler(out, this)
+ }
+ })
+}
diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/Stages.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/Stages.scala
index 1559a550dc..68bd931dd4 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/impl/Stages.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/impl/Stages.scala
@@ -80,6 +80,7 @@ import pekko.stream.Attributes._
val processor = name("processor")
val identityOp = name("identityOp")
val delimiterFraming = name("delimiterFraming")
+ val switch = name("switch")
val initial = name("initial")
val completion = name("completion")
diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Switch.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Switch.scala
new file mode 100644
index 0000000000..6d446d32c6
--- /dev/null
+++ b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Switch.scala
@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.impl.fusing
+
+import org.apache.pekko
+import pekko.annotation.InternalApi
+import pekko.stream.Attributes
+import pekko.stream.FlowShape
+import pekko.stream.Graph
+import pekko.stream.Inlet
+import pekko.stream.Outlet
+import pekko.stream.SourceShape
+import pekko.stream.impl.Stages.DefaultAttributes
+import pekko.stream.scaladsl.Source
+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
+ */
+@InternalApi private[pekko] final class Switch[T, M]
+ extends GraphStage[FlowShape[Graph[SourceShape[T], M], T]] {
+ private val in = Inlet[Graph[SourceShape[T], M]]("switch.in")
+ private val out = Outlet[T]("switch.out")
+
+ override def initialAttributes = DefaultAttributes.switch
+
+ override val shape = FlowShape(in, out)
+
+ override def createLogic(enclosingAttributes: Attributes) =
+ new GraphStageLogic(shape) with InHandler with OutHandler {
+
+ var source = OptionVal.none[SubSinkInlet[T]]
+
+ override def preStart(): Unit = {
+ pull(in)
+ super.preStart()
+ }
+
+ override def onPush(): Unit = {
+ val source = grab(in)
+ setupCurrentSource(source)
+ tryPull(in)
+ }
+
+ override def onUpstreamFinish(): Unit = if (source.isEmpty) completeStage()
+
+ override def onPull(): Unit = {
+ if (isAvailable(out)) tryPushOut()
+ }
+
+ setHandlers(in, out, this)
+
+ def tryPushOut(): Unit = {
+ source match {
+ case OptionVal.Some(src) =>
+ if (src.isAvailable) {
+ push(out, src.grab())
+ if (!src.isClosed) src.pull()
+ else removeCurrentSource(completeIfClosed = true)
+ }
+ case OptionVal.None =>
+ }
+ }
+
+ def setupCurrentSource(source: Graph[SourceShape[T], M]): Unit = {
+ cancelCurrentSource()
+ removeCurrentSource(completeIfClosed = false)
+ val sinkIn = new SubSinkInlet[T]("SwitchSink")
+ sinkIn.setHandler(new InHandler {
+ override def onPush(): Unit = {
+ if (isAvailable(out)) {
+ push(out, sinkIn.grab())
+ sinkIn.pull()
+ }
+ }
+
+ override def onUpstreamFinish(): Unit = {
+ if (!sinkIn.isAvailable) removeCurrentSource(completeIfClosed = true)
+ }
+ })
+ sinkIn.pull()
+ this.source = OptionVal.Some(sinkIn)
+ val graph = Source.fromGraph(source).to(sinkIn.sink)
+ subFusingMaterializer.materialize(graph, defaultAttributes = enclosingAttributes)
+ }
+
+ def removeCurrentSource(completeIfClosed: Boolean): Unit = {
+ source = OptionVal.none
+ if (completeIfClosed && isClosed(in)) completeStage()
+ }
+
+ private def cancelCurrentSource(): Unit = {
+ source match {
+ case OptionVal.Some(src) =>
+ src.cancel()
+ case OptionVal.None =>
+ }
+ }
+
+ override def postStop(): Unit = cancelCurrentSource()
+
+ }
+
+ override def toString: String = s"Switch"
+}
diff --git a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala
index d8b6d2f813..19f922e4f4 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala
@@ -2759,6 +2759,25 @@ final class Flow[In, Out, Mat](delegate: scaladsl.Flow[In, Out, Mat]) extends Gr
def flatMapMerge[T, M](breadth: Int, f: function.Function[Out, _ <: Graph[SourceShape[T], M]]): Flow[In, T, Mat] =
new Flow(delegate.flatMapMerge(breadth, o => f(o)))
+ /**
+ * 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").
+ *
+ * '''Emits when''' the current substream has an element available
+ *
+ * '''Backpressures when''' never
+ *
+ * '''Completes when''' upstream completes and the current substream completes
+ *
+ * '''Cancels when''' downstream cancels
+ *
+ * @since 1.2.0
+ */
+ def switchMap[T, M](f: function.Function[Out, _ <: Graph[SourceShape[T], M]]): Flow[In, T, Mat] =
+ new Flow(delegate.switchMap(o => f(o)))
+
/**
* Concatenate the given [[Source]] to this [[Flow]], meaning that once this
* Flow’s input is exhausted and all result elements have been generated,
diff --git a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala
index 5241f8b7c2..31e572f4e8 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala
@@ -4263,6 +4263,25 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[
def flatMapMerge[T, M](breadth: Int, f: function.Function[Out, _ <: Graph[SourceShape[T], M]]): Source[T, Mat] =
new Source(delegate.flatMapMerge(breadth, o => f(o)))
+ /**
+ * 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").
+ *
+ * '''Emits when''' the current substream has an element available
+ *
+ * '''Backpressures when''' never
+ *
+ * '''Completes when''' upstream completes and the current substream completes
+ *
+ * '''Cancels when''' downstream cancels
+ *
+ * @since 1.2.0
+ */
+ def switchMap[T, M](f: function.Function[Out, _ <: Graph[SourceShape[T], M]]): Source[T, Mat] =
+ new Source(delegate.switchMap(o => f(o)))
+
/**
* If the first element has not passed through this operator before the provided timeout, the stream is failed
* with a [[org.apache.pekko.stream.InitialTimeoutException]].
diff --git a/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubFlow.scala b/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubFlow.scala
index 79d7ea4e43..2a0d3a036d 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubFlow.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubFlow.scala
@@ -1799,6 +1799,25 @@ class SubFlow[In, Out, Mat](
def flatMapMerge[T, M](breadth: Int, f: function.Function[Out, _ <: Graph[SourceShape[T], M]]): SubFlow[In, T, Mat] =
new SubFlow(delegate.flatMapMerge(breadth, o => f(o)))
+ /**
+ * 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").
+ *
+ * '''Emits when''' the current substream has an element available
+ *
+ * '''Backpressures when''' never
+ *
+ * '''Completes when''' upstream completes and the current substream completes
+ *
+ * '''Cancels when''' downstream cancels
+ *
+ * @since 1.2.0
+ */
+ def switchMap[T, M](f: function.Function[Out, _ <: Graph[SourceShape[T], M]]): SubFlow[In, T, Mat] =
+ new SubFlow(delegate.switchMap(o => f(o)))
+
/**
* Concatenate the given [[Source]] to this [[Flow]], meaning that once this
* Flow’s input is exhausted and all result elements have been generated,
diff --git a/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubSource.scala b/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubSource.scala
index 3092262b7d..3b3bbfbf7f 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubSource.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubSource.scala
@@ -1773,6 +1773,25 @@ class SubSource[Out, Mat](
def flatMapMerge[T, M](breadth: Int, f: function.Function[Out, _ <: Graph[SourceShape[T], M]]): SubSource[T, Mat] =
new SubSource(delegate.flatMapMerge(breadth, o => f(o)))
+ /**
+ * 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").
+ *
+ * '''Emits when''' the current substream has an element available
+ *
+ * '''Backpressures when''' never
+ *
+ * '''Completes when''' upstream completes and the current substream completes
+ *
+ * '''Cancels when''' downstream cancels
+ *
+ * @since 1.2.0
+ */
+ def switchMap[T, M](f: function.Function[Out, _ <: Graph[SourceShape[T], M]]): SubSource[T, Mat] =
+ new SubSource(delegate.switchMap(o => f(o)))
+
/**
* Concatenate the given [[Source]] to this [[Flow]], meaning that once this
* Flow’s input is exhausted and all result elements have been generated,
diff --git a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala
index a12a0cd9c1..83cc8f0704 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala
@@ -2804,6 +2804,25 @@ trait FlowOps[+Out, +Mat] {
def flattenMerge[T, M](breadth: Int)(implicit ev: Out <:< Graph[SourceShape[T], M]): Repr[T] =
flatMapMerge(breadth, ev)
+ /**
+ * 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").
+ *
+ * '''Emits when''' the current substream has an element available
+ *
+ * '''Backpressures when''' never
+ *
+ * '''Completes when''' upstream completes and the current substream completes
+ *
+ * '''Cancels when''' downstream cancels
+ *
+ * @since 1.2.0
+ */
+ def switchMap[T, M](f: Out => Graph[SourceShape[T], M]): Repr[T] =
+ 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
* with a [[org.apache.pekko.stream.InitialTimeoutException]].
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org