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 2025/10/19 16:38:37 UTC
(pekko) 01/01: feat: Add missing onErrorResume to SubFlow and SubSource (#2337)
This is an automated email from the ASF dual-hosted git repository.
hepin pushed a commit to branch 1.3.x-onErrorResume
in repository https://gitbox.apache.org/repos/asf/pekko.git
commit ca32bf248955144fbe3ccf049f7877295e15c360
Author: He-Pin(kerr) <he...@gmail.com>
AuthorDate: Mon Oct 20 00:26:12 2025 +0800
feat: Add missing onErrorResume to SubFlow and SubSource (#2337)
(cherry picked from commit d23bab48e792544af28cff0932d94a00739e64b5)
# Conflicts:
# stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala
---
.../org/apache/pekko/stream/javadsl/Flow.scala | 10 +--
.../org/apache/pekko/stream/javadsl/Source.scala | 7 +-
.../org/apache/pekko/stream/javadsl/SubFlow.scala | 76 +++++++++++++++++++++
.../apache/pekko/stream/javadsl/SubSource.scala | 77 ++++++++++++++++++++++
4 files changed, 163 insertions(+), 7 deletions(-)
diff --git a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala
index 6054ebc09a..76582a7302 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala
@@ -24,6 +24,7 @@ import scala.annotation.unchecked.uncheckedVariance
import scala.collection.immutable
import scala.concurrent.duration.FiniteDuration
import scala.reflect.ClassTag
+import scala.util.control.NonFatal
import org.apache.pekko
import pekko.Done
@@ -44,6 +45,7 @@ import pekko.util.OptionConverters._
import pekko.util.Timeout
import pekko.util.unused
import pekko.util.ccompat._
+
import org.reactivestreams.Processor
object Flow {
@@ -2366,7 +2368,7 @@ final class Flow[In, Out, Mat](delegate: scaladsl.Flow[In, Out, Mat]) extends Gr
*/
def onErrorResume[T >: Out](fallback: function.Function[_ >: Throwable, _ <: Graph[SourceShape[T], NotUsed]])
: javadsl.Flow[In, T, Mat] = new Flow(delegate.recoverWith {
- case ex: Throwable => fallback(ex)
+ case NonFatal(ex) => fallback(ex)
})
/**
@@ -2392,7 +2394,7 @@ final class Flow[In, Out, Mat](delegate: scaladsl.Flow[In, Out, Mat]) extends Gr
clazz: Class[_ <: Throwable],
fallback: function.Function[_ >: Throwable, _ <: Graph[SourceShape[T], NotUsed]])
: javadsl.Flow[In, T, Mat] = new Flow(delegate.recoverWith {
- case ex: Throwable if clazz.isInstance(ex) => fallback(ex)
+ case NonFatal(ex) if clazz.isInstance(ex) => fallback(ex)
})
/**
@@ -2411,14 +2413,14 @@ final class Flow[In, Out, Mat](delegate: scaladsl.Flow[In, Out, Mat]) extends Gr
* '''Cancels when''' downstream cancels
*
* @param predicate Predicate which determines if the exception should be handled
- * @param function Function which produces a Source to continue the stream
+ * @param fallback Function which produces a Source to continue the stream
* @since 1.3.0
*/
def onErrorResume[T >: Out](
predicate: function.Predicate[_ >: Throwable],
fallback: function.Function[_ >: Throwable, _ <: Graph[SourceShape[T], NotUsed]])
: javadsl.Flow[In, T, Mat] = new Flow(delegate.recoverWith {
- case ex: Throwable if predicate.test(ex) => fallback(ex)
+ case NonFatal(ex) if predicate.test(ex) => fallback(ex)
})
/**
diff --git a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala
index 70eefdf3d0..b962114b7f 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala
@@ -24,6 +24,7 @@ import scala.collection.immutable
import scala.concurrent.{ Future, Promise }
import scala.concurrent.duration.FiniteDuration
import scala.reflect.ClassTag
+import scala.util.control.NonFatal
import org.apache.pekko
import pekko.{ Done, NotUsed }
@@ -2637,7 +2638,7 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[
def onErrorResume[T >: Out](
fallback: function.Function[_ >: Throwable, _ <: Graph[SourceShape[T], NotUsed]]): javadsl.Source[T, Mat] =
new Source(delegate.recoverWith {
- case ex: Throwable => fallback(ex)
+ case NonFatal(ex) => fallback(ex)
})
/**
@@ -2663,7 +2664,7 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[
clazz: Class[_ <: Throwable],
fallback: function.Function[_ >: Throwable, _ <: Graph[SourceShape[T], NotUsed]]): javadsl.Source[T, Mat] =
new Source(delegate.recoverWith {
- case ex: Throwable if clazz.isInstance(ex) => fallback(ex)
+ case NonFatal(ex) if clazz.isInstance(ex) => fallback(ex)
})
/**
@@ -2689,7 +2690,7 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[
predicate: function.Predicate[_ >: Throwable],
fallback: function.Function[_ >: Throwable, _ <: Graph[SourceShape[T], NotUsed]]): javadsl.Source[T, Mat] =
new Source(delegate.recoverWith {
- case ex: Throwable if predicate.test(ex) => fallback(ex)
+ case NonFatal(ex) if predicate.test(ex) => fallback(ex)
})
/**
diff --git a/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubFlow.scala b/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubFlow.scala
index 72e1f77691..60b1d09d7a 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubFlow.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubFlow.scala
@@ -22,6 +22,7 @@ import scala.annotation.unchecked.uncheckedVariance
import scala.collection.immutable
import scala.concurrent.duration.FiniteDuration
import scala.reflect.ClassTag
+import scala.util.control.NonFatal
import org.apache.pekko
import pekko.NotUsed
@@ -1546,6 +1547,81 @@ class SubFlow[In, Out, Mat](
errorConsumer: function.Procedure[_ >: Throwable]): SubFlow[In, Out, Mat] =
new SubFlow(delegate.onErrorContinue(p.test(_))(errorConsumer.apply(_)))
+ /**
+ * Transform a failure signal into a Source of elements provided by a factory function.
+ * This allows to continue processing with another stream when a failure occurs.
+ *
+ * Since the underlying failure signal onError arrives out-of-band, it might jump over existing elements.
+ * This operator can recover the failure signal, but not the skipped elements, which will be dropped.
+ *
+ * '''Emits when''' element is available from the upstream or upstream is failed and fallback Source produces an element
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' upstream completes or upstream failed with exception and fallback Source completes
+ *
+ * '''Cancels when''' downstream cancels
+ *
+ * @param fallback Function which produces a Source to continue the stream
+ * @since 1.3.0
+ */
+ def onErrorResume[T >: Out](
+ fallback: function.Function[_ >: Throwable, _ <: Graph[SourceShape[T], NotUsed]]): SubFlow[In, T, Mat] =
+ new SubFlow(delegate.recoverWith {
+ case NonFatal(ex) => fallback(ex)
+ })
+
+ /**
+ * Transform a failure signal into a stream of elements provided by a factory function.
+ * This allows to continue processing with another stream when a failure occurs.
+ *
+ * Since the underlying failure signal onError arrives out-of-band, it might jump over existing elements.
+ * This operator can recover the failure signal, but not the skipped elements, which will be dropped.
+ *
+ * '''Emits when''' element is available from the upstream or upstream is failed and fallback Source produces an element
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' upstream completes or upstream failed with exception and fallback Source completes
+ *
+ * '''Cancels when''' downstream cancels
+ *
+ * @param clazz the class object of the failure cause
+ * @param fallback Function which produces a Source to continue the stream
+ * @since 1.3.0
+ */
+ def onErrorResume[T >: Out](clazz: Class[_ <: Throwable],
+ fallback: function.Function[_ >: Throwable, _ <: Graph[SourceShape[T], NotUsed]]): SubFlow[In, T, Mat] =
+ new SubFlow(delegate.recoverWith {
+ case NonFatal(ex) if clazz.isInstance(ex) => fallback(ex)
+ })
+
+ /**
+ * Transform a failure signal into a stream of elements provided by a factory function.
+ * This allows to continue processing with another stream when a failure occurs.
+ *
+ * Since the underlying failure signal onError arrives out-of-band, it might jump over existing elements.
+ * This operator can recover the failure signal, but not the skipped elements, which will be dropped.
+ *
+ * '''Emits when''' element is available from the upstream or upstream is failed and fallback Source produces an element
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' upstream completes or upstream failed with exception and fallback Source completes
+ *
+ * '''Cancels when''' downstream cancels
+ *
+ * @param predicate Predicate which determines if the exception should be handled
+ * @param fallback Function which produces a Source to continue the stream
+ * @since 1.3.0
+ */
+ def onErrorResume[T >: Out](
+ predicate: function.Predicate[_ >: Throwable],
+ fallback: function.Function[_ >: Throwable, _ <: Graph[SourceShape[T], NotUsed]]): SubFlow[In, T, Mat] =
+ new SubFlow(delegate.recoverWith {
+ case NonFatal(ex) if predicate.test(ex) => fallback(ex)
+ })
+
/**
* While similar to [[recover]] this operator can be used to transform an error signal to a different one *without* logging
* it as an error in the process. So in that sense it is NOT exactly equivalent to `recover(t => throw t2)` since recover
diff --git a/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubSource.scala b/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubSource.scala
index ca02f9c9cf..8a9df8401b 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubSource.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubSource.scala
@@ -22,6 +22,7 @@ import scala.annotation.unchecked.uncheckedVariance
import scala.collection.immutable
import scala.concurrent.duration.FiniteDuration
import scala.reflect.ClassTag
+import scala.util.control.NonFatal
import org.apache.pekko
import pekko.NotUsed
@@ -1521,6 +1522,82 @@ class SubSource[Out, Mat](
errorConsumer: function.Procedure[_ >: Throwable]): SubSource[Out, Mat] =
new SubSource(delegate.onErrorContinue(p.test(_))(errorConsumer.apply(_)))
+ /**
+ * Transform a failure signal into a Source of elements provided by a factory function.
+ * This allows to continue processing with another stream when a failure occurs.
+ *
+ * Since the underlying failure signal onError arrives out-of-band, it might jump over existing elements.
+ * This operator can recover the failure signal, but not the skipped elements, which will be dropped.
+ *
+ * '''Emits when''' element is available from the upstream or upstream is failed and fallback Source produces an element
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' upstream completes or upstream failed with exception and fallback Source completes
+ *
+ * '''Cancels when''' downstream cancels
+ *
+ * @param fallback Function which produces a Source to continue the stream
+ * @since 1.3.0
+ */
+ def onErrorResume[T >: Out](
+ fallback: function.Function[_ >: Throwable, _ <: Graph[SourceShape[T], NotUsed]]): SubSource[T, Mat] =
+ new SubSource(delegate.recoverWith {
+ case NonFatal(ex) => fallback(ex)
+ })
+
+ /**
+ * Transform a failure signal into a stream of elements provided by a factory function.
+ * This allows to continue processing with another stream when a failure occurs.
+ *
+ * Since the underlying failure signal onError arrives out-of-band, it might jump over existing elements.
+ * This operator can recover the failure signal, but not the skipped elements, which will be dropped.
+ *
+ * '''Emits when''' element is available from the upstream or upstream is failed and fallback Source produces an element
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' upstream completes or upstream failed with exception and fallback Source completes
+ *
+ * '''Cancels when''' downstream cancels
+ *
+ * @param clazz the class object of the failure cause
+ * @param fallback Function which produces a Source to continue the stream
+ * @since 1.3.0
+ */
+ def onErrorResume[T >: Out](
+ clazz: Class[_ <: Throwable],
+ fallback: function.Function[_ >: Throwable, _ <: Graph[SourceShape[T], NotUsed]]): SubSource[T, Mat] =
+ new SubSource(delegate.recoverWith {
+ case NonFatal(ex) if clazz.isInstance(ex) => fallback(ex)
+ })
+
+ /**
+ * Transform a failure signal into a stream of elements provided by a factory function.
+ * This allows to continue processing with another stream when a failure occurs.
+ *
+ * Since the underlying failure signal onError arrives out-of-band, it might jump over existing elements.
+ * This operator can recover the failure signal, but not the skipped elements, which will be dropped.
+ *
+ * '''Emits when''' element is available from the upstream or upstream is failed and fallback Source produces an element
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' upstream completes or upstream failed with exception and fallback Source completes
+ *
+ * '''Cancels when''' downstream cancels
+ *
+ * @param predicate Predicate which determines if the exception should be handled
+ * @param fallback Function which produces a Source to continue the stream
+ * @since 1.3.0
+ */
+ def onErrorResume[T >: Out](
+ predicate: function.Predicate[_ >: Throwable],
+ fallback: function.Function[_ >: Throwable, _ <: Graph[SourceShape[T], NotUsed]]): SubSource[T, Mat] =
+ new SubSource(delegate.recoverWith {
+ case NonFatal(ex) if predicate.test(ex) => fallback(ex)
+ })
+
/**
* While similar to [[recover]] this operator can be used to transform an error signal to a different one *without* logging
* it as an error in the process. So in that sense it is NOT exactly equivalent to `recover(t => throw t2)` since recover
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org