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-unsubscr...@pekko.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org --------------------------------------------------------------------- To unsubscribe, e-mail: notifications-unsubscr...@pekko.apache.org For additional commands, e-mail: notifications-h...@pekko.apache.org
