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:26:17 UTC
(pekko) branch main updated: 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 main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new d23bab48e7 feat: Add missing onErrorResume to SubFlow and SubSource (#2337)
d23bab48e7 is described below
commit d23bab48e792544af28cff0932d94a00739e64b5
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)
---
.../org/apache/pekko/stream/javadsl/Flow.scala | 15 +++--
.../org/apache/pekko/stream/javadsl/Source.scala | 13 ++--
.../org/apache/pekko/stream/javadsl/SubFlow.scala | 76 +++++++++++++++++++++
.../apache/pekko/stream/javadsl/SubSource.scala | 77 ++++++++++++++++++++++
4 files changed, 168 insertions(+), 13 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 b8be99a2b9..b922fad77e 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
@@ -26,6 +26,7 @@ import scala.jdk.DurationConverters._
import scala.jdk.FutureConverters._
import scala.jdk.OptionConverters._
import scala.reflect.ClassTag
+import scala.util.control.NonFatal
import org.apache.pekko
import pekko.Done
@@ -2198,11 +2199,11 @@ final class Flow[In, Out, Mat](delegate: scaladsl.Flow[In, Out, Mat]) extends Gr
* '''Cancels when''' downstream cancels
*
* @param fallback Function which produces a Source to continue the stream
- * @since 2.0.0
+ * @since 1.3.0
*/
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)
})
/**
@@ -2222,13 +2223,13 @@ final class Flow[In, Out, Mat](delegate: scaladsl.Flow[In, Out, Mat]) extends Gr
*
* @param clazz the class object of the failure cause
* @param fallback Function which produces a Source to continue the stream
- * @since 2.0.0
+ * @since 1.3.0
*/
def onErrorResume[T >: Out](
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)
})
/**
@@ -2247,14 +2248,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
- * @since 2.0.0
+ * @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 127146a1a9..e5844cd00a 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
@@ -27,6 +27,7 @@ import scala.jdk.DurationConverters._
import scala.jdk.FutureConverters._
import scala.jdk.OptionConverters._
import scala.reflect.ClassTag
+import scala.util.control.NonFatal
import org.apache.pekko
import pekko.{ Done, NotUsed }
@@ -2432,12 +2433,12 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[
* '''Cancels when''' downstream cancels
*
* @param fallback Function which produces a Source to continue the stream
- * @since 2.0.0
+ * @since 1.3.0
*/
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)
})
/**
@@ -2457,13 +2458,13 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[
*
* @param clazz the class object of the failure cause
* @param fallback Function which produces a Source to continue the stream
- * @since 2.0.0
+ * @since 1.3.0
*/
def onErrorResume[T >: Out](
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)
})
/**
@@ -2483,13 +2484,13 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[
*
* @param predicate Predicate which determines if the exception should be handled
* @param fallback Function which produces a Source to continue the stream
- * @since 2.0.0
+ * @since 1.3.0
*/
def onErrorResume[T >: Out](
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 8da6eb21c0..fe6015ae61 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
@@ -24,6 +24,7 @@ import scala.jdk.DurationConverters._
import scala.jdk.FutureConverters._
import scala.jdk.OptionConverters._
import scala.reflect.ClassTag
+import scala.util.control.NonFatal
import org.apache.pekko
import pekko.NotUsed
@@ -1445,6 +1446,81 @@ final 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 5521a2f157..1cfc839a4b 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
@@ -24,6 +24,7 @@ import scala.jdk.DurationConverters._
import scala.jdk.FutureConverters._
import scala.jdk.OptionConverters._
import scala.reflect.ClassTag
+import scala.util.control.NonFatal
import org.apache.pekko
import pekko.NotUsed
@@ -1422,6 +1423,82 @@ final 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