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

Merge pull request #16186 from ktoso/fix-remove-dup-fold-ktoso

Browse source 
=str removing duplicate fold sink definition on Sink
This commit is contained in:
Konrad Malawski 2014-10-31 11:54:54 +01:00
commit 292afaa4d0
2 changed files with 11 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

11
akka-stream-tests/src/test/java/akka/stream/javadsl/SinkTest.java
View file

@ -6,6 +6,7 @@ package akka.stream.javadsl;
import java.util.ArrayList;
import java.util.List;
import akka.stream.javadsl.japi.Function2;
import org.junit.ClassRule;
import org.junit.Test;
import org.reactivestreams.Publisher;
@ -44,4 +45,14 @@ public class SinkTest {
assert Await.result(future, Duration.create("1 second")).equals(1);
}
@Test
public void mustBeAbleToUseFold() throws Exception {
KeyedSink<Integer, Future<Integer>> foldSink = Sink.fold(0, new Function2<Integer, Integer, Integer>() {
@Override public Integer apply(Integer arg1, Integer arg2) throws Exception {
return arg1 + arg2;
}
});
Future<Integer> integerFuture = Source.from(new ArrayList<Integer>()).runWith(foldSink, materializer);
}
}

12
akka-stream/src/main/scala/akka/stream/javadsl/Sink.scala
View file

@ -109,18 +109,6 @@ object Sink {
def future[In]: KeyedSink[In, Future[In]] =
new KeyedSink(scaladsl.Sink.future[In])
/**
* A `Sink` that will invoke the given function for every received element, giving it its previous
* output (or the given `zero` value) and the element as input.
* The returned [[scala.concurrent.Future]] will be completed with value of the final
* function evaluation when the input stream ends, or completed with `Failure`
* if there is an error is signaled in the stream.
*/
def fold[U, T](zero: U, f: Function[akka.japi.Pair[U, T], U]): KeyedSink[T, Future[U]] = {
val sSink = scaladsl.Sink.fold[U, T](zero) { case (a, b) ⇒ f.apply(akka.japi.Pair(a, b)) }
new KeyedSink(sSink)
}
}
/**