feat: Add switchMap operator

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

@@ -0,0 +1,29 @@
+# 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.
+
+@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 (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
+
+@@@
+
+

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)

stream-tests/src/test/java/org/apache/pekko/stream/javadsl/FlowTest.java

@@ -787,6 +787,32 @@ public void mustBeAbleToUseFlatMapMerge() throws Exception {
     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);

stream-tests/src/test/java/org/apache/pekko/stream/javadsl/SourceTest.java

@@ -509,6 +509,30 @@ public void mustBeAbleToUseFlatMapMerge() throws Exception {
     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);