You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pekko.apache.org by he...@apache.org on 2023/09/03 10:41:34 UTC
[incubator-pekko] branch main updated: =str Fold InHandler and OutHandler into GraphStageLogic for recoverWith.
This is an automated email from the ASF dual-hosted git repository.
hepin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-pekko.git
The following commit(s) were added to refs/heads/main by this push:
new 194602c849 =str Fold InHandler and OutHandler into GraphStageLogic for recoverWith.
194602c849 is described below
commit 194602c849ac8469ded9f368971df5312e8d6c6c
Author: He-Pin <he...@alibaba-inc.com>
AuthorDate: Sat Sep 2 21:35:28 2023 +0800
=str Fold InHandler and OutHandler into GraphStageLogic for recoverWith.
---
.../org/apache/pekko/stream/impl/fusing/Ops.scala | 61 +++++++++-------------
1 file changed, 26 insertions(+), 35 deletions(-)
diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Ops.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Ops.scala
index 3d9ff981e2..6995ae114c 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Ops.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Ops.scala
@@ -2157,52 +2157,43 @@ private[pekko] object TakeWithin {
val pf: PartialFunction[Throwable, Graph[SourceShape[T], M]])
extends SimpleLinearGraphStage[T] {
- override def initialAttributes = DefaultAttributes.recoverWith
+ override def initialAttributes: Attributes = DefaultAttributes.recoverWith
- override def createLogic(attr: Attributes) = new GraphStageLogic(shape) {
- var attempt = 0
-
- setHandler(in,
- new InHandler {
- override def onPush(): Unit = push(out, grab(in))
-
- override def onUpstreamFailure(ex: Throwable) = onFailure(ex)
- })
-
- setHandler(out,
- new OutHandler {
- override def onPull(): Unit = pull(in)
- })
+ override def createLogic(attr: Attributes) =
+ new GraphStageLogic(shape) with InHandler with OutHandler {
+ var attempt = 0
- def onFailure(ex: Throwable) =
- if ((maximumRetries < 0 || attempt < maximumRetries) && pf.isDefinedAt(ex)) {
- switchTo(pf(ex))
- attempt += 1
- } else
- failStage(ex)
+ override def onPush(): Unit = push(out, grab(in))
+ override def onUpstreamFailure(ex: Throwable): Unit = onFailure(ex)
- def switchTo(source: Graph[SourceShape[T], M]): Unit = {
- val sinkIn = new SubSinkInlet[T]("RecoverWithSink")
+ override def onPull(): Unit = pull(in)
- sinkIn.setHandler(new InHandler {
- override def onPush(): Unit = push(out, sinkIn.grab())
+ def onFailure(ex: Throwable): Unit =
+ if ((maximumRetries < 0 || attempt < maximumRetries) && pf.isDefinedAt(ex)) {
+ switchTo(pf(ex))
+ attempt += 1
+ } else
+ failStage(ex)
- override def onUpstreamFinish(): Unit = completeStage()
+ def switchTo(source: Graph[SourceShape[T], M]): Unit = {
+ val sinkIn = new SubSinkInlet[T]("RecoverWithSink") with InHandler with OutHandler { self =>
+ override def onPush(): Unit = push(out, self.grab())
+ override def onUpstreamFinish(): Unit = completeStage()
+ override def onUpstreamFailure(ex: Throwable): Unit = onFailure(ex)
- override def onUpstreamFailure(ex: Throwable) = onFailure(ex)
- })
+ override def onPull(): Unit = self.pull()
+ override def onDownstreamFinish(cause: Throwable): Unit = self.cancel(cause)
- val outHandler = new OutHandler {
- override def onPull(): Unit = sinkIn.pull()
+ setHandler(self)
+ }
- override def onDownstreamFinish(cause: Throwable): Unit = sinkIn.cancel(cause)
+ Source.fromGraph(source).runWith(sinkIn.sink)(interpreter.subFusingMaterializer)
+ setHandler(out, sinkIn)
+ if (isAvailable(out)) sinkIn.pull()
}
- Source.fromGraph(source).runWith(sinkIn.sink)(interpreter.subFusingMaterializer)
- setHandler(out, outHandler)
- if (isAvailable(out)) sinkIn.pull()
+ setHandlers(in, out, this)
}
- }
override def toString: String = "RecoverWith"
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org