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 17:19:39 UTC
(pekko) branch 1.3.x-recover created (now ccdf5d8d52)
This is an automated email from the ASF dual-hosted git repository.
hepin pushed a change to branch 1.3.x-recover
in repository https://gitbox.apache.org/repos/asf/pekko.git
at ccdf5d8d52 feat: Add more recover operators for java dsl. (#2336)
This branch includes the following new commits:
new ccdf5d8d52 feat: Add more recover operators for java dsl. (#2336)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org
(pekko) 01/01: feat: Add more recover operators for java dsl. (#2336)
Posted by he...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hepin pushed a commit to branch 1.3.x-recover
in repository https://gitbox.apache.org/repos/asf/pekko.git
commit ccdf5d8d5215564cf97fa94cbd67832c71f5d55f
Author: He-Pin(kerr) <he...@gmail.com>
AuthorDate: Mon Oct 20 01:10:56 2025 +0800
feat: Add more recover operators for java dsl. (#2336)
(cherry picked from commit 569b1263f7012ebcc1c87f226b62cc7e56307abb)
---
.../org/apache/pekko/stream/javadsl/Flow.scala | 73 ++++++++++++++++-
.../org/apache/pekko/stream/javadsl/Source.scala | 75 ++++++++++++++++-
.../org/apache/pekko/stream/javadsl/SubFlow.scala | 95 +++++++++++++++++++++-
.../apache/pekko/stream/javadsl/SubSource.scala | 87 +++++++++++++++++++-
4 files changed, 325 insertions(+), 5 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..c748aee0ee 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
@@ -2022,6 +2022,7 @@ final class Flow[In, Out, Mat](delegate: scaladsl.Flow[In, Out, Mat]) extends Gr
/**
* Recover allows to send last element on failure and gracefully complete the stream
+ *
* 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.
*
@@ -2039,7 +2040,8 @@ final class Flow[In, Out, Mat](delegate: scaladsl.Flow[In, Out, Mat]) extends Gr
new Flow(delegate.recover(pf))
/**
- * Recover allows to send last element on failure and gracefully complete the stream
+ * Recover allows to send last element on failure and gracefully complete the stream.
+ *
* 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.
*
@@ -2058,6 +2060,75 @@ final class Flow[In, Out, Mat](delegate: scaladsl.Flow[In, Out, Mat]) extends Gr
case elem if clazz.isInstance(elem) => supplier.get()
}
+ /**
+ * Recover allows to send last element on failure and gracefully complete the stream.
+ *
+ * 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.
+ *
+ * Throwing an exception inside `recover` _will_ be logged on ERROR level automatically.
+ *
+ * '''Emits when''' element is available from the upstream or upstream is failed and pf returns an element
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' upstream completes or upstream failed with exception pf can handle
+ *
+ * '''Cancels when''' downstream cancels
+ *
+ * @since 1.3.0
+ */
+ def recover(clazz: Class[_ <: Throwable], fallbackValue: Out): javadsl.Flow[In, Out, Mat] =
+ recover {
+ case elem if clazz.isInstance(elem) => fallbackValue
+ }
+
+ /**
+ * Recover allows to send last element on failure and gracefully complete the stream.
+ *
+ * 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.
+ *
+ * Throwing an exception inside `recover` _will_ be logged on ERROR level automatically.
+ *
+ * '''Emits when''' element is available from the upstream or upstream is failed and pf returns an element
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' upstream completes or upstream failed with exception pf can handle
+ *
+ * '''Cancels when''' downstream cancels
+ *
+ * @since 1.3.0
+ */
+ def recover(p: function.Predicate[_ >: Throwable], creator: function.Creator[Out]): javadsl.Flow[In, Out, Mat] =
+ recover {
+ case elem if p.test(elem) => creator.create()
+ }
+
+ /**
+ * Recover allows to send last element on failure and gracefully complete the stream.
+ *
+ * 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.
+ *
+ * Throwing an exception inside `recover` _will_ be logged on ERROR level automatically.
+ *
+ * '''Emits when''' element is available from the upstream or upstream is failed and pf returns an element
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' upstream completes or upstream failed with exception pf can handle
+ *
+ * '''Cancels when''' downstream cancels
+ *
+ * @since 1.3.0
+ */
+ def recover(p: function.Predicate[_ >: Throwable], fallbackValue: Out): javadsl.Flow[In, Out, Mat] =
+ recover {
+ case elem if p.test(elem) => fallbackValue
+ }
+
/**
* 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/Source.scala b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala
index 70eefdf3d0..499bc07d01 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
@@ -2294,7 +2294,8 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[
new Source(delegate.wireTap(f(_)))
/**
- * Recover allows to send last element on failure and gracefully complete the stream
+ * Recover allows to send last element on failure and gracefully complete the stream.
+ *
* 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.
*
@@ -2312,7 +2313,8 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[
new Source(delegate.recover(pf))
/**
- * Recover allows to send last element on failure and gracefully complete the stream
+ * Recover allows to send last element on failure and gracefully complete the stream.
+ *
* 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.
*
@@ -2331,6 +2333,75 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[
case elem if clazz.isInstance(elem) => supplier.get()
}
+ /**
+ * Recover allows to send last element on failure and gracefully complete the stream.
+ *
+ * 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.
+ *
+ * Throwing an exception inside `recover` _will_ be logged on ERROR level automatically.
+ *
+ * '''Emits when''' element is available from the upstream or upstream is failed and pf returns an element
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' upstream completes or upstream failed with exception pf can handle
+ *
+ * '''Cancels when''' downstream cancels
+ *
+ * @since 1.3.0
+ */
+ def recover(clazz: Class[_ <: Throwable], fallbackValue: Out): javadsl.Source[Out, Mat] =
+ recover {
+ case elem if clazz.isInstance(elem) => fallbackValue
+ }
+
+ /**
+ * Recover allows to send last element on failure and gracefully complete the stream.
+ *
+ * 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.
+ *
+ * Throwing an exception inside `recover` _will_ be logged on ERROR level automatically.
+ *
+ * '''Emits when''' element is available from the upstream or upstream is failed and pf returns an element
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' upstream completes or upstream failed with exception pf can handle
+ *
+ * '''Cancels when''' downstream cancels
+ *
+ * @since 1.3.0
+ */
+ def recover(p: function.Predicate[_ >: Throwable], creator: function.Creator[Out]): javadsl.Source[Out, Mat] =
+ recover {
+ case elem if p.test(elem) => creator.create()
+ }
+
+ /**
+ * Recover allows to send last element on failure and gracefully complete the stream.
+ *
+ * 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.
+ *
+ * Throwing an exception inside `recover` _will_ be logged on ERROR level automatically.
+ *
+ * '''Emits when''' element is available from the upstream or upstream is failed and pf returns an element
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' upstream completes or upstream failed with exception pf can handle
+ *
+ * '''Cancels when''' downstream cancels
+ *
+ * @since 1.3.0
+ */
+ def recover(p: function.Predicate[_ >: Throwable], fallbackValue: Out): javadsl.Source[Out, Mat] =
+ recover {
+ case elem if p.test(elem) => fallbackValue
+ }
+
/**
* 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/SubFlow.scala b/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubFlow.scala
index 72e1f77691..bdbb51ff7d 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
@@ -1349,7 +1349,8 @@ class SubFlow[In, Out, Mat](
new SubFlow(delegate.dropWhile(p.test))
/**
- * Recover allows to send last element on failure and gracefully complete the stream
+ * Recover allows to send last element on failure and gracefully complete the stream.
+ *
* 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.
*
@@ -1366,6 +1367,98 @@ class SubFlow[In, Out, Mat](
def recover(pf: PartialFunction[Throwable, Out]): SubFlow[In, Out, Mat] =
new SubFlow(delegate.recover(pf))
+ /**
+ * Recover allows to send last element on failure and gracefully complete the stream.
+ *
+ * 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.
+ *
+ * Throwing an exception inside `recover` _will_ be logged on ERROR level automatically.
+ *
+ * '''Emits when''' element is available from the upstream or upstream is failed and pf returns an element
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' upstream completes or upstream failed with exception pf can handle
+ *
+ * '''Cancels when''' downstream cancels
+ *
+ * @since 1.3.0
+ */
+ def recover(clazz: Class[_ <: Throwable], creator: function.Creator[Out]): SubFlow[In, Out, Mat] =
+ new SubFlow(delegate.recover {
+ case elem if clazz.isInstance(elem) => creator.create()
+ })
+
+ /**
+ * Recover allows to send last element on failure and gracefully complete the stream.
+ *
+ * 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.
+ *
+ * Throwing an exception inside `recover` _will_ be logged on ERROR level automatically.
+ *
+ * '''Emits when''' element is available from the upstream or upstream is failed and pf returns an element
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' upstream completes or upstream failed with exception pf can handle
+ *
+ * '''Cancels when''' downstream cancels
+ *
+ * @since 1.3.0
+ */
+ def recover(clazz: Class[_ <: Throwable], fallbackValue: Out): SubFlow[In, Out, Mat] =
+ new SubFlow(delegate.recover {
+ case elem if clazz.isInstance(elem) => fallbackValue
+ })
+
+ /**
+ * Recover allows to send last element on failure and gracefully complete the stream.
+ *
+ * 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.
+ *
+ * Throwing an exception inside `recover` _will_ be logged on ERROR level automatically.
+ *
+ * '''Emits when''' element is available from the upstream or upstream is failed and pf returns an element
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' upstream completes or upstream failed with exception pf can handle
+ *
+ * '''Cancels when''' downstream cancels
+ *
+ * @since 1.3.0
+ */
+ def recover(p: function.Predicate[_ >: Throwable], creator: function.Creator[Out]): SubFlow[In, Out, Mat] =
+ new SubFlow(delegate.recover {
+ case elem if p.test(elem) => creator.create()
+ })
+
+ /**
+ * Recover allows to send last element on failure and gracefully complete the stream.
+ *
+ * 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.
+ *
+ * Throwing an exception inside `recover` _will_ be logged on ERROR level automatically.
+ *
+ * '''Emits when''' element is available from the upstream or upstream is failed and pf returns an element
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' upstream completes or upstream failed with exception pf can handle
+ *
+ * '''Cancels when''' downstream cancels
+ *
+ * @since 1.3.0
+ */
+ def recover(p: function.Predicate[_ >: Throwable], fallbackValue: Out): SubFlow[In, Out, Mat] =
+ new SubFlow(delegate.recover {
+ case elem if p.test(elem) => fallbackValue
+ })
+
/**
* RecoverWith allows to switch to alternative Source on flow failure. It will stay in effect after
* a failure has been recovered so that each time there is a failure it is fed into the `pf` and a new
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..4b765ee954 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
@@ -1331,7 +1331,8 @@ class SubSource[Out, Mat](
new SubSource(delegate.delayWith(() => DelayStrategy.asScala(delayStrategySupplier.get), overFlowStrategy))
/**
- * Recover allows to send last element on failure and gracefully complete the stream
+ * Recover allows to send last element on failure and gracefully complete the stream.
+ *
* 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.
*
@@ -1346,6 +1347,90 @@ class SubSource[Out, Mat](
def recover(pf: PartialFunction[Throwable, Out]): SubSource[Out, Mat] =
new SubSource(delegate.recover(pf))
+ /**
+ * Recover allows to send last element on failure and gracefully complete the stream.
+ *
+ * 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 pf returns an element
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' upstream completes or upstream failed with exception pf can handle
+ *
+ * '''Cancels when''' downstream cancels
+ *
+ * @since 1.3.0
+ */
+ def recover(clazz: Class[_ <: Throwable], creator: function.Creator[Out]): SubSource[Out, Mat] =
+ new SubSource(delegate.recover {
+ case elem if clazz.isInstance(elem) => creator.create()
+ })
+
+ /**
+ * Recover allows to send last element on failure and gracefully complete the stream.
+ *
+ * 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 pf returns an element
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' upstream completes or upstream failed with exception pf can handle
+ *
+ * '''Cancels when''' downstream cancels
+ *
+ * @since 1.3.0
+ */
+ def recover(clazz: Class[_ <: Throwable], fallbackValue: Out): SubSource[Out, Mat] =
+ new SubSource(delegate.recover {
+ case elem if clazz.isInstance(elem) => fallbackValue
+ })
+
+ /**
+ * Recover allows to send last element on failure and gracefully complete the stream.
+ *
+ * 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 pf returns an element
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' upstream completes or upstream failed with exception pf can handle
+ *
+ * '''Cancels when''' downstream cancels
+ *
+ * @since 1.3.0
+ */
+ def recover(p: function.Predicate[_ >: Throwable], creator: function.Creator[Out]): SubSource[Out, Mat] =
+ new SubSource(delegate.recover {
+ case elem if p.test(elem) => creator.create()
+ })
+
+ /**
+ * Recover allows to send last element on failure and gracefully complete the stream.
+ *
+ * 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 pf returns an element
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' upstream completes or upstream failed with exception pf can handle
+ *
+ * '''Cancels when''' downstream cancels
+ *
+ * @since 1.3.0
+ */
+ def recover(p: function.Predicate[_ >: Throwable], fallbackValue: Out): SubSource[Out, Mat] =
+ new SubSource(delegate.recover {
+ case elem if p.test(elem) => fallbackValue
+ })
+
/**
* RecoverWith allows to switch to alternative Source on flow failure. It will stay in effect after
* a failure has been recovered so that each time there is a failure it is fed into the `pf` and a new
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org