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

convert watchTermination to CompletionStage for Java

Browse source
This commit is contained in:
Roland Kuhn 2016-01-25 11:14:31 +01:00
parent 41d8b1fadd
commit d49d2545ec
3 changed files with 7 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -490,11 +490,11 @@ public class FlowTest extends StreamTest {
@Test @Test
public void mustBeAbleToUseWatchTermination() throws Exception { public void mustBeAbleToUseWatchTermination() throws Exception {
final List<String> input = Arrays.asList("A", "B", "C"); final List<String> input = Arrays.asList("A", "B", "C");
Future<Done> future = Source.from(input) CompletionStage<Done> future = Source.from(input)
.watchTermination(Keep.<NotUsed, Future<Done>>right()) .watchTermination(Keep.right())
.to(Sink.ignore()).run(materializer); .to(Sink.ignore()).run(materializer);
assertEquals(Done.getInstance(), Await.result(future, FiniteDuration.create(3, TimeUnit.SECONDS))); assertEquals(Done.getInstance(), future.toCompletableFuture().get(3, TimeUnit.SECONDS));
} }
@Test @Test

4
akka-stream/src/main/scala/akka/stream/javadsl/Flow.scala
View file

@ -1563,8 +1563,8 @@ final class Flow[-In, +Out, +Mat](delegate: scaladsl.Flow[In, Out, Mat]) extends
* from downstream. It fails with the same error when received error message from * from downstream. It fails with the same error when received error message from
* downstream. * downstream.
*/ */
def watchTermination[M]()(matF: function.Function2[Mat, Future[Done], M]): javadsl.Flow[In, Out, M] = def watchTermination[M]()(matF: function.Function2[Mat, CompletionStage[Done], M]): javadsl.Flow[In, Out, M] =
new Flow(delegate.watchTermination()(combinerToScala(matF))) new Flow(delegate.watchTermination()((left, right) => matF(left, right.toJava)))
/** /**
* Delays the initial element by the specified duration. * Delays the initial element by the specified duration.

4
akka-stream/src/main/scala/akka/stream/javadsl/Source.scala
View file

@ -1738,8 +1738,8 @@ final class Source[+Out, +Mat](delegate: scaladsl.Source[Out, Mat]) extends Grap
* from downstream. It fails with the same error when received error message from * from downstream. It fails with the same error when received error message from
* downstream. * downstream.
*/ */
def watchTermination[M]()(matF: function.Function2[Mat, Future[Done], M]): javadsl.Source[Out, M] = def watchTermination[M]()(matF: function.Function2[Mat, CompletionStage[Done], M]): javadsl.Source[Out, M] =
new Source(delegate.watchTermination()(combinerToScala(matF))) new Source(delegate.watchTermination()((left, right) => matF(left, right.toJava)))
/** /**
* Delays the initial element by the specified duration. * Delays the initial element by the specified duration.