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/08/23 15:13:30 UTC
(pekko) 01/01: chore: Remove deprecated methods in Futures
This is an automated email from the ASF dual-hosted git repository.
hepin pushed a commit to branch removeFutures
in repository https://gitbox.apache.org/repos/asf/pekko.git
commit d53e9bfb4b8f059a7798a570821dce351ee63a83
Author: He-Pin <he...@gmail.com>
AuthorDate: Sat Aug 23 23:13:15 2025 +0800
chore: Remove deprecated methods in Futures
---
.../org/apache/pekko/dispatch/JavaFutureTests.java | 336 ---------------------
.../org/apache/pekko/pattern/PatternsTest.java | 75 +++--
.../remove-deprecated-methods.excludes | 20 ++
.../scala/org/apache/pekko/dispatch/Future.scala | 269 -----------------
.../java/jdocs/actor/FaultHandlingDocSample.java | 20 +-
docs/src/test/java/jdocs/future/FutureDocTest.java | 36 ++-
6 files changed, 84 insertions(+), 672 deletions(-)
diff --git a/actor-tests/src/test/java/org/apache/pekko/dispatch/JavaFutureTests.java b/actor-tests/src/test/java/org/apache/pekko/dispatch/JavaFutureTests.java
index bed0488ff2..0e936f9c5b 100644
--- a/actor-tests/src/test/java/org/apache/pekko/dispatch/JavaFutureTests.java
+++ b/actor-tests/src/test/java/org/apache/pekko/dispatch/JavaFutureTests.java
@@ -13,8 +13,6 @@
package org.apache.pekko.dispatch;
-import org.apache.pekko.japi.function.Function;
-import org.apache.pekko.japi.function.Function2;
import org.apache.pekko.testkit.PekkoJUnitActorSystemResource;
import org.apache.pekko.actor.ActorSystem;
@@ -28,13 +26,7 @@ import scala.concurrent.duration.Duration;
import org.junit.Test;
import static org.junit.Assert.*;
-import java.io.IOException;
-import java.util.Optional;
-import java.util.concurrent.Callable;
-import java.util.LinkedList;
-import java.lang.Iterable;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import static org.apache.pekko.japi.Util.classTag;
@@ -56,297 +48,6 @@ public class JavaFutureTests extends JUnitSuite {
assertEquals(42, cs.get(3, TimeUnit.SECONDS).intValue());
}
- @Test
- public void mustBeAbleToMapAFuture() throws Exception {
-
- Future<String> f1 =
- Futures.future(
- new Callable<String>() {
- public String call() {
- return "Hello";
- }
- },
- system.dispatcher());
-
- Future<String> f2 =
- f1.map(
- new Mapper<String, String>() {
- public String apply(String s) {
- return s + " World";
- }
- },
- system.dispatcher());
-
- assertEquals("Hello World", Await.result(f2, timeout));
- }
-
- @Test
- public void mustBeAbleToExecuteAnOnResultCallback() throws Throwable {
- final CountDownLatch latch = new CountDownLatch(1);
- Promise<String> cf = Futures.promise();
- Future<String> f = cf.future();
- f.onComplete(
- new OnComplete<String>() {
- public void onComplete(Throwable t, String r) {
- if ("foo".equals(r)) latch.countDown();
- }
- },
- system.dispatcher());
-
- cf.success("foo");
- assertTrue(latch.await(5, TimeUnit.SECONDS));
- assertEquals("foo", Await.result(f, timeout));
- }
-
- @Test
- public void mustBeAbleToExecuteAnOnExceptionCallback() throws Throwable {
- final CountDownLatch latch = new CountDownLatch(1);
- Promise<String> cf = Futures.promise();
- Future<String> f = cf.future();
- f.onComplete(
- new OnComplete<String>() {
- public void onComplete(Throwable t, String r) {
- // 'null instanceof ...' is always false
- if (t instanceof NullPointerException) latch.countDown();
- }
- },
- system.dispatcher());
-
- Throwable exception = new NullPointerException();
- cf.failure(exception);
- assertTrue(latch.await(5, TimeUnit.SECONDS));
- assertEquals(exception, f.value().get().failed().get());
- }
-
- @Test
- public void mustBeAbleToExecuteAnOnCompleteCallback() throws Throwable {
- final CountDownLatch latch = new CountDownLatch(1);
- Promise<String> cf = Futures.promise();
- Future<String> f = cf.future();
- f.onComplete(
- new OnComplete<String>() {
- public void onComplete(Throwable t, String r) {
- latch.countDown();
- }
- },
- system.dispatcher());
-
- cf.success("foo");
- assertTrue(latch.await(5, TimeUnit.SECONDS));
- assertEquals("foo", Await.result(f, timeout));
- }
-
- @Test
- public void mustBeAbleToForeachAFuture() throws Throwable {
- final CountDownLatch latch = new CountDownLatch(1);
- Promise<String> cf = Futures.promise();
- Future<String> f = cf.future();
- f.foreach(
- new Foreach<String>() {
- public void each(String future) {
- latch.countDown();
- }
- },
- system.dispatcher());
-
- cf.success("foo");
- assertTrue(latch.await(5, TimeUnit.SECONDS));
- assertEquals("foo", Await.result(f, timeout));
- }
-
- @Test
- public void mustBeAbleToFlatMapAFuture() throws Throwable {
- final CountDownLatch latch = new CountDownLatch(1);
- Promise<String> cf = Futures.promise();
- cf.success("1000");
- Future<String> f = cf.future();
- Future<Integer> r =
- f.flatMap(
- new Mapper<String, Future<Integer>>() {
- public Future<Integer> checkedApply(String r) throws Throwable {
- if (false) throw new IOException("Just here to make sure this compiles.");
- latch.countDown();
- Promise<Integer> cf = Futures.promise();
- cf.success(Integer.parseInt(r));
- return cf.future();
- }
- },
- system.dispatcher());
-
- assertEquals("1000", Await.result(f, timeout));
- assertEquals(1000, Await.result(r, timeout).intValue());
- assertTrue(latch.await(5, TimeUnit.SECONDS));
- }
-
- @Test
- public void mustBeAbleToFilterAFuture() throws Throwable {
- final CountDownLatch latch = new CountDownLatch(1);
- Promise<String> cf = Futures.promise();
- Future<String> f = cf.future();
- Future<String> r =
- f.filter(
- Filter.filterOf(
- new Function<String, Boolean>() {
- public Boolean apply(String r) {
- latch.countDown();
- return r.equals("foo");
- }
- }),
- system.dispatcher());
-
- cf.success("foo");
- assertTrue(latch.await(5, TimeUnit.SECONDS));
- assertEquals("foo", Await.result(f, timeout));
- assertEquals("foo", Await.result(r, timeout));
- }
-
- // TODO: Improve this test, perhaps with an Actor
- @Test
- @SuppressWarnings("deprecation")
- public void mustSequenceAFutureList() throws Exception {
- LinkedList<Future<String>> listFutures = new LinkedList<>();
- LinkedList<String> listExpected = new LinkedList<>();
-
- for (int i = 0; i < 10; i++) {
- listExpected.add("test");
- listFutures.add(
- Futures.future(
- new Callable<String>() {
- public String call() {
- return "test";
- }
- },
- system.dispatcher()));
- }
-
- Future<Iterable<String>> futureList = Futures.sequence(listFutures, system.dispatcher());
-
- assertEquals(listExpected, Await.result(futureList, timeout));
- }
-
- // TODO: Improve this test, perhaps with an Actor
- @Test
- @SuppressWarnings("deprecation")
- public void foldForJavaApiMustWork() throws Exception {
- LinkedList<Future<String>> listFutures = new LinkedList<>();
- StringBuilder expected = new StringBuilder();
-
- for (int i = 0; i < 10; i++) {
- expected.append("test");
- listFutures.add(
- Futures.future(
- new Callable<String>() {
- public String call() {
- return "test";
- }
- },
- system.dispatcher()));
- }
-
- Future<String> result =
- Futures.fold(
- "",
- listFutures,
- new Function2<String, String, String>() {
- public String apply(String r, String t) {
- return r + t;
- }
- },
- system.dispatcher());
-
- assertEquals(expected.toString(), Await.result(result, timeout));
- }
-
- @Test
- @SuppressWarnings("deprecation")
- public void reduceForJavaApiMustWork() throws Exception {
- LinkedList<Future<String>> listFutures = new LinkedList<>();
- StringBuilder expected = new StringBuilder();
-
- for (int i = 0; i < 10; i++) {
- expected.append("test");
- listFutures.add(
- Futures.future(
- new Callable<String>() {
- public String call() {
- return "test";
- }
- },
- system.dispatcher()));
- }
-
- Future<String> result =
- Futures.reduce(
- listFutures,
- new Function2<String, String, String>() {
- public String apply(String r, String t) {
- return r + t;
- }
- },
- system.dispatcher());
-
- assertEquals(expected.toString(), Await.result(result, timeout));
- }
-
- @Test
- @SuppressWarnings("deprecation")
- public void traverseForJavaApiMustWork() throws Exception {
- LinkedList<String> listStrings = new LinkedList<>();
- LinkedList<String> expectedStrings = new LinkedList<>();
-
- for (int i = 0; i < 10; i++) {
- expectedStrings.add("TEST");
- listStrings.add("test");
- }
-
- Future<Iterable<String>> result =
- Futures.traverse(
- listStrings,
- new Function<String, Future<String>>() {
- public Future<String> apply(final String r) {
- return Futures.future(
- new Callable<String>() {
- public String call() {
- return r.toUpperCase();
- }
- },
- system.dispatcher());
- }
- },
- system.dispatcher());
-
- assertEquals(expectedStrings, Await.result(result, timeout));
- }
-
- @Test
- @SuppressWarnings("deprecation")
- public void findForJavaApiMustWork() throws Exception {
- LinkedList<Future<Integer>> listFutures = new LinkedList<>();
- for (int i = 0; i < 10; i++) {
- final Integer fi = i;
- listFutures.add(
- Futures.future(
- new Callable<Integer>() {
- public Integer call() {
- return fi;
- }
- },
- system.dispatcher()));
- }
- final Integer expect = 5;
- Future<Optional<Integer>> f =
- Futures.find(
- listFutures,
- new Function<Integer, Boolean>() {
- public Boolean apply(Integer i) {
- return i == 5;
- }
- },
- system.dispatcher());
-
- assertEquals(expect, Await.result(f, timeout).get());
- }
-
@Test
public void blockMustBeCallable() throws Exception {
Promise<String> p = Futures.promise();
@@ -366,41 +67,4 @@ public class JavaFutureTests extends JUnitSuite {
assertEquals("foo", Await.result(p.future(), d));
}
- @Test
- public void recoverToMustBeCallable() throws Exception {
- final IllegalStateException fail = new IllegalStateException("OHNOES");
- Promise<Object> p = Futures.promise();
- Future<Object> f =
- p.future()
- .recover(
- new Recover<Object>() {
- public Object recover(Throwable t) throws Throwable {
- if (t == fail) return "foo";
- else throw t;
- }
- },
- system.dispatcher());
- Duration d = Duration.create(1, TimeUnit.SECONDS);
- p.failure(fail);
- assertEquals("foo", Await.result(f, d));
- }
-
- @Test
- public void recoverWithToMustBeCallable() throws Exception {
- final IllegalStateException fail = new IllegalStateException("OHNOES");
- Promise<Object> p = Futures.promise();
- Future<Object> f =
- p.future()
- .recoverWith(
- new Recover<Future<Object>>() {
- public Future<Object> recover(Throwable t) throws Throwable {
- if (t == fail) return Futures.successful("foo");
- else throw t;
- }
- },
- system.dispatcher());
- Duration d = Duration.create(1, TimeUnit.SECONDS);
- p.failure(fail);
- assertEquals("foo", Await.result(f, d));
- }
}
diff --git a/actor-tests/src/test/java/org/apache/pekko/pattern/PatternsTest.java b/actor-tests/src/test/java/org/apache/pekko/pattern/PatternsTest.java
index fb3e166f53..d29fbf978d 100644
--- a/actor-tests/src/test/java/org/apache/pekko/pattern/PatternsTest.java
+++ b/actor-tests/src/test/java/org/apache/pekko/pattern/PatternsTest.java
@@ -14,12 +14,13 @@
package org.apache.pekko.pattern;
import org.apache.pekko.actor.*;
+import org.apache.pekko.dispatch.CompletionStages;
import org.apache.pekko.dispatch.Futures;
import org.apache.pekko.testkit.PekkoJUnitActorSystemResource;
import org.apache.pekko.testkit.PekkoSpec;
import org.apache.pekko.testkit.TestProbe;
import org.apache.pekko.util.Timeout;
-import org.junit.Assert;
+
import org.junit.ClassRule;
import org.junit.Test;
import org.scalatestplus.junit.JUnitSuite;
@@ -321,33 +322,39 @@ public class PatternsTest extends JUnitSuite {
@Test(expected = IllegalStateException.class)
@SuppressWarnings("deprecation")
public void testAfterFailedCallable() throws Exception {
- Callable<Future<String>> failedCallable =
- () -> Futures.failed(new IllegalStateException("Illegal!"));
+ Callable<CompletionStage<String>> failedCallable =
+ () -> Futures.failedCompletionStage(new IllegalStateException("Illegal!"));
- Future<String> delayedFuture =
- Patterns.after(
- scala.concurrent.duration.Duration.create(200, "millis"),
- system.scheduler(),
- ec,
- failedCallable);
+ CompletionStage<String> delayedFuture =
+ Patterns.after(Duration.ofMillis(200), system.scheduler(), ec, failedCallable);
- Future<String> resultFuture = Futures.firstCompletedOf(Arrays.asList(delayedFuture), ec);
- Await.result(resultFuture, scala.concurrent.duration.FiniteDuration.apply(3, SECONDS));
+ CompletionStage<String> resultFuture =
+ CompletionStages.firstCompletedOf(Arrays.asList(delayedFuture));
+ try {
+ resultFuture.toCompletableFuture().get(3, SECONDS);
+ } catch (ExecutionException e) {
+ throw (IllegalStateException) e.getCause();
+ }
}
@Test(expected = IllegalStateException.class)
@SuppressWarnings("deprecation")
public void testAfterFailedFuture() throws Exception {
- Future<String> delayedFuture =
+ CompletionStage<String> delayedFuture =
Patterns.after(
- scala.concurrent.duration.Duration.create(200, "millis"),
+ Duration.ofMillis(200),
system.scheduler(),
ec,
- () -> Futures.failed(new IllegalStateException("Illegal!")));
+ () -> Futures.failedCompletionStage(new IllegalStateException("Illegal!")));
- Future<String> resultFuture = Futures.firstCompletedOf(Arrays.asList(delayedFuture), ec);
- Await.result(resultFuture, FiniteDuration.apply(3, SECONDS));
+ CompletionStage<String> resultFuture =
+ CompletionStages.firstCompletedOf(Arrays.asList(delayedFuture));
+ try {
+ resultFuture.toCompletableFuture().get(3, SECONDS);
+ } catch (ExecutionException e) {
+ throw (IllegalStateException) e.getCause();
+ }
}
@Test
@@ -355,15 +362,16 @@ public class PatternsTest extends JUnitSuite {
public void testAfterSuccessfulCallable() throws Exception {
final String expected = "Hello";
- Future<String> delayedFuture =
+ CompletionStage<String> delayedFuture =
Patterns.after(
- scala.concurrent.duration.Duration.create(200, "millis"),
+ Duration.ofMillis(200),
system.scheduler(),
ec,
- () -> Futures.successful(expected));
+ () -> CompletableFuture.completedFuture(expected));
- Future<String> resultFuture = Futures.firstCompletedOf(Arrays.asList(delayedFuture), ec);
- final String actual = Await.result(resultFuture, FiniteDuration.apply(3, SECONDS));
+ CompletionStage<String> resultFuture =
+ CompletionStages.firstCompletedOf(Arrays.asList(delayedFuture));
+ final String actual = resultFuture.toCompletableFuture().get(3, SECONDS);
assertEquals(expected, actual);
}
@@ -373,16 +381,17 @@ public class PatternsTest extends JUnitSuite {
public void testAfterSuccessfulFuture() throws Exception {
final String expected = "Hello";
- Future<String> delayedFuture =
+ CompletionStage<String> delayedFuture =
Patterns.after(
- scala.concurrent.duration.Duration.create(200, "millis"),
+ Duration.ofMillis(200),
system.scheduler(),
ec,
- () -> Futures.successful(expected));
+ () -> CompletableFuture.completedFuture(expected));
- Future<String> resultFuture = Futures.firstCompletedOf(Arrays.asList(delayedFuture), ec);
+ CompletionStage<String> resultFuture =
+ CompletionStages.firstCompletedOf(Arrays.asList(delayedFuture));
- final String actual = Await.result(resultFuture, FiniteDuration.apply(3, SECONDS));
+ final String actual = resultFuture.toCompletableFuture().get(3, SECONDS);
assertEquals(expected, actual);
}
@@ -391,19 +400,19 @@ public class PatternsTest extends JUnitSuite {
public void testAfterFiniteDuration() throws Exception {
final String expected = "Hello";
- Future<String> delayedFuture =
+ CompletionStage<String> delayedFuture =
Patterns.after(
- scala.concurrent.duration.Duration.create(200, "millis"),
+ Duration.ofMillis(200),
system.scheduler(),
ec,
- () -> Futures.successful("world"));
+ () -> CompletableFuture.completedFuture("world"));
- Future<String> immediateFuture = Futures.future(() -> expected, ec);
+ CompletionStage<String> immediateFuture = CompletableFuture.completedFuture(expected);
- Future<String> resultFuture =
- Futures.firstCompletedOf(Arrays.asList(delayedFuture, immediateFuture), ec);
+ CompletionStage<String> resultFuture =
+ CompletionStages.firstCompletedOf(Arrays.asList(delayedFuture, immediateFuture));
- final String actual = Await.result(resultFuture, FiniteDuration.apply(3, SECONDS));
+ final String actual = resultFuture.toCompletableFuture().get(3, SECONDS);
assertEquals(expected, actual);
}
diff --git a/actor/src/main/mima-filters/2.0.x.backwards.excludes/remove-deprecated-methods.excludes b/actor/src/main/mima-filters/2.0.x.backwards.excludes/remove-deprecated-methods.excludes
index a82b50ce13..bb2e4eb40d 100644
--- a/actor/src/main/mima-filters/2.0.x.backwards.excludes/remove-deprecated-methods.excludes
+++ b/actor/src/main/mima-filters/2.0.x.backwards.excludes/remove-deprecated-methods.excludes
@@ -79,3 +79,23 @@ ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.pattern.RestartImp
ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.pattern.StopImpliesFailure")
ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.pattern.StopImpliesFailure$")
ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.serialization.Serialization.deserialize")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.dispatch.Futures.traverse")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.dispatch.Futures.sequence")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.dispatch.Futures.reduce")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.dispatch.Futures.fold")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.dispatch.Futures.firstCompletedOf")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.dispatch.Futures.find")
+ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.dispatch.Filter")
+ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.dispatch.Filter$")
+ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.dispatch.Foreach")
+ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.dispatch.Mapper")
+ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.dispatch.OnComplete")
+ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.dispatch.OnFailure")
+ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.dispatch.OnSuccess")
+ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.dispatch.Recover")
+ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.dispatch.japi")
+ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.dispatch.japi$")
+ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.dispatch.japi$BooleanFunctionBridge")
+ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.dispatch.japi$CallbackBridge")
+ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.dispatch.japi$RecoverBridge")
+ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.dispatch.japi$UnitFunctionBridge")
diff --git a/actor/src/main/scala/org/apache/pekko/dispatch/Future.scala b/actor/src/main/scala/org/apache/pekko/dispatch/Future.scala
index bcbf1bbb45..833b2d5bd6 100644
--- a/actor/src/main/scala/org/apache/pekko/dispatch/Future.scala
+++ b/actor/src/main/scala/org/apache/pekko/dispatch/Future.scala
@@ -13,22 +13,16 @@
package org.apache.pekko.dispatch
-import java.lang.{ Iterable => JIterable }
-import java.util.{ LinkedList => JLinkedList, Optional }
import java.util.concurrent.{ Callable, Executor, ExecutorService }
import java.util.concurrent.CompletableFuture
import java.util.concurrent.CompletionStage
import scala.concurrent.{ ExecutionContext, ExecutionContextExecutor, ExecutionContextExecutorService, Future, Promise }
-import scala.runtime.{ AbstractPartialFunction, BoxedUnit }
-import scala.util.{ Failure, Success, Try }
import org.apache.pekko
import pekko.annotation.InternalStableApi
-import pekko.compat
import pekko.dispatch.internal.SameThreadExecutionContext
import pekko.japi.function.Procedure
-import pekko.util.unused
/**
* ExecutionContexts is the Java API for ExecutionContexts
@@ -104,7 +98,6 @@ object ExecutionContexts {
* Futures is the Java API for Futures and Promises
*/
object Futures {
- import org.apache.pekko.util.ccompat.JavaConverters._
/**
* Convert a Scala Future to a Java CompletionStage.
@@ -153,266 +146,4 @@ object Futures {
f.obtrudeException(ex)
f
}
-
- /**
- * Returns a Future that will hold the optional result of the first Future with a result that matches the predicate
- */
- @deprecated("Use `CompletionStages#find` instead.", since = "Pekko 1.2.0")
- def find[T <: AnyRef](
- futures: JIterable[Future[T]],
- predicate: pekko.japi.function.Function[T, java.lang.Boolean],
- executor: ExecutionContext): Future[Optional[T]] = {
- import pekko.util.OptionConverters._
- import pekko.dispatch.ExecutionContexts.parasitic
- compat.Future.find[T](futures.asScala)(predicate.apply(_))(executor)
- .map(_.toJava)(parasitic)
- }
-
- /**
- * Returns a Future to the result of the first future in the list that is completed
- */
- @deprecated("Use `CompletionStages#firstCompletedOf` instead.", since = "Pekko 1.2.0")
- def firstCompletedOf[T <: AnyRef](futures: JIterable[Future[T]], executor: ExecutionContext): Future[T] =
- Future.firstCompletedOf(futures.asScala)(executor)
-
- /**
- * A non-blocking fold over the specified futures, with the start value of the given zero.
- * The fold is performed on the thread where the last future is completed,
- * the result will be the first failure of any of the futures, or any failure in the actual fold,
- * or the result of the fold.
- */
- @deprecated("Use `CompletionStages#fold` instead.", since = "Pekko 1.2.0")
- def fold[T <: AnyRef, R <: AnyRef](
- zero: R,
- futures: JIterable[Future[T]],
- fun: pekko.japi.function.Function2[R, T, R],
- executor: ExecutionContext): Future[R] =
- compat.Future.fold(futures.asScala)(zero)(fun.apply)(executor)
-
- /**
- * Reduces the results of the supplied futures and binary function.
- */
- @deprecated("Use `CompletionStages#reduce` instead.", since = "Pekko 1.2.0")
- def reduce[T <: AnyRef, R >: T](
- futures: JIterable[Future[T]],
- fun: pekko.japi.function.Function2[R, T, R],
- executor: ExecutionContext): Future[R] =
- compat.Future.reduce[T, R](futures.asScala)(fun.apply)(executor)
-
- /**
- * Simple version of [[#traverse]]. Transforms a JIterable[Future[A]] into a Future[JIterable[A]].
- * Useful for reducing many Futures into a single Future.
- */
- @deprecated("Use `CompletionStages#sequence` instead.", since = "Pekko 1.2.0")
- def sequence[A](in: JIterable[Future[A]], executor: ExecutionContext): Future[JIterable[A]] = {
- implicit val d = executor
- in.asScala.foldLeft(Future(new JLinkedList[A]())) { (fr, fa) =>
- for (r <- fr; a <- fa) yield { r.add(a); r }
- }
- }
-
- /**
- * Transforms a JIterable[A] into a Future[JIterable[B]] using the provided Function A => Future[B].
- * This is useful for performing a parallel map. For example, to apply a function to all items of a list
- * in parallel.
- */
- @deprecated("Use `CompletionStages#traverse` instead.", since = "Pekko 1.2.0")
- def traverse[A, B](in: JIterable[A], fn: pekko.japi.function.Function[A, Future[B]], executor: ExecutionContext)
- : Future[JIterable[B]] = {
- implicit val d = executor
- in.asScala.foldLeft(Future(new JLinkedList[B]())) { (fr, a) =>
- val fb = fn(a)
- for (r <- fr; b <- fb) yield { r.add(b); r }
- }
- }
-}
-
-/**
- * This class contains bridge classes between Scala and Java.
- * Internal use only.
- */
-private[dispatch] object japi {
- class CallbackBridge[-T] extends AbstractPartialFunction[T, BoxedUnit] {
- override final def isDefinedAt(t: T): Boolean = true
- override final def apply(t: T): BoxedUnit = {
- internal(t)
- BoxedUnit.UNIT
- }
- protected def internal(@unused result: T): Unit = ()
- }
-
- class RecoverBridge[+T] extends AbstractPartialFunction[Throwable, T] {
- override final def isDefinedAt(t: Throwable): Boolean = true
- override final def apply(t: Throwable): T = internal(t)
- protected def internal(@unused result: Throwable): T = null.asInstanceOf[T]
- }
-
- class BooleanFunctionBridge[-T] extends scala.Function1[T, Boolean] {
- override final def apply(t: T): Boolean = internal(t)
- protected def internal(@unused result: T): Boolean = false
- }
-
- class UnitFunctionBridge[-T] extends (T => BoxedUnit) {
- final def apply$mcLJ$sp(l: Long): BoxedUnit = { internal(l.asInstanceOf[T]); BoxedUnit.UNIT }
- final def apply$mcLI$sp(i: Int): BoxedUnit = { internal(i.asInstanceOf[T]); BoxedUnit.UNIT }
- final def apply$mcLF$sp(f: Float): BoxedUnit = { internal(f.asInstanceOf[T]); BoxedUnit.UNIT }
- final def apply$mcLD$sp(d: Double): BoxedUnit = { internal(d.asInstanceOf[T]); BoxedUnit.UNIT }
- override final def apply(t: T): BoxedUnit = { internal(t); BoxedUnit.UNIT }
- protected def internal(@unused result: T): Unit = ()
- }
-}
-
-/**
- * Callback for when a Future is completed successfully
- * SAM (Single Abstract Method) class
- *
- * Java API
- */
-abstract class OnSuccess[-T] extends japi.CallbackBridge[T] {
- protected final override def internal(result: T) = onSuccess(result)
-
- /**
- * This method will be invoked once when/if a Future that this callback is registered on
- * becomes successfully completed
- */
- @throws(classOf[Throwable])
- def onSuccess(result: T): Unit
-}
-
-/**
- * Callback for when a Future is completed with a failure
- * SAM (Single Abstract Method) class
- *
- * Java API
- */
-abstract class OnFailure extends japi.CallbackBridge[Throwable] {
- protected final override def internal(failure: Throwable) = onFailure(failure)
-
- /**
- * This method will be invoked once when/if a Future that this callback is registered on
- * becomes completed with a failure
- */
- @throws(classOf[Throwable])
- def onFailure(failure: Throwable): Unit
-}
-
-/**
- * Callback for when a Future is completed with either failure or a success
- * SAM (Single Abstract Method) class
- *
- * Java API
- */
-abstract class OnComplete[-T] extends japi.CallbackBridge[Try[T]] {
- protected final override def internal(value: Try[T]): Unit = value match {
- case Failure(t) => onComplete(t, null.asInstanceOf[T])
- case Success(r) => onComplete(null, r)
- }
-
- /**
- * This method will be invoked once when/if a Future that this callback is registered on
- * becomes completed with a failure or a success.
- * In the case of success then "failure" will be null, and in the case of failure the "success" will be null.
- */
- @throws(classOf[Throwable])
- def onComplete(failure: Throwable, success: T): Unit
-}
-
-/**
- * Callback for the Future.recover operation that conditionally turns failures into successes.
- *
- * SAM (Single Abstract Method) class
- *
- * Java API
- */
-abstract class Recover[+T] extends japi.RecoverBridge[T] {
- protected final override def internal(result: Throwable): T = recover(result)
-
- /**
- * This method will be invoked once when/if the Future this recover callback is registered on
- * becomes completed with a failure.
- *
- * @return a successful value for the passed in failure
- * Throws the passed in failure to propagate it.
- *
- * Java API
- */
- @throws(classOf[Throwable])
- def recover(failure: Throwable): T
-}
-
-/**
- * <i><b>Java API (not recommended):</b></i>
- * Callback for the Future.filter operation that creates a new Future which will
- * conditionally contain the success of another Future.
- *
- * Unfortunately it is not possible to express the type of a Scala filter in
- * Java: Function1[T, Boolean], where “Boolean” is the primitive type. It is
- * possible to use `Future.filter` by constructing such a function indirectly:
- *
- * {{{
- * import static org.apache.pekko.dispatch.Filter.filterOf;
- * Future<String> f = ...;
- * f.filter(filterOf(new Function<String, Boolean>() {
- * @Override
- * public Boolean apply(String s) {
- * ...
- * }
- * }));
- * }}}
- *
- * However, `Future.filter` exists mainly to support Scala’s for-comprehensions,
- * thus Java users should prefer `Future.map`, translating non-matching values
- * to failure cases.
- */
-object Filter {
- def filterOf[T](f: pekko.japi.function.Function[T, java.lang.Boolean]): (T => Boolean) =
- new Function1[T, Boolean] { def apply(result: T): Boolean = f(result).booleanValue() }
-}
-
-/**
- * Callback for the Future.foreach operation that will be invoked if the Future that this callback
- * is registered on becomes completed with a success. This method is essentially the same operation
- * as onSuccess.
- *
- * SAM (Single Abstract Method) class
- * Java API
- */
-abstract class Foreach[-T] extends japi.UnitFunctionBridge[T] {
- override final def internal(t: T): Unit = each(t)
-
- /**
- * This method will be invoked once when/if a Future that this callback is registered on
- * becomes successfully completed
- */
- @throws(classOf[Throwable])
- def each(result: T): Unit
-}
-
-/**
- * Callback for the Future.map and Future.flatMap operations that will be invoked
- * if the Future that this callback is registered on becomes completed with a success.
- * This callback is the equivalent of an org.apache.pekko.japi.Function
- *
- * Override "apply" normally, or "checkedApply" if you need to throw checked exceptions.
- *
- * SAM (Single Abstract Method) class
- *
- * Java API
- */
-abstract class Mapper[-T, +R] extends scala.runtime.AbstractFunction1[T, R] {
-
- /**
- * Override this method to perform the map operation, by default delegates to "checkedApply"
- * which by default throws an UnsupportedOperationException.
- */
- def apply(parameter: T): R = checkedApply(parameter)
-
- /**
- * Override this method if you need to throw checked exceptions
- *
- * Throws UnsupportedOperation by default.
- */
- @throws(classOf[Throwable])
- def checkedApply(@unused parameter: T): R =
- throw new UnsupportedOperationException("Mapper.checkedApply has not been implemented")
}
diff --git a/docs/src/test/java/jdocs/actor/FaultHandlingDocSample.java b/docs/src/test/java/jdocs/actor/FaultHandlingDocSample.java
index cb00bc7a2b..1e74d7baaf 100644
--- a/docs/src/test/java/jdocs/actor/FaultHandlingDocSample.java
+++ b/docs/src/test/java/jdocs/actor/FaultHandlingDocSample.java
@@ -22,15 +22,12 @@ import java.util.Map;
import java.time.Duration;
import org.apache.pekko.actor.*;
-import org.apache.pekko.dispatch.Mapper;
import org.apache.pekko.event.LoggingReceive;
import org.apache.pekko.japi.pf.DeciderBuilder;
import org.apache.pekko.pattern.Patterns;
-import org.apache.pekko.util.Timeout;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
-import static org.apache.pekko.japi.Util.classTag;
import static org.apache.pekko.actor.SupervisorStrategy.restart;
import static org.apache.pekko.actor.SupervisorStrategy.stop;
import static org.apache.pekko.actor.SupervisorStrategy.escalate;
@@ -127,7 +124,7 @@ public class FaultHandlingDocSample {
* sender of the Start message of current Progress. The Worker supervise the CounterService.
*/
public static class Worker extends AbstractLoggingActor {
- final Timeout askTimeout = Timeout.create(Duration.ofSeconds(5));
+ final Duration askTimeout = Duration.ofSeconds(5);
// The sender of the initial Start message will continuously be notified
// about progress
@@ -175,17 +172,10 @@ public class FaultHandlingDocSample {
counterService.tell(new Increment(1), getSelf());
counterService.tell(new Increment(1), getSelf());
// Send current progress to the initial sender
- pipe(
- Patterns.ask(counterService, GetCurrentCount, askTimeout)
- .mapTo(classTag(CurrentCount.class))
- .map(
- new Mapper<CurrentCount, Progress>() {
- public Progress apply(CurrentCount c) {
- return new Progress(100.0 * c.count / totalCount);
- }
- },
- getContext().dispatcher()),
- getContext().dispatcher())
+ pipe(Patterns.ask(counterService, GetCurrentCount, askTimeout)
+ .thenApply(CurrentCount.class::cast)
+ .thenApply(c -> new Progress(100.0 * c.count / totalCount)),
+ getContext().dispatcher())
.to(progressListener);
})
.build(),
diff --git a/docs/src/test/java/jdocs/future/FutureDocTest.java b/docs/src/test/java/jdocs/future/FutureDocTest.java
index eae68153b9..82b80eefe3 100644
--- a/docs/src/test/java/jdocs/future/FutureDocTest.java
+++ b/docs/src/test/java/jdocs/future/FutureDocTest.java
@@ -14,18 +14,13 @@
package jdocs.future;
import org.apache.pekko.actor.typed.ActorSystem;
-import org.apache.pekko.dispatch.Futures;
+import org.apache.pekko.dispatch.CompletionStages;
import org.apache.pekko.pattern.Patterns;
import org.apache.pekko.testkit.PekkoJUnitActorSystemResource;
import org.apache.pekko.testkit.PekkoSpec;
-import org.apache.pekko.util.Timeout;
-import org.apache.pekko.util.FutureConverters;
import jdocs.AbstractJavaTest;
import org.junit.ClassRule;
import org.junit.Test;
-import scala.concurrent.Await;
-import scala.concurrent.ExecutionContext;
-import scala.concurrent.Future;
import java.time.Duration;
import java.util.Arrays;
@@ -34,7 +29,6 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import static org.apache.pekko.actor.typed.javadsl.Adapter.toTyped;
-import static org.apache.pekko.dispatch.Futures.future;
// #imports
// #imports
@@ -48,10 +42,9 @@ public class FutureDocTest extends AbstractJavaTest {
private final ActorSystem<Void> system = toTyped(actorSystemResource.getSystem());
- @Test(expected = java.util.concurrent.CompletionException.class)
+ @Test(expected = IllegalStateException.class)
@SuppressWarnings("deprecation")
public void useAfter() throws Exception {
- final ExecutionContext ec = system.executionContext();
// #after
CompletionStage<String> failWithException =
CompletableFuture.supplyAsync(
@@ -61,18 +54,23 @@ public class FutureDocTest extends AbstractJavaTest {
CompletionStage<String> delayed =
Patterns.after(Duration.ofMillis(200), system, () -> failWithException);
// #after
- Future<String> future =
- future(
+ CompletionStage<String> future =
+ CompletableFuture.supplyAsync(
() -> {
- Thread.sleep(1000);
+ try {
+ Thread.sleep(1000);
+ } catch (Throwable ex) {
+ // ignore
+ }
return "foo";
- },
- ec);
- Future<String> result =
- Futures.firstCompletedOf(
- Arrays.<Future<String>>asList(future, FutureConverters.asScala(delayed)), ec);
- Timeout timeout = Timeout.create(Duration.ofSeconds(2));
- Await.result(result, timeout.duration());
+ });
+ CompletionStage<String> result =
+ CompletionStages.firstCompletedOf(Arrays.asList(future, delayed));
+ try {
+ result.toCompletableFuture().get(2, SECONDS);
+ } catch (Throwable ex) {
+ throw (Exception) ex.getCause();
+ }
}
@Test
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org