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

add zipLatest and zipLatestWith operators and graphstages

Browse source
This commit is contained in:
Jonas Chapuis 2018-08-27 18:33:58 +02:00
parent 25079cb568
commit a098e0b743
17 changed files with 1104 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -1077,6 +1077,38 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[
matF: function.Function2[Mat, M, M2]): javadsl.Source[Out @uncheckedVariance Pair T, M2] =
this.viaMat(Flow.create[Out].zipMat(that, Keep.right[NotUsed, M]), matF)
/**
* Combine the elements of 2 streams into a stream of tuples, picking always the latest element of each.
*
* A `ZipLatest` has a `left` and a `right` input port and one `out` port.
*
* No element is emitted until at least one element from each Source becomes available.
*
* '''Emits when''' all of the inputs have at least an element available, and then each time an element becomes
* * available on either of the inputs
*
* '''Backpressures when''' downstream backpressures
*
* '''Completes when''' any upstream completes
*
* '''Cancels when''' downstream cancels
*/
def zipLatest[T](that: Graph[SourceShape[T], _]): javadsl.Source[Out @uncheckedVariance Pair T, Mat] =
zipLatestMat(that, Keep.left)
/**
* Combine the elements of current [[Source]] and the given one into a stream of tuples, picking always the latest element of each.
*
* It is recommended to use the internally optimized `Keep.left` and `Keep.right` combiners
* where appropriate instead of manually writing functions that pass through one of the values.
*
* @see [[#zipLatest]].
*/
def zipLatestMat[T, M, M2](
that: Graph[SourceShape[T], M],
matF: function.Function2[Mat, M, M2]): javadsl.Source[Out @uncheckedVariance Pair T, M2] =
this.viaMat(Flow.create[Out].zipLatestMat(that, Keep.right[NotUsed, M]), matF)
/**
* Put together the elements of current [[Source]] and the given one
* into a stream of combined elements using a combiner function.
@ -1109,6 +1141,44 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[
matF: function.Function2[Mat, M, M2]): javadsl.Source[Out3, M2] =
new Source(delegate.zipWithMat[Out2, Out3, M, M2](that)(combinerToScala(combine))(combinerToScala(matF)))
/**
* Combine the elements of multiple streams into a stream of combined elements using a combiner function,
* picking always the latest of the elements of each source.
*
* No element is emitted until at least one element from each Source becomes available. Whenever a new
* element appears, the zipping function is invoked with a tuple containing the new element
* and the other last seen elements.
*
* '''Emits when''' all of the inputs have at least an element available, and then each time an element becomes
* available on either of the inputs
*
* '''Backpressures when''' downstream backpressures
*
* '''Completes when''' any of the upstreams completes
*
* '''Cancels when''' downstream cancels
*/
def zipLatestWith[Out2, Out3](
that: Graph[SourceShape[Out2], _],
combine: function.Function2[Out, Out2, Out3]): javadsl.Source[Out3, Mat] =
new Source(delegate.zipLatestWith[Out2, Out3](that)(combinerToScala(combine)))
/**
* Put together the elements of current [[Source]] and the given one
* into a stream of combined elements using a combiner function,
* picking always the latest of the elements of each source.
*
* It is recommended to use the internally optimized `Keep.left` and `Keep.right` combiners
* where appropriate instead of manually writing functions that pass through one of the values.
*
* @see [[#zipLatestWith]].
*/
def zipLatestWithMat[Out2, Out3, M, M2](
that: Graph[SourceShape[Out2], M],
combine: function.Function2[Out, Out2, Out3],
matF: function.Function2[Mat, M, M2]): javadsl.Source[Out3, M2] =
new Source(delegate.zipLatestWithMat[Out2, Out3, M, M2](that)(combinerToScala(combine))(combinerToScala(matF)))
/**
* Combine the elements of current [[Source]] into a stream of tuples consisting
* of all elements paired with their index. Indices start at 0.