raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 11

Re-implement javadsl testkit (#22240)

Browse source 
* re-implement javadsl testkit

* fix mima problem

* rebase master

* move ImplicitSender/DefaultTimeout to scaladsl

* undo the change of moving scala api

* fix return type and add doc

* resolve conflicts and add more comments
This commit is contained in:
Hawstein 2017-03-17 03:02:47 +08:00 • committed by Patrik Nordwall
parent 3643f18ded
commit 6434cbe868
93 changed files with 1147 additions and 619 deletions
Download patch file Download diff file Expand all files Collapse all files

62
akka-stream-tests/src/test/java/akka/stream/javadsl/FlowTest.java
View file

@ -17,7 +17,7 @@ import akka.stream.javadsl.GraphDSL.Builder;
import akka.stream.stage.*;
import akka.testkit.AkkaSpec;
import akka.stream.testkit.TestPublisher;
import akka.testkit.JavaTestKit;
import akka.testkit.javadsl.TestKit;
import org.junit.ClassRule;
import org.junit.Ignore;
import org.junit.Test;
@ -51,7 +51,7 @@ public class FlowTest extends StreamTest {
@Test
public void mustBeAbleToUseSimpleOperators() {
final JavaTestKit probe = new JavaTestKit(system);
final TestKit probe = new TestKit(system);
final String[] lookup = { "a", "b", "c", "d", "e", "f" };
final java.lang.Iterable<Integer> input = Arrays.asList(0, 1, 2, 3, 4, 5);
final Source<Integer, NotUsed> ints = Source.from(input);
@ -87,7 +87,7 @@ public class FlowTest extends StreamTest {
@Test
public void mustBeAbleToUseDropWhile() throws Exception {
final JavaTestKit probe = new JavaTestKit(system);
final TestKit probe = new TestKit(system);
final Source<Integer, NotUsed> source = Source.from(Arrays.asList(0, 1, 2, 3));
final Flow<Integer, Integer, NotUsed> flow =
Flow.of(Integer.class).dropWhile(elem -> elem < 2);
@ -102,7 +102,7 @@ public class FlowTest extends StreamTest {
@Test
public void mustBeAbleToUseStatefulMaponcat() throws Exception {
final JavaTestKit probe = new JavaTestKit(system);
final TestKit probe = new TestKit(system);
final java.lang.Iterable<Integer> input = Arrays.asList(1, 2, 3, 4, 5);
final Source<Integer, NotUsed> ints = Source.from(input);
final Flow<Integer, Integer, NotUsed> flow = Flow.of(Integer.class).statefulMapConcat(
@ -124,7 +124,7 @@ public class FlowTest extends StreamTest {
@Test
public void mustBeAbleToUseIntersperse() throws Exception {
final JavaTestKit probe = new JavaTestKit(system);
final TestKit probe = new TestKit(system);
final Source<String, NotUsed> source = Source.from(Arrays.asList("0", "1", "2", "3"));
final Flow<String, String, NotUsed> flow = Flow.of(String.class).intersperse("[", ",", "]");
@ -145,7 +145,7 @@ public class FlowTest extends StreamTest {
@Test
public void mustBeAbleToUseIntersperseAndConcat() throws Exception {
final JavaTestKit probe = new JavaTestKit(system);
final TestKit probe = new TestKit(system);
final Source<String, NotUsed> source = Source.from(Arrays.asList("0", "1", "2", "3"));
final Flow<String, String, NotUsed> flow = Flow.of(String.class).intersperse(",");
@ -165,7 +165,7 @@ public class FlowTest extends StreamTest {
@Test
public void mustBeAbleToUseTakeWhile() throws Exception {
final JavaTestKit probe = new JavaTestKit(system);
final TestKit probe = new TestKit(system);
final Source<Integer, NotUsed> source = Source.from(Arrays.asList(0, 1, 2, 3));
final Flow<Integer, Integer, NotUsed> flow = Flow.of(Integer.class).takeWhile
(new Predicate<Integer>() {
@ -189,7 +189,7 @@ public class FlowTest extends StreamTest {
@Ignore("StatefulStage to be converted to GraphStage when Java Api is available (#18817)") @Test
public void mustBeAbleToUseTransform() {
final JavaTestKit probe = new JavaTestKit(system);
final TestKit probe = new TestKit(system);
final Iterable<Integer> input = Arrays.asList(0, 1, 2, 3, 4, 5, 6, 7);
// duplicate each element, stop after 4 elements, and emit sum to the end
final Flow<Integer, Integer, NotUsed> flow = Flow.of(Integer.class).via(new GraphStage<FlowShape<Integer, Integer>>() {
@ -383,7 +383,7 @@ public class FlowTest extends StreamTest {
@Test
public void mustBeAbleToUseZip() {
final JavaTestKit probe = new JavaTestKit(system);
final TestKit probe = new TestKit(system);
final Iterable<String> input1 = Arrays.asList("A", "B", "C");
final Iterable<Integer> input2 = Arrays.asList(1, 2, 3);
@ -407,7 +407,7 @@ public class FlowTest extends StreamTest {
}
})).run(materializer);
List<Object> output = Arrays.asList(probe.receiveN(3));
List<Object> output = probe.receiveN(3);
List<Pair<String, Integer>> expected = Arrays.asList(new Pair<String, Integer>("A", 1), new Pair<String, Integer>(
"B", 2), new Pair<String, Integer>("C", 3));
assertEquals(expected, output);
@ -415,7 +415,7 @@ public class FlowTest extends StreamTest {
@Test
public void mustBeAbleToUseConcat() {
final JavaTestKit probe = new JavaTestKit(system);
final TestKit probe = new TestKit(system);
final Iterable<String> input1 = Arrays.asList("A", "B", "C");
final Iterable<String> input2 = Arrays.asList("D", "E", "F");
@ -428,13 +428,13 @@ public class FlowTest extends StreamTest {
}
}, materializer);
List<Object> output = Arrays.asList(probe.receiveN(6));
List<Object> output = probe.receiveN(6);
assertEquals(Arrays.asList("A", "B", "C", "D", "E", "F"), output);
}
@Test
public void mustBeAbleToUsePrepend() {
final JavaTestKit probe = new JavaTestKit(system);
final TestKit probe = new TestKit(system);
final Iterable<String> input1 = Arrays.asList("A", "B", "C");
final Iterable<String> input2 = Arrays.asList("D", "E", "F");
@ -447,13 +447,13 @@ public class FlowTest extends StreamTest {
}
}, materializer);
List<Object> output = Arrays.asList(probe.receiveN(6));
List<Object> output = probe.receiveN(6);
assertEquals(Arrays.asList("A", "B", "C", "D", "E", "F"), output);
}
@Test
public void mustBeAbleToUsePrefixAndTail() throws Exception {
final JavaTestKit probe = new JavaTestKit(system);
final TestKit probe = new TestKit(system);
final Iterable<Integer> input = Arrays.asList(1, 2, 3, 4, 5, 6);
final Flow<Integer, Pair<List<Integer>, Source<Integer, NotUsed>>, NotUsed> flow = Flow.of(Integer.class).prefixAndTail(3);
CompletionStage<Pair<List<Integer>, Source<Integer, NotUsed>>> future =
@ -468,7 +468,7 @@ public class FlowTest extends StreamTest {
@Test
public void mustBeAbleToUseConcatAllWithSources() throws Exception {
final JavaTestKit probe = new JavaTestKit(system);
final TestKit probe = new TestKit(system);
final Iterable<Integer> input1 = Arrays.asList(1, 2, 3);
final Iterable<Integer> input2 = Arrays.asList(4, 5);
@ -488,7 +488,7 @@ public class FlowTest extends StreamTest {
@Test
public void mustBeAbleToUseFlatMapMerge() throws Exception {
final JavaTestKit probe = new JavaTestKit(system);
final TestKit probe = new TestKit(system);
final Iterable<Integer> input1 = Arrays.asList(0, 1, 2, 3, 4, 5, 6, 7, 8, 9);
final Iterable<Integer> input2 = Arrays.asList(10, 11, 12, 13, 14, 15, 16, 17, 18, 19);
final Iterable<Integer> input3 = Arrays.asList(20, 21, 22, 23, 24, 25, 26, 27, 28, 29);
@ -520,7 +520,7 @@ public class FlowTest extends StreamTest {
@Test
public void mustBeAbleToUseBuffer() throws Exception {
final JavaTestKit probe = new JavaTestKit(system);
final TestKit probe = new TestKit(system);
final List<String> input = Arrays.asList("A", "B", "C");
final Flow<String, List<String>, NotUsed> flow = Flow.of(String.class).buffer(2, OverflowStrategy.backpressure()).grouped(4);
final CompletionStage<List<String>> future = Source.from(input).via(flow)
@ -542,7 +542,7 @@ public class FlowTest extends StreamTest {
@Test
public void mustBeAbleToUseConflate() throws Exception {
final JavaTestKit probe = new JavaTestKit(system);
final TestKit probe = new TestKit(system);
final List<String> input = Arrays.asList("A", "B", "C");
final Flow<String, String, NotUsed> flow = Flow.of(String.class).conflateWithSeed(new Function<String, String>() {
@Override
@ -568,7 +568,7 @@ public class FlowTest extends StreamTest {
@Test
public void mustBeAbleToUseBatch() throws Exception {
final JavaTestKit probe = new JavaTestKit(system);
final TestKit probe = new TestKit(system);
final List<String> input = Arrays.asList("A", "B", "C");
final Flow<String, String, NotUsed> flow = Flow.of(String.class).batch(3L, new Function<String, String>() {
@Override
@ -588,7 +588,7 @@ public class FlowTest extends StreamTest {
@Test
public void mustBeAbleToUseBatchWeighted() throws Exception {
final JavaTestKit probe = new JavaTestKit(system);
final TestKit probe = new TestKit(system);
final List<String> input = Arrays.asList("A", "B", "C");
final Flow<String, String, NotUsed> flow = Flow.of(String.class).batchWeighted(3L, new Function<String, Object>() {
@Override
@ -613,7 +613,7 @@ public class FlowTest extends StreamTest {
@Test
public void mustBeAbleToUseExpand() throws Exception {
final JavaTestKit probe = new JavaTestKit(system);
final TestKit probe = new TestKit(system);
final List<String> input = Arrays.asList("A", "B", "C");
final Flow<String, String, NotUsed> flow = Flow.of(String.class).expand(in -> Stream.iterate(in, i -> i).iterator());
final Sink<String, CompletionStage<String>> sink = Sink.<String>head();
@ -624,7 +624,7 @@ public class FlowTest extends StreamTest {
@Test
public void mustBeAbleToUseMapAsync() throws Exception {
final JavaTestKit probe = new JavaTestKit(system);
final TestKit probe = new TestKit(system);
final Iterable<String> input = Arrays.asList("a", "b", "c");
final Flow<String, String, NotUsed> flow = Flow.of(String.class).mapAsync(4, elem -> CompletableFuture.completedFuture(elem.toUpperCase()));
Source.from(input).via(flow).runForeach(new Procedure<String>() {
@ -640,7 +640,7 @@ public class FlowTest extends StreamTest {
@Test
public void mustBeAbleToRecover() throws Exception {
final TestPublisher.ManualProbe<Integer> publisherProbe = TestPublisher.manualProbe(true,system);
final JavaTestKit probe = new JavaTestKit(system);
final TestKit probe = new TestKit(system);
final Source<Integer, NotUsed> source = Source.fromPublisher(publisherProbe);
final Flow<Integer, Integer, NotUsed> flow = Flow.of(Integer.class).map(
@ -674,7 +674,7 @@ public class FlowTest extends StreamTest {
@Test
public void mustBeAbleToRecoverWithSource() throws Exception {
final TestPublisher.ManualProbe<Integer> publisherProbe = TestPublisher.manualProbe(true,system);
final JavaTestKit probe = new JavaTestKit(system);
final TestKit probe = new TestKit(system);
final Iterable<Integer> recover = Arrays.asList(55, 0);
final Source<Integer, NotUsed> source = Source.fromPublisher(publisherProbe);
@ -709,7 +709,7 @@ public class FlowTest extends StreamTest {
@Test
public void mustBeAbleToMaterializeIdentityWithJavaFlow() throws Exception {
final JavaTestKit probe = new JavaTestKit(system);
final TestKit probe = new TestKit(system);
final List<String> input = Arrays.asList("A", "B", "C");
Flow<String,String,NotUsed> otherFlow = Flow.of(String.class);
@ -726,7 +726,7 @@ public class FlowTest extends StreamTest {
@Test
public void mustBeAbleToMaterializeIdentityToJavaSink() throws Exception {
final JavaTestKit probe = new JavaTestKit(system);
final TestKit probe = new TestKit(system);
final List<String> input = Arrays.asList("A", "B", "C");
Flow<String,String,NotUsed> otherFlow = Flow.of(String.class);
@ -755,7 +755,7 @@ public class FlowTest extends StreamTest {
}
}));
final JavaTestKit probe = new JavaTestKit(system);
final TestKit probe = new TestKit(system);
Source<String, ActorRef> source = Source.actorRef(1, OverflowStrategy.dropNew());
final ActorRef actor = source.toMat(sink, Keep.<ActorRef, NotUsed>left()).run(materializer);
probe.watch(actor);
@ -765,7 +765,7 @@ public class FlowTest extends StreamTest {
@Test
public void mustBeAbleToUseZipWith() throws Exception {
final JavaTestKit probe = new JavaTestKit(system);
final TestKit probe = new TestKit(system);
final Iterable<String> input1 = Arrays.asList("A", "B", "C");
final Iterable<String> input2 = Arrays.asList("D", "E", "F");
@ -786,7 +786,7 @@ public class FlowTest extends StreamTest {
@Test
public void mustBeAbleToUseZip2() throws Exception {
final JavaTestKit probe = new JavaTestKit(system);
final TestKit probe = new TestKit(system);
final Iterable<String> input1 = Arrays.asList("A", "B", "C");
final Iterable<String> input2 = Arrays.asList("D", "E", "F");
@ -804,7 +804,7 @@ public class FlowTest extends StreamTest {
@Test
public void mustBeAbleToUseMerge2() {
final JavaTestKit probe = new JavaTestKit(system);
final TestKit probe = new TestKit(system);
final Iterable<String> input1 = Arrays.asList("A", "B", "C");
final Iterable<String> input2 = Arrays.asList("D", "E", "F");