raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 10

chore: Clean up some duplicated code.

Browse source 
---------

Co-authored-by: Matthew de Detrich <matthew.dedetrich@aiven.io>
This commit is contained in:
He-Pin(kerr) 2024-01-20 16:38:13 +08:00 • committed by GitHub
parent b1ec854b8c
commit 637d72af7a
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 8 additions and 6 deletions
Download patch file Download diff file Expand all files Collapse all files

14
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Ops.scala
View file

@ -193,9 +193,8 @@ import pekko.util.ccompat._
} catch {
case NonFatal(ex) =>
decider(ex) match {
case Supervision.Stop => failStage(ex)
case Supervision.Resume => if (!hasBeenPulled(in)) pull(in)
case Supervision.Restart => if (!hasBeenPulled(in)) pull(in)
case Supervision.Stop => failStage(ex)
case _ => pull(in)
}
}
@ -267,9 +266,12 @@ private[stream] object Collect {
} catch {
case NonFatal(ex) =>
decider(ex) match {
case Supervision.Stop => failStage(ex)
case Supervision.Resume => if (!hasBeenPulled(in)) pull(in)
case Supervision.Restart => if (!hasBeenPulled(in)) pull(in)
case Supervision.Stop => failStage(ex)
case _ =>
// The !hasBeenPulled(in) check is not required here since it
// isn't possible to do an additional pull(in) due to the nature
// of how collect works
pull(in)
}
}