Enormous merge with master which probably led to the indirect unfortunate deaths of several kittens
This commit is contained in:
Viktor Klang
commit
e959493e12
85 changed files with 983 additions and 1333 deletions
|
|
@ -3,6 +3,8 @@ package akka.dispatch;
|
|||
import akka.actor.Timeout;
|
||||
import akka.actor.ActorSystem;
|
||||
|
||||
import akka.japi.*;
|
||||
import akka.util.Duration;
|
||||
import org.junit.AfterClass;
|
||||
import org.junit.BeforeClass;
|
||||
import org.junit.Test;
|
||||
|
|
@ -13,23 +15,19 @@ import java.lang.Iterable;
|
|||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import akka.japi.Function;
|
||||
import akka.japi.Function2;
|
||||
import akka.japi.Procedure;
|
||||
import akka.japi.Option;
|
||||
import akka.testkit.AkkaSpec;
|
||||
|
||||
public class JavaFutureTests {
|
||||
|
||||
private static ActorSystem system;
|
||||
private static FutureFactory ff;
|
||||
private static Timeout t;
|
||||
|
||||
private final Duration timeout = Duration.create(5, TimeUnit.SECONDS);
|
||||
|
||||
@BeforeClass
|
||||
public static void beforeAll() {
|
||||
system = ActorSystem.create("JavaFutureTests", AkkaSpec.testConf());
|
||||
t = system.settings().ActorTimeout();
|
||||
ff = new FutureFactory(system.dispatcher(), t);
|
||||
}
|
||||
|
||||
@AfterClass
|
||||
|
|
@ -41,96 +39,76 @@ public class JavaFutureTests {
|
|||
@Test
|
||||
public void mustBeAbleToMapAFuture() {
|
||||
|
||||
Future<String> f1 = ff.future(new Callable<String>() {
|
||||
Future<String> f1 = Futures.future(new Callable<String>() {
|
||||
public String call() {
|
||||
return "Hello";
|
||||
}
|
||||
});
|
||||
}, system.dispatcher());
|
||||
|
||||
Future<String> f2 = f1.map(new Function<String, String>() {
|
||||
public String apply(String s) {
|
||||
return s + " World";
|
||||
}
|
||||
}, t);
|
||||
});
|
||||
|
||||
assertEquals("Hello World", f2.get());
|
||||
assertEquals("Hello World", Await.result(f2, timeout));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void mustBeAbleToExecuteAnOnResultCallback() throws Throwable {
|
||||
final CountDownLatch latch = new CountDownLatch(1);
|
||||
Promise<String> cf = new akka.dispatch.DefaultPromise<String>(1000, TimeUnit.MILLISECONDS, system
|
||||
.dispatcherFactory().defaultGlobalDispatcher());
|
||||
Promise<String> cf = Futures.promise(system.dispatcher());
|
||||
Future<String> f = cf;
|
||||
f.onResult(new Procedure<String>() {
|
||||
public void apply(String result) {
|
||||
if (result.equals("foo"))
|
||||
latch.countDown();
|
||||
}
|
||||
f.onSuccess(new Procedure<String>() {
|
||||
public void apply(String result) {
|
||||
if (result.equals("foo"))
|
||||
latch.countDown();
|
||||
}
|
||||
});
|
||||
|
||||
cf.completeWithResult("foo");
|
||||
cf.success("foo");
|
||||
assertTrue(latch.await(5000, TimeUnit.MILLISECONDS));
|
||||
assertEquals(f.get(), "foo");
|
||||
assertEquals(Await.result(f, timeout), "foo");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void mustBeAbleToExecuteAnOnExceptionCallback() throws Throwable {
|
||||
final CountDownLatch latch = new CountDownLatch(1);
|
||||
Promise<String> cf = new akka.dispatch.DefaultPromise<String>(1000, TimeUnit.MILLISECONDS, system
|
||||
.dispatcherFactory().defaultGlobalDispatcher());
|
||||
Promise<String> cf = Futures.promise(system.dispatcher());
|
||||
Future<String> f = cf;
|
||||
f.onException(new Procedure<Throwable>() {
|
||||
public void apply(Throwable t) {
|
||||
if (t instanceof NullPointerException)
|
||||
latch.countDown();
|
||||
}
|
||||
f.onFailure(new Procedure<Throwable>() {
|
||||
public void apply(Throwable t) {
|
||||
if (t instanceof NullPointerException)
|
||||
latch.countDown();
|
||||
}
|
||||
});
|
||||
|
||||
Throwable exception = new NullPointerException();
|
||||
cf.completeWithException(exception);
|
||||
cf.failure(exception);
|
||||
assertTrue(latch.await(5000, TimeUnit.MILLISECONDS));
|
||||
assertEquals(f.exception().get(), exception);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void mustBeAbleToExecuteAnOnTimeoutCallback() throws Throwable {
|
||||
final CountDownLatch latch = new CountDownLatch(1);
|
||||
Promise<String> cf = new akka.dispatch.DefaultPromise<String>(1000, TimeUnit.MILLISECONDS, system
|
||||
.dispatcherFactory().defaultGlobalDispatcher());
|
||||
Future<String> f = cf;
|
||||
f.onTimeout(new Procedure<Future<String>>() {
|
||||
public void apply(Future<String> future) {
|
||||
latch.countDown();
|
||||
}
|
||||
});
|
||||
|
||||
assertTrue(latch.await(5000, TimeUnit.MILLISECONDS));
|
||||
assertTrue(f.value().isEmpty());
|
||||
assertEquals(f.value().get().left().get(), exception);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void mustBeAbleToExecuteAnOnCompleteCallback() throws Throwable {
|
||||
final CountDownLatch latch = new CountDownLatch(1);
|
||||
Promise<String> cf = new akka.dispatch.DefaultPromise<String>(1000, TimeUnit.MILLISECONDS, system
|
||||
.dispatcherFactory().defaultGlobalDispatcher());
|
||||
Promise<String> cf = Futures.promise(system.dispatcher());
|
||||
Future<String> f = cf;
|
||||
f.onComplete(new Procedure<Future<String>>() {
|
||||
public void apply(akka.dispatch.Future<String> future) {
|
||||
f.onComplete(new Procedure2<Throwable,String>() {
|
||||
public void apply(Throwable t, String r) {
|
||||
latch.countDown();
|
||||
}
|
||||
});
|
||||
|
||||
cf.completeWithResult("foo");
|
||||
cf.success("foo");
|
||||
assertTrue(latch.await(5000, TimeUnit.MILLISECONDS));
|
||||
assertEquals(f.get(), "foo");
|
||||
assertEquals(Await.result(f, timeout), "foo");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void mustBeAbleToForeachAFuture() throws Throwable {
|
||||
final CountDownLatch latch = new CountDownLatch(1);
|
||||
Promise<String> cf = new akka.dispatch.DefaultPromise<String>(1000, TimeUnit.MILLISECONDS, system
|
||||
.dispatcherFactory().defaultGlobalDispatcher());
|
||||
Promise<String> cf = Futures.promise(system.dispatcher());
|
||||
Future<String> f = cf;
|
||||
f.foreach(new Procedure<String>() {
|
||||
public void apply(String future) {
|
||||
|
|
@ -138,50 +116,47 @@ public class JavaFutureTests {
|
|||
}
|
||||
});
|
||||
|
||||
cf.completeWithResult("foo");
|
||||
cf.success("foo");
|
||||
assertTrue(latch.await(5000, TimeUnit.MILLISECONDS));
|
||||
assertEquals(f.get(), "foo");
|
||||
assertEquals(Await.result(f, timeout), "foo");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void mustBeAbleToFlatMapAFuture() throws Throwable {
|
||||
final CountDownLatch latch = new CountDownLatch(1);
|
||||
Promise<String> cf = new akka.dispatch.DefaultPromise<String>(1000, TimeUnit.MILLISECONDS, system
|
||||
.dispatcherFactory().defaultGlobalDispatcher());
|
||||
cf.completeWithResult("1000");
|
||||
Promise<String> cf = Futures.promise(system.dispatcher());
|
||||
cf.success("1000");
|
||||
Future<String> f = cf;
|
||||
Future<Integer> r = f.flatMap(new Function<String, Future<Integer>>() {
|
||||
public Future<Integer> apply(String r) {
|
||||
latch.countDown();
|
||||
Promise<Integer> cf = new akka.dispatch.DefaultPromise<Integer>(1000, TimeUnit.MILLISECONDS, system
|
||||
.dispatcherFactory().defaultGlobalDispatcher());
|
||||
cf.completeWithResult(Integer.parseInt(r));
|
||||
Promise<Integer> cf = Futures.promise(system.dispatcher());
|
||||
cf.success(Integer.parseInt(r));
|
||||
return cf;
|
||||
}
|
||||
}, t);
|
||||
});
|
||||
|
||||
assertEquals(f.get(), "1000");
|
||||
assertEquals(r.get().intValue(), 1000);
|
||||
assertEquals(Await.result(f, timeout), "1000");
|
||||
assertEquals(Await.result(r, timeout).intValue(), 1000);
|
||||
assertTrue(latch.await(5000, TimeUnit.MILLISECONDS));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void mustBeAbleToFilterAFuture() throws Throwable {
|
||||
final CountDownLatch latch = new CountDownLatch(1);
|
||||
Promise<String> cf = new akka.dispatch.DefaultPromise<String>(1000, TimeUnit.MILLISECONDS, system
|
||||
.dispatcherFactory().defaultGlobalDispatcher());
|
||||
Promise<String> cf = Futures.promise(system.dispatcher());
|
||||
Future<String> f = cf;
|
||||
Future<String> r = f.filter(new Function<String, Boolean>() {
|
||||
public Boolean apply(String r) {
|
||||
latch.countDown();
|
||||
return r.equals("foo");
|
||||
}
|
||||
}, t);
|
||||
});
|
||||
|
||||
cf.completeWithResult("foo");
|
||||
cf.success("foo");
|
||||
assertTrue(latch.await(5000, TimeUnit.MILLISECONDS));
|
||||
assertEquals(f.get(), "foo");
|
||||
assertEquals(r.get(), "foo");
|
||||
assertEquals(Await.result(f, timeout), "foo");
|
||||
assertEquals(Await.result(r, timeout), "foo");
|
||||
}
|
||||
|
||||
// TODO: Improve this test, perhaps with an Actor
|
||||
|
|
@ -192,16 +167,16 @@ public class JavaFutureTests {
|
|||
|
||||
for (int i = 0; i < 10; i++) {
|
||||
listExpected.add("test");
|
||||
listFutures.add(ff.future(new Callable<String>() {
|
||||
listFutures.add(Futures.future(new Callable<String>() {
|
||||
public String call() {
|
||||
return "test";
|
||||
}
|
||||
}));
|
||||
}, system.dispatcher()));
|
||||
}
|
||||
|
||||
Future<Iterable<String>> futureList = ff.sequence(listFutures, t);
|
||||
Future<Iterable<String>> futureList = Futures.sequence(listFutures, system.dispatcher());
|
||||
|
||||
assertEquals(futureList.get(), listExpected);
|
||||
assertEquals(Await.result(futureList, timeout), listExpected);
|
||||
}
|
||||
|
||||
// TODO: Improve this test, perhaps with an Actor
|
||||
|
|
@ -212,20 +187,20 @@ public class JavaFutureTests {
|
|||
|
||||
for (int i = 0; i < 10; i++) {
|
||||
expected.append("test");
|
||||
listFutures.add(ff.future(new Callable<String>() {
|
||||
listFutures.add(Futures.future(new Callable<String>() {
|
||||
public String call() {
|
||||
return "test";
|
||||
}
|
||||
}));
|
||||
}, system.dispatcher()));
|
||||
}
|
||||
|
||||
Future<String> result = ff.fold("", 15000, listFutures, new Function2<String, String, String>() {
|
||||
Future<String> result = Futures.fold("", listFutures, new Function2<String, String, String>() {
|
||||
public String apply(String r, String t) {
|
||||
return r + t;
|
||||
}
|
||||
});
|
||||
}, system.dispatcher());
|
||||
|
||||
assertEquals(result.get(), expected.toString());
|
||||
assertEquals(Await.result(result, timeout), expected.toString());
|
||||
}
|
||||
|
||||
@Test
|
||||
|
|
@ -235,20 +210,20 @@ public class JavaFutureTests {
|
|||
|
||||
for (int i = 0; i < 10; i++) {
|
||||
expected.append("test");
|
||||
listFutures.add(ff.future(new Callable<String>() {
|
||||
listFutures.add(Futures.future(new Callable<String>() {
|
||||
public String call() {
|
||||
return "test";
|
||||
}
|
||||
}));
|
||||
}, system.dispatcher()));
|
||||
}
|
||||
|
||||
Future<String> result = ff.reduce(listFutures, 15000, new Function2<String, String, String>() {
|
||||
Future<String> result = Futures.reduce(listFutures, new Function2<String, String, String>() {
|
||||
public String apply(String r, String t) {
|
||||
return r + t;
|
||||
}
|
||||
});
|
||||
}, system.dispatcher());
|
||||
|
||||
assertEquals(result.get(), expected.toString());
|
||||
assertEquals(Await.result(result, timeout), expected.toString());
|
||||
}
|
||||
|
||||
@Test
|
||||
|
|
@ -261,17 +236,17 @@ public class JavaFutureTests {
|
|||
listStrings.add("test");
|
||||
}
|
||||
|
||||
Future<Iterable<String>> result = ff.traverse(listStrings, t, new Function<String, Future<String>>() {
|
||||
Future<Iterable<String>> result = Futures.traverse(listStrings, new Function<String, Future<String>>() {
|
||||
public Future<String> apply(final String r) {
|
||||
return ff.future(new Callable<String>() {
|
||||
return Futures.future(new Callable<String>() {
|
||||
public String call() {
|
||||
return r.toUpperCase();
|
||||
}
|
||||
});
|
||||
}, system.dispatcher());
|
||||
}
|
||||
});
|
||||
}, system.dispatcher());
|
||||
|
||||
assertEquals(result.get(), expectedStrings);
|
||||
assertEquals(Await.result(result, timeout), expectedStrings);
|
||||
}
|
||||
|
||||
@Test
|
||||
|
|
@ -279,20 +254,28 @@ public class JavaFutureTests {
|
|||
LinkedList<Future<Integer>> listFutures = new LinkedList<Future<Integer>>();
|
||||
for (int i = 0; i < 10; i++) {
|
||||
final Integer fi = i;
|
||||
listFutures.add(ff.future(new Callable<Integer>() {
|
||||
listFutures.add(Futures.future(new Callable<Integer>() {
|
||||
public Integer call() {
|
||||
return fi;
|
||||
}
|
||||
}));
|
||||
}, system.dispatcher()));
|
||||
}
|
||||
final Integer expect = 5;
|
||||
Future<Option<Integer>> f = ff.find(listFutures, new Function<Integer, Boolean>() {
|
||||
Future<Option<Integer>> f = Futures.find(listFutures, new Function<Integer, Boolean>() {
|
||||
public Boolean apply(Integer i) {
|
||||
return i == 5;
|
||||
}
|
||||
}, t);
|
||||
}, system.dispatcher());
|
||||
|
||||
final Integer got = f.get().get();
|
||||
assertEquals(expect, got);
|
||||
assertEquals(expect, Await.result(f, timeout));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void BlockMustBeCallable() {
|
||||
Promise<String> p = Futures.promise(system.dispatcher());
|
||||
Duration d = Duration.create(1, TimeUnit.SECONDS);
|
||||
p.success("foo");
|
||||
Await.ready(p, d);
|
||||
assertEquals(Await.result(p, d), "foo");
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue