You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@pekko.apache.org by "MartinHH (via GitHub)" <gi...@apache.org> on 2025/03/05 14:20:07 UTC
[PR] feat: switchMap [pekko]
MartinHH opened a new pull request, #1787:
URL: https://github.com/apache/pekko/pull/1787
Implements #974
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add switchMap operator [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #1787:
URL: https://github.com/apache/pekko/pull/1787#issuecomment-2705649175
@pjfanning @MartinHH BTW, the Kotlin flow is using `flatmapLatest` as a `switchMap`, I'm not sure which name is better, maybe `switchMap` is good enough?
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: switchMap [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1787:
URL: https://github.com/apache/pekko/pull/1787#discussion_r1983220056
##########
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Switch.scala:
##########
@@ -0,0 +1,119 @@
+/*
+ * 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
+
+/**
+ * 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) {
+
+ var source = Option.empty[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 onUpstreamFinish(): Unit = if (source.isEmpty) completeStage()
+ })
+
+ setHandler(out,
+ new OutHandler {
+ override def onPull(): Unit = {
+ if (isAvailable(out)) tryPushOut()
+ }
+ })
+
+ def tryPushOut(): Unit = {
+ source.foreach { src =>
+ if (src.isAvailable) {
+ push(out, src.grab())
+ if (!src.isClosed) src.pull()
+ else removeCurrentSource(completeIfClosed = true)
+ }
+ }
+ }
+
+ def setSource(source: Graph[SourceShape[T], M]): Unit = {
+ cancelCurrentSource()
+ removeCurrentSource(completeIfClosed = false)
+ val sinkIn = new SubSinkInlet[T]("SwitchSink")
Review Comment:
I think we can avoid some materialization if the source is `SingleSource` or IterableSource.
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add switchMap operator [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1787:
URL: https://github.com/apache/pekko/pull/1787#discussion_r1983224383
##########
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Switch.scala:
##########
@@ -0,0 +1,119 @@
+/*
+ * 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
+
+/**
+ * INTERNAL API
+ */
+@InternalApi private[pekko] final class Switch[T, M]
Review Comment:
Maybe `SwitchMap`?
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: switchMap [pekko]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on code in PR #1787:
URL: https://github.com/apache/pekko/pull/1787#discussion_r1981566425
##########
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Switch.scala:
##########
@@ -0,0 +1,109 @@
+/*
+ * 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.
+ */
+package org.apache.pekko.stream.impl.fusing
+
+import org.apache.pekko.annotation.InternalApi
Review Comment:
see comment elsewhere about scala imports for `import org.apache.pekko`
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: switchMap [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1787:
URL: https://github.com/apache/pekko/pull/1787#discussion_r1983217646
##########
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Switch.scala:
##########
@@ -0,0 +1,119 @@
+/*
+ * 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
+
+/**
+ * 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) {
+
+ var source = Option.empty[SubSinkInlet[T]]
+
+ override def preStart(): Unit = {
+ pull(in)
+ super.preStart()
+ }
+
+ setHandler(in,
+ new InHandler {
Review Comment:
fold the in handler and the outhandler to the graphstagelogic to avoid some allocation
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add switchMap operator [pekko]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on code in PR #1787:
URL: https://github.com/apache/pekko/pull/1787#discussion_r1983940803
##########
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.
Review Comment:
it would be good to be more explicit in the docs about the behaviour
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add switchMap operator [pekko]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on code in PR #1787:
URL: https://github.com/apache/pekko/pull/1787#discussion_r1984575324
##########
docs/src/main/paradox/stream/operators/Source-or-Flow/switchMap.md:
##########
@@ -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").
Review Comment:
could you remove the 2nd space in `this operator`?
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add switchMap operator [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1787:
URL: https://github.com/apache/pekko/pull/1787#discussion_r1984586936
##########
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
Review Comment:
this `never` is really great!
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: switchMap [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #1787:
URL: https://github.com/apache/pekko/pull/1787#issuecomment-2703636619
Thank you for this. I have some pr that avoids the materialization of subSource, which you can refer to.
https://github.com/apache/pekko/pull/1775
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add switchMap operator [pekko]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on code in PR #1787:
URL: https://github.com/apache/pekko/pull/1787#discussion_r1983950766
##########
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Switch.scala:
##########
@@ -0,0 +1,119 @@
+/*
+ * 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
+
+/**
+ * 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) {
+
+ var source = Option.empty[SubSinkInlet[T]]
Review Comment:
https://github.com/apache/pekko/blob/ba05791fa0ea347a5c6ac417bdd6c080b305e363/actor/src/main/scala/org/apache/pekko/util/OptionVal.scala#L22
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: switchMap [pekko]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on PR #1787:
URL: https://github.com/apache/pekko/pull/1787#issuecomment-2701352933
needs
1. `sbt javafmtAll` -- some of the Java files are failing the style check
2. doc link issue
```
[03-05 15:14:56.422] [error] java.lang.RuntimeException: Unable to extract details from /home/runner/work/pekko/pekko/docs/src/main/paradox/stream/operators/Source-or-Flow/switchMap.md
[03-05 15:14:56.422] [error] at StreamOperatorsIndexGenerator$.getDetails(StreamOperatorsIndexGenerator.scala:281)
[03-05 15:14:56.422] [error] at StreamOperatorsIndexGenerator$.$anonfun$generateAlphabeticalIndex$13(StreamOperatorsIndexGenerator.scala:209)
[03-05 15:14:56.422] [error] at scala.collection.immutable.List.map(List.scala:297)
[03-05 15:14:56.422] [error] at StreamOperatorsIndexGenerator$.$anonfun$generateAlphabeticalIndex$1(StreamOperatorsIndexGenerator.scala:207)
[03-05 15:14:56.423] [error] at scala.Function1.$anonfun$compose$1(Function1.scala:49)
[03-05 15:14:56.423] [error] at sbt.internal.util.$tilde$greater.$anonfun$$u2219$1(TypeFunctions.scala:63)
[03-05 15:14:56.423] [error] at sbt.std.Transform$$anon$4.work(Transform.scala:69)
[03-05 15:14:56.423] [error] at sbt.Execute.$anonfun$submit$2(Execute.scala:283)
[03-05 15:14:56.423] [error] at sbt.internal.util.ErrorHandling$.wideConvert(ErrorHandling.scala:24)
[03-05 15:14:56.423] [error] at sbt.Execute.work(Execute.scala:292)
[03-05 15:14:56.423] [error] at sbt.Execute.$anonfun$submit$1(Execute.scala:283)
[03-05 15:14:56.423] [error] at sbt.ConcurrentRestrictions$$anon$4.$anonfun$submitValid$1(ConcurrentRestrictions.scala:265)
[03-05 15:14:56.423] [error] at sbt.CompletionService$$anon$2.call(CompletionService.scala:65)
[03-05 15:14:56.423] [error] at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317)
[03-05 15:14:56.423] [error] at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:572)
[03-05 15:14:56.423] [error] at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317)
[03-05 15:14:56.423] [error] at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
[03-05 15:14:56.423] [error] at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
[03-05 15:14:56.423] [error] at java.base/java.lang.Thread.run(Thread.java:1583)
[03-05 15:14:56.423] [error] Caused by: java.lang.IllegalArgumentException: requirement failed: category link in /home/runner/work/pekko/pekko/docs/src/main/paradox/stream/operators/Source-or-Flow/switchMap.md should start with @ref, but saw ""
[03-05 15:14:56.423] [error] at scala.Predef$.require(Predef.scala:281)
[03-05 15:14:56.423] [error] at StreamOperatorsIndexGenerator$.getDetails(StreamOperatorsIndexGenerator.scala:273)
[03-05 15:14:56.423] [error] at StreamOperatorsIndexGenerator$.$anonfun$generateAlphabeticalIndex$13(StreamOperatorsIndexGenerator.scala:209)
[03-05 15:14:56.423] [error] at scala.collection.immutable.List.map(List.scala:297)
[03-05 15:14:56.423] [error] at StreamOperatorsIndexGenerator$.$anonfun$generateAlphabeticalIndex$1(StreamOperatorsIndexGenerator.scala:207)
[03-05 15:14:56.423] [error] at scala.Function1.$anonfun$compose$1(Function1.scala:49)
[03-05 15:14:56.423] [error] at sbt.internal.util.$tilde$greater.$anonfun$$u2219$1(TypeFunctions.scala:63)
[03-05 15:14:56.423] [error] at sbt.std.Transform$$anon$4.work(Transform.scala:69)
[03-05 15:14:56.423] [error] at sbt.Execute.$anonfun$submit$2(Execute.scala:283)
[03-05 15:14:56.423] [error] at sbt.internal.util.ErrorHandling$.wideConvert(ErrorHandling.scala:24)
[03-05 15:14:56.423] [error] at sbt.Execute.work(Execute.scala:292)
[03-05 15:14:56.423] [error] at sbt.Execute.$anonfun$submit$1(Execute.scala:283)
[03-05 15:14:56.423] [error] at sbt.ConcurrentRestrictions$$anon$4.$anonfun$submitValid$1(ConcurrentRestrictions.scala:265)
[03-05 15:14:56.423] [error] at sbt.CompletionService$$anon$2.call(CompletionService.scala:65)
[03-05 15:14:56.423] [error] at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317)
[03-05 15:14:56.423] [error] at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:572)
[03-05 15:14:56.423] [error] at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317)
[03-05 15:14:56.423] [error] at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
[03-05 15:14:56.423] [error] at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
[03-05 15:14:56.423] [error] at java.base/java.lang.Thread.run(Thread.java:1583)
[03-05 15:14:56.433] [error] (docs / Compile / managedResources) Unable to extract details from /home/runner/work/pekko/pekko/docs/src/main/paradox/stream/operators/Source-or-Flow/switchMap.md
```
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add switchMap operator [pekko]
Posted by "MartinHH (via GitHub)" <gi...@apache.org>.
MartinHH commented on code in PR #1787:
URL: https://github.com/apache/pekko/pull/1787#discussion_r1984550609
##########
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Switch.scala:
##########
@@ -0,0 +1,119 @@
+/*
+ * 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
+
+/**
+ * INTERNAL API
+ */
+@InternalApi private[pekko] final class Switch[T, M]
Review Comment:
I'd be fine with renaming this, but maybe it's worth to explain my choice of name before we decide whether to rename:
- This `GraphStage` does not evaluate the lambda - similar to `flatMapMerge` and `flatMapConcat`, the evalation of `f` is delegated to the `map` operator and then only the "flattening" is done in the `GraphStage`. (My reasons for implementing it that way: consistency with the afforementioned operators and less redundancy in code because I did not need to care about things like `f` throwing.)
- For those other operators that delegate evaluation of `f` to `map`, the naming scheme seems to be to not have `Map` in the name of the `GraphStage` (the stage for `flatMapMerge` is called `FlattenMerge`)
- The transformation that is implemented by this `GraphStage` ("switching" without the "mapping" part) corresponds to what an operator that is named `switch` in both ReactiveX and Monix does
So following my train of thought, if this stage were named `SwitchMap` one should move the evaluation of `f` into the `GraphStage`.
Don't get me wrong: I'd be fine with any solution (keep it as it is; rename to `SwitchMap` and move evaluation of `f` into the `GraphStage` or just rename and keep evaluation of `f` where it is), I just wanted to share my reasoning before I make any changes.
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add switchMap operator [pekko]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on code in PR #1787:
URL: https://github.com/apache/pekko/pull/1787#discussion_r1984574544
##########
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).
Review Comment:
typo - should be flattened
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add switchMap operator [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #1787:
URL: https://github.com/apache/pekko/pull/1787#issuecomment-2705734055
@MartinHH Yes, I see the code in Spring Ai, which is using Flux#switchMap too. Thanks for the input.
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add switchMap operator [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1787:
URL: https://github.com/apache/pekko/pull/1787#discussion_r1984624469
##########
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Switch.scala:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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)
+ setSource(source)
+ tryPull(in)
+ }
+
+ override def onUpstreamFinish(): Unit = if (source.isEmpty) completeStage()
+
+ override def onPull(): Unit = {
+ if (isAvailable(out)) tryPushOut()
+ }
+
+ setHandler(in, this)
+ setHandler(out, this)
+
+ def tryPushOut(): Unit = {
+ if (source.isDefined) {
+ val src = source.get
+ if (src.isAvailable) {
+ push(out, src.grab())
+ if (!src.isClosed) src.pull()
+ else removeCurrentSource(completeIfClosed = true)
+ }
+ }
+ }
+
+ def setSource(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 = {
+ if (source.isDefined) {
+ source.get.cancel()
+ }
+ }
+
+ override def postStop(): Unit = cancelCurrentSource()
Review Comment:
I think we should cancel the current source in `onDownStreamFinished` instead of here?
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add switchMap operator [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1787:
URL: https://github.com/apache/pekko/pull/1787#discussion_r1984625614
##########
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Switch.scala:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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)
+ setSource(source)
+ tryPull(in)
+ }
+
+ override def onUpstreamFinish(): Unit = if (source.isEmpty) completeStage()
+
+ override def onPull(): Unit = {
+ if (isAvailable(out)) tryPushOut()
+ }
+
+ setHandler(in, this)
+ setHandler(out, this)
+
+ def tryPushOut(): Unit = {
+ if (source.isDefined) {
+ val src = source.get
+ if (src.isAvailable) {
+ push(out, src.grab())
+ if (!src.isClosed) src.pull()
+ else removeCurrentSource(completeIfClosed = true)
+ }
+ }
+ }
+
+ def setSource(source: Graph[SourceShape[T], M]): Unit = {
Review Comment:
`setupCurrentSource` maybe a better name?
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: switchMap [pekko]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on code in PR #1787:
URL: https://github.com/apache/pekko/pull/1787#discussion_r1981563391
##########
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Switch.scala:
##########
@@ -0,0 +1,109 @@
+/*
+ * 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
Review Comment:
brand new source files with no copying from existing source files should have the standard Apache source header not this one.
https://github.com/apache/pekko/blob/main/project/AddMetaInfLicenseFiles.scala#L1-L16
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add switchMap operator [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1787:
URL: https://github.com/apache/pekko/pull/1787#discussion_r1984623132
##########
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Switch.scala:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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)
+ setSource(source)
+ tryPull(in)
+ }
+
+ override def onUpstreamFinish(): Unit = if (source.isEmpty) completeStage()
+
+ override def onPull(): Unit = {
+ if (isAvailable(out)) tryPushOut()
+ }
+
+ setHandler(in, this)
+ setHandler(out, this)
+
+ def tryPushOut(): Unit = {
+ if (source.isDefined) {
+ val src = source.get
+ if (src.isAvailable) {
+ push(out, src.grab())
+ if (!src.isClosed) src.pull()
+ else removeCurrentSource(completeIfClosed = true)
+ }
+ }
+ }
+
+ def setSource(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 = {
+ if (source.isDefined) {
Review Comment:
better with pattern matching
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add switchMap operator [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1787:
URL: https://github.com/apache/pekko/pull/1787#discussion_r1984621694
##########
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Switch.scala:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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)
+ setSource(source)
+ tryPull(in)
+ }
+
+ override def onUpstreamFinish(): Unit = if (source.isEmpty) completeStage()
+
+ override def onPull(): Unit = {
+ if (isAvailable(out)) tryPushOut()
+ }
+
+ setHandler(in, this)
+ setHandler(out, this)
Review Comment:
setHandlers(in, out, this)
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add switchMap operator [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1787:
URL: https://github.com/apache/pekko/pull/1787#discussion_r1984619549
##########
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))))
Review Comment:
You implemented it with a `map` and `switch` cool.
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add switchMap operator [pekko]
Posted by "MartinHH (via GitHub)" <gi...@apache.org>.
MartinHH commented on code in PR #1787:
URL: https://github.com/apache/pekko/pull/1787#discussion_r1986085553
##########
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Switch.scala:
##########
@@ -0,0 +1,119 @@
+/*
+ * 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
+
+/**
+ * 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
Review Comment:
I implemented it, but not within this file (because the actual lambda `f` is not available here): https://github.com/MartinHH/pekko/blob/90c8ea8b7aca4ecd0c5e39b94e101174d8ddd617/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala#L2824
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: switchMap [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1787:
URL: https://github.com/apache/pekko/pull/1787#discussion_r1983219028
##########
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Switch.scala:
##########
@@ -0,0 +1,119 @@
+/*
+ * 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
+
+/**
+ * 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) {
+
+ var source = Option.empty[SubSinkInlet[T]]
Review Comment:
use OptionVal to avoid boxing.
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: switchMap [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1787:
URL: https://github.com/apache/pekko/pull/1787#discussion_r1983216569
##########
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.
Review Comment:
and then the old stream is been cancelled.
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: switchMap [pekko]
Posted by "MartinHH (via GitHub)" <gi...@apache.org>.
MartinHH commented on code in PR #1787:
URL: https://github.com/apache/pekko/pull/1787#discussion_r1981503698
##########
stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowSwitchSpec.scala:
##########
@@ -0,0 +1,297 @@
+/*
+ * 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>
+ */
Review Comment:
I'm not sure if this is needed, but: a good fraction of my test code is a result of copy/paste/modify from `org.apache.pekko.stream.scaladsl.FlowFlattenMergeSpec` which is code "inherited" from Lightbend. (See various comments `// copied (with modifications) from org.apache.pekko.stream.scaladsl.FlowFlattenMergeSpec` below.)
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add switchMap operator [pekko]
Posted by "MartinHH (via GitHub)" <gi...@apache.org>.
MartinHH commented on PR #1787:
URL: https://github.com/apache/pekko/pull/1787#issuecomment-2706749144
I submitted another set of suggested changes. Looking at the sunny weather forecast, I doubt that I will get to figure out how to properly implement avoiding materialization for certain sources in the coming days, so I'd suggest to merge if this my latest changes get approval.
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add switchMap operator [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #1787:
URL: https://github.com/apache/pekko/pull/1787#issuecomment-2705384610
I'm Ok with the current shape too, which we can polish it later, but let's wait a little longer for @MartinHH if he has some more updates, wdyt.
I would also like to have https://github.com/apache/pekko/pull/1702 later If you have time @pjfanning .
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add switchMap operator [pekko]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on PR #1787:
URL: https://github.com/apache/pekko/pull/1787#issuecomment-2704757734
@He-Pin @MartinHH I am comfortable with merging this with just maybe the rename of the Switch class to SwitchMap and then following up with an optimisation PR that avoids allocations (but uses longer and more complicated code).
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: switchMap [pekko]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on code in PR #1787:
URL: https://github.com/apache/pekko/pull/1787#discussion_r1981559570
##########
stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowSwitchSpec.scala:
##########
@@ -0,0 +1,297 @@
+/*
+ * 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.NotUsed
Review Comment:
one nit is that the project code style is to `import org.apache.pekko` and then the subsequent imports can begin `import pekko.stream...` (ie omit the org.apache bit).
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: switchMap [pekko]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on code in PR #1787:
URL: https://github.com/apache/pekko/pull/1787#discussion_r1981554945
##########
stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowSwitchSpec.scala:
##########
@@ -0,0 +1,297 @@
+/*
+ * 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>
+ */
Review Comment:
Yes - if you copy code from a class that has headers like this into a new source file then that source file needs the headers from the original file(s).
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add switchMap operator [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1787:
URL: https://github.com/apache/pekko/pull/1787#discussion_r1983223893
##########
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Switch.scala:
##########
@@ -0,0 +1,119 @@
+/*
+ * 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
+
+/**
+ * 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
Review Comment:
better with the lambda line number too.
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add switchMap operator [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin merged PR #1787:
URL: https://github.com/apache/pekko/pull/1787
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: switchMap [pekko]
Posted by "MartinHH (via GitHub)" <gi...@apache.org>.
MartinHH commented on PR #1787:
URL: https://github.com/apache/pekko/pull/1787#issuecomment-2701399453
> needs
>
> 1. `sbt javafmtAll` -- some of the Java files are failing the style check
>
> 2. doc link issue
Thanks - and sorry for choosing the "submit your PR straight away ... without reading the rest of this wonderful doc" route.
😉
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add switchMap operator [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #1787:
URL: https://github.com/apache/pekko/pull/1787#issuecomment-2708214403
@MartinHH, I think we can merge this. I can prepare to continue a PR after your excellent work.
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add switchMap operator [pekko]
Posted by "MartinHH (via GitHub)" <gi...@apache.org>.
MartinHH commented on code in PR #1787:
URL: https://github.com/apache/pekko/pull/1787#discussion_r1985255352
##########
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Switch.scala:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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)
+ setSource(source)
+ tryPull(in)
+ }
+
+ override def onUpstreamFinish(): Unit = if (source.isEmpty) completeStage()
+
+ override def onPull(): Unit = {
+ if (isAvailable(out)) tryPushOut()
+ }
+
+ setHandler(in, this)
+ setHandler(out, this)
+
+ def tryPushOut(): Unit = {
+ if (source.isDefined) {
+ val src = source.get
+ if (src.isAvailable) {
+ push(out, src.grab())
+ if (!src.isClosed) src.pull()
+ else removeCurrentSource(completeIfClosed = true)
+ }
+ }
+ }
+
+ def setSource(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 = {
+ if (source.isDefined) {
Review Comment:
I stopped being passionate about such things so I just followed your suggestion, although I do not agree 😉 .
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add switchMap operator [pekko]
Posted by "MartinHH (via GitHub)" <gi...@apache.org>.
MartinHH commented on code in PR #1787:
URL: https://github.com/apache/pekko/pull/1787#discussion_r1984572244
##########
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Switch.scala:
##########
@@ -0,0 +1,119 @@
+/*
+ * 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
+
+/**
+ * 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) {
+
+ var source = Option.empty[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 onUpstreamFinish(): Unit = if (source.isEmpty) completeStage()
+ })
+
+ setHandler(out,
+ new OutHandler {
+ override def onPull(): Unit = {
+ if (isAvailable(out)) tryPushOut()
+ }
+ })
+
+ def tryPushOut(): Unit = {
+ source.foreach { src =>
+ if (src.isAvailable) {
+ push(out, src.grab())
+ if (!src.isClosed) src.pull()
+ else removeCurrentSource(completeIfClosed = true)
+ }
+ }
+ }
+
+ def setSource(source: Graph[SourceShape[T], M]): Unit = {
+ cancelCurrentSource()
+ removeCurrentSource(completeIfClosed = false)
+ val sinkIn = new SubSinkInlet[T]("SwitchSink")
Review Comment:
I had a brief shot at that before I submitted the PR but then refrained from that because my initial naive approach lead to inconsistencies (my test case "not behave differently for substreams Source.single(x) and Source(List(x))" results from that).
The thing I found challenging: I'd expect that as long as further values are available from upstream, any previous values from upstream are discarded without emitting a single item.
For example, the following should and currently does result in Seq(2, 3) (with the 1 being discarded):
```scala
Source(List(Source(List(1)), Source(List(2, 3))))
.switchMap(identity)
.runWith(Sink.seq)
```
I'd expect the same for
```scala
Source(List(Source.single(1), Source(List(2, 3))))
.switchMap(identity)
.runWith(Sink.seq)
```
That means that for example, I detect a `SingleSource` as current substream, I cannot just emit the single value, but need to check wether further items are available from upstream before doing that (and discard the single value if there are).
At the point where I had understood that, I decided to keep the initial implementation simple.
Now that almost everything else seems to be more or less agreed on, I might try to give it another shot sometime during the coming week and see if I can get that optimization done without breaking the above expectations (maybe one of you has some pointers?). But I might end up at a point where I'd suggest to merge this at is it is (then hoping for someone else to get that optoimization done).
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add switchMap operator [pekko]
Posted by "MartinHH (via GitHub)" <gi...@apache.org>.
MartinHH commented on code in PR #1787:
URL: https://github.com/apache/pekko/pull/1787#discussion_r1985256204
##########
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Switch.scala:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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)
+ setSource(source)
+ tryPull(in)
+ }
+
+ override def onUpstreamFinish(): Unit = if (source.isEmpty) completeStage()
+
+ override def onPull(): Unit = {
+ if (isAvailable(out)) tryPushOut()
+ }
+
+ setHandler(in, this)
+ setHandler(out, this)
+
+ def tryPushOut(): Unit = {
+ if (source.isDefined) {
+ val src = source.get
+ if (src.isAvailable) {
+ push(out, src.grab())
+ if (!src.isClosed) src.pull()
+ else removeCurrentSource(completeIfClosed = true)
+ }
+ }
+ }
+
+ def setSource(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 = {
+ if (source.isDefined) {
+ source.get.cancel()
+ }
+ }
+
+ override def postStop(): Unit = cancelCurrentSource()
Review Comment:
I'm not sure. I followed the example of `FlattenMerge` where the cancelling of substreams is done in `postStop()`: https://github.com/apache/pekko/blob/5f64ccbedd97602941b04472c0c72ff5308979c5/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/StreamOfStreams.scala#L146 (and `onDownStreamFinished` is not overriden).
I'd expect that `postStop` will eventually be called in any case and don't see a risk of `cancelCurrentSource()` being called from here when it shouldn't, so doing it in this "catch-all" handler feels safer to me. (But maybe I'm just missing the significant difference?)
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add switchMap operator [pekko]
Posted by "MartinHH (via GitHub)" <gi...@apache.org>.
MartinHH commented on PR #1787:
URL: https://github.com/apache/pekko/pull/1787#issuecomment-2708307656
Ok, I think I'm done here, so go ahead and merge as you feel.
Btw: thanks for the welcoming experience on here!
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add switchMap operator [pekko]
Posted by "MartinHH (via GitHub)" <gi...@apache.org>.
MartinHH commented on code in PR #1787:
URL: https://github.com/apache/pekko/pull/1787#discussion_r1984576698
##########
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.
Review Comment:
I was struggling to find the right trade of between brevety (after all, there is a one-line-limit enforced for this part here) and explicitness. I now added a little bit more to this line and even more in the desciption section below.
(I'd be very open for any specific wording suggestions if you feel the updated version is not clear enough.)
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add switchMap operator [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #1787:
URL: https://github.com/apache/pekko/pull/1787#issuecomment-2708478072
Thank you very much, it completes the pekko stream for ai scenarios.
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add switchMap operator [pekko]
Posted by "MartinHH (via GitHub)" <gi...@apache.org>.
MartinHH commented on code in PR #1787:
URL: https://github.com/apache/pekko/pull/1787#discussion_r1984550609
##########
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Switch.scala:
##########
@@ -0,0 +1,119 @@
+/*
+ * 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
+
+/**
+ * INTERNAL API
+ */
+@InternalApi private[pekko] final class Switch[T, M]
Review Comment:
I'd be fine with renaming this, but maybe it's worth to explain my choice of name before we decide whether to rename:
- This `GraphStage` does not evaluate the lambda - similar to `flatMapMerge` and `flatMapConcat`, the evalation of `f` is delegated to the `map` operator and then only the "flattening" is done in the `GraphStage`. (My reasons for implementing it that way: consistency with the afforementioned operators and less redundancy in code because I did not need to care about things like `f` throwing.)
- For those other operators that delegate evaluation of `f` to `map`, the naming scheme seems to be to not have `Map` in the name of the `GraphStage` (the stage for `flatMapMerge` is called `FlattenMerge`)
- The transformation that is implemented by this `GraphStage` ("switching" without the "mapping" part) corresponds to what an operator that is named `switch` in both ReactiveX and Monix does
So following my thrain of thought, if this stage were named `SwitchMap` one should move the evaluation of `f` into the `GraphStage`.
Don't get me wrong: I'd be fine with any solution (keep it as it is; rename to `SwitchMap` and move evaluation of `f` into the `GraphStage` or just rename and keep evaluation of `f` where it is), I just wanted to share my reasoning before I make any changes.
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add switchMap operator [pekko]
Posted by "MartinHH (via GitHub)" <gi...@apache.org>.
MartinHH commented on code in PR #1787:
URL: https://github.com/apache/pekko/pull/1787#discussion_r1986040387
##########
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
Review Comment:
On second thought: this might be formally correct as it never backpressures on upstream (and backpressure on upstream is what is to be adressed here if I'm not mistaken), but maybe it would be more helpful to mention the backpressure on substream, e.g.:
```suggestion
* '''Backpressures when''' never on upstream (backpressures on current substream when downstream backpressures)
```
wdyt?
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add switchMap operator [pekko]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on code in PR #1787:
URL: https://github.com/apache/pekko/pull/1787#discussion_r1986083114
##########
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Switch.scala:
##########
@@ -0,0 +1,119 @@
+/*
+ * 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
+
+/**
+ * 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
Review Comment:
@He-Pin do you still want this change?
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add switchMap operator [pekko]
Posted by "MartinHH (via GitHub)" <gi...@apache.org>.
MartinHH commented on PR #1787:
URL: https://github.com/apache/pekko/pull/1787#issuecomment-2705688990
> @pjfanning @MartinHH BTW, the Kotlin flow is using `flatmapLatest` as a `switchMap`, I'm not sure which name is better, maybe `switchMap` is good enough?
I'm kind of biased to the naming from certain libraries (RxScala, Monix) because I learned about all those stream operators back when the official scala coursera courses had Eik Meijer teach RxScala, so for me, this is "switchMap". Given that Rx is still very popular in certain ecosystems (e.g. for Js/Ts Frontend), I'd say one can't go wrong follwing their naming, but then again: I don't really mind.
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org