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

Merge pull request #19396 from akka/wip-18948-eagerComplete-RK

Browse source 
#18948 add eagerComplete parameter to FlowOps.merge
This commit is contained in:
Roland Kuhn 2016-01-11 15:42:59 +01:00
commit 72eb3f539d
8 changed files with 67 additions and 40 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -1223,7 +1223,22 @@ final class Flow[-In, +Out, +Mat](delegate: scaladsl.Flow[In, Out, Mat]) extends
* '''Cancels when''' downstream cancels
*/
def merge[T >: Out](that: Graph[SourceShape[T], _]): javadsl.Flow[In, T, Mat] =
new Flow(delegate.merge(that))
merge(that, eagerComplete = false)
/**
* Merge the given [[Source]] to this [[Flow]], taking elements as they arrive from input streams,
* picking randomly when several elements ready.
*
* '''Emits when''' one of the inputs has an element available
*
* '''Backpressures when''' downstream backpressures
*
* '''Completes when''' all upstreams complete (eagerComplete=false) or one upstream completes (eagerComplete=true), default value is `false`
*
* '''Cancels when''' downstream cancels
*/
def merge[T >: Out](that: Graph[SourceShape[T], _], eagerComplete: Boolean): javadsl.Flow[In, T, Mat] =
new Flow(delegate.merge(that, eagerComplete))
/**
* Merge the given [[Source]] to this [[Flow]], taking elements as they arrive from input streams,
@ -1233,6 +1248,17 @@ final class Flow[-In, +Out, +Mat](delegate: scaladsl.Flow[In, Out, Mat]) extends
*/
def mergeMat[T >: Out, M, M2](that: Graph[SourceShape[T], M],
matF: function.Function2[Mat, M, M2]): javadsl.Flow[In, T, M2] =
mergeMat(that, matF, eagerComplete = false)
/**
* Merge the given [[Source]] to this [[Flow]], taking elements as they arrive from input streams,
* picking randomly when several elements ready.
*
* @see [[#merge]]
*/
def mergeMat[T >: Out, M, M2](that: Graph[SourceShape[T], M],
matF: function.Function2[Mat, M, M2],
eagerComplete: Boolean): javadsl.Flow[In, T, M2] =
new Flow(delegate.mergeMat(that)(combinerToScala(matF)))
/**

28
akka-stream/src/main/scala/akka/stream/javadsl/Graph.scala
View file

@ -16,7 +16,7 @@ import akka.stream.impl.ConstantFun
*
* '''Backpressures when''' downstream backpressures
*
* '''Completes when''' all upstreams complete (eagerClose=false) or one upstream completes (eagerClose=true)
* '''Completes when''' all upstreams complete (eagerComplete=false) or one upstream completes (eagerComplete=true)
*
* '''Cancels when''' downstream cancels
*/
@ -36,20 +36,20 @@ object Merge {
/**
* Create a new `Merge` stage with the specified output type.
*
* @param eagerClose set to true in order to make this stage eagerly
* @param eagerComplete set to true in order to make this stage eagerly
* finish as soon as one of its inputs completes
*/
def create[T](inputPorts: Int, eagerClose: Boolean): Graph[UniformFanInShape[T, T], Unit] =
scaladsl.Merge(inputPorts, eagerClose = eagerClose)
def create[T](inputPorts: Int, eagerComplete: Boolean): Graph[UniformFanInShape[T, T], Unit] =
scaladsl.Merge(inputPorts, eagerComplete = eagerComplete)
/**
* Create a new `Merge` stage with the specified output type.
*
* @param eagerClose set to true in order to make this stage eagerly
* @param eagerComplete set to true in order to make this stage eagerly
* finish as soon as one of its inputs completes
*/
def create[T](clazz: Class[T], inputPorts: Int, eagerClose: Boolean): Graph[UniformFanInShape[T, T], Unit] =
create(inputPorts, eagerClose)
def create[T](clazz: Class[T], inputPorts: Int, eagerComplete: Boolean): Graph[UniformFanInShape[T, T], Unit] =
create(inputPorts, eagerComplete)
}
/**
@ -61,7 +61,7 @@ object Merge {
*
* '''Backpressures when''' downstream backpressures
*
* '''Completes when''' all upstreams complete (eagerClose=false) or one upstream completes (eagerClose=true)
* '''Completes when''' all upstreams complete (eagerComplete=false) or one upstream completes (eagerComplete=true)
*
* '''Cancels when''' downstream cancels
*/
@ -80,20 +80,20 @@ object MergePreferred {
/**
* Create a new `MergePreferred` stage with the specified output type.
*
* @param eagerClose set to true in order to make this stage eagerly
* @param eagerComplete set to true in order to make this stage eagerly
* finish as soon as one of its inputs completes
*/
def create[T](secondaryPorts: Int, eagerClose: Boolean): Graph[scaladsl.MergePreferred.MergePreferredShape[T], Unit] =
scaladsl.MergePreferred(secondaryPorts, eagerClose = eagerClose)
def create[T](secondaryPorts: Int, eagerComplete: Boolean): Graph[scaladsl.MergePreferred.MergePreferredShape[T], Unit] =
scaladsl.MergePreferred(secondaryPorts, eagerComplete = eagerComplete)
/**
* Create a new `MergePreferred` stage with the specified output type.
*
* @param eagerClose set to true in order to make this stage eagerly
* @param eagerComplete set to true in order to make this stage eagerly
* finish as soon as one of its inputs completes
*/
def create[T](clazz: Class[T], secondaryPorts: Int, eagerClose: Boolean): Graph[scaladsl.MergePreferred.MergePreferredShape[T], Unit] =
create(secondaryPorts, eagerClose)
def create[T](clazz: Class[T], secondaryPorts: Int, eagerComplete: Boolean): Graph[scaladsl.MergePreferred.MergePreferredShape[T], Unit] =
create(secondaryPorts, eagerComplete)
}

14
akka-stream/src/main/scala/akka/stream/scaladsl/Flow.scala
View file

@ -1460,17 +1460,17 @@ trait FlowOps[+Out, +Mat] {
*
* '''Backpressures when''' downstream backpressures
*
* '''Completes when''' all upstreams complete
* '''Completes when''' all upstreams complete (eagerComplete=false) or one upstream completes (eagerComplete=true), default value is `false`
*
* '''Cancels when''' downstream cancels
*/
def merge[U >: Out, M](that: Graph[SourceShape[U], M]): Repr[U] =
via(mergeGraph(that))
def merge[U >: Out, M](that: Graph[SourceShape[U], M], eagerComplete: Boolean = false): Repr[U] =
via(mergeGraph(that, eagerComplete))
protected def mergeGraph[U >: Out, M](that: Graph[SourceShape[U], M]): Graph[FlowShape[Out @uncheckedVariance, U], M] =
protected def mergeGraph[U >: Out, M](that: Graph[SourceShape[U], M], eagerComplete: Boolean): Graph[FlowShape[Out @uncheckedVariance, U], M] =
GraphDSL.create(that) { implicit b ⇒
r ⇒
val merge = b.add(Merge[U](2))
val merge = b.add(Merge[U](2, eagerComplete))
r ~> merge.in(1)
FlowShape(merge.in(0), merge.out)
}
@ -1721,8 +1721,8 @@ trait FlowOpsMat[+Out, +Mat] extends FlowOps[Out, Mat] {
*
* @see [[#merge]].
*/
def mergeMat[U >: Out, Mat2, Mat3](that: Graph[SourceShape[U], Mat2])(matF: (Mat, Mat2) ⇒ Mat3): ReprMat[U, Mat3] =
viaMat(mergeGraph(that))(matF)
def mergeMat[U >: Out, Mat2, Mat3](that: Graph[SourceShape[U], Mat2], eagerComplete: Boolean = false)(matF: (Mat, Mat2) ⇒ Mat3): ReprMat[U, Mat3] =
viaMat(mergeGraph(that, eagerComplete))(matF)
/**
* Interleave is a deterministic merge of the given [[Source]] with elements of this [[Flow]].

21
akka-stream/src/main/scala/akka/stream/scaladsl/Graph.scala
View file

@ -18,9 +18,9 @@ object Merge {
* Create a new `Merge` with the specified number of input ports.
*
* @param inputPorts number of input ports
* @param eagerClose if true, the merge will complete as soon as one of its inputs completes.
* @param eagerComplete if true, the merge will complete as soon as one of its inputs completes.
*/
def apply[T](inputPorts: Int, eagerClose: Boolean = false): Merge[T] = new Merge(inputPorts, eagerClose)
def apply[T](inputPorts: Int, eagerComplete: Boolean = false): Merge[T] = new Merge(inputPorts, eagerComplete)
}
@ -32,12 +32,13 @@ object Merge {
*
* '''Backpressures when''' downstream backpressures
*
* '''Completes when''' all upstreams complete (eagerClose=false) or one upstream completes (eagerClose=true)
* '''Completes when''' all upstreams complete (eagerComplete=false) or one upstream completes (eagerComplete=true), default value is `false`
*
* '''Cancels when''' downstream cancels
*/
final class Merge[T] private (val inputPorts: Int, val eagerClose: Boolean) extends GraphStage[UniformFanInShape[T, T]] {
final class Merge[T] private (val inputPorts: Int, val eagerComplete: Boolean) extends GraphStage[UniformFanInShape[T, T]] {
require(inputPorts > 1, "A Merge must have more than 1 input port")
val in: immutable.IndexedSeq[Inlet[T]] = Vector.tabulate(inputPorts)(i ⇒ Inlet[T]("Merge.in" + i))
val out: Outlet[T] = Outlet[T]("Merge.out")
override def initialAttributes = Attributes.name("Merge")
@ -82,7 +83,7 @@ final class Merge[T] private (val inputPorts: Int, val eagerClose: Boolean) exte
}
override def onUpstreamFinish() =
if (eagerClose) {
if (eagerComplete) {
in.foreach(cancel)
runningUpstreams = 0
if (!pending) completeStage()
@ -118,9 +119,9 @@ object MergePreferred {
* Create a new `MergePreferred` with the specified number of secondary input ports.
*
* @param secondaryPorts number of secondary input ports
* @param eagerClose if true, the merge will complete as soon as one of its inputs completes.
* @param eagerComplete if true, the merge will complete as soon as one of its inputs completes.
*/
def apply[T](secondaryPorts: Int, eagerClose: Boolean = false): MergePreferred[T] = new MergePreferred(secondaryPorts, eagerClose)
def apply[T](secondaryPorts: Int, eagerComplete: Boolean = false): MergePreferred[T] = new MergePreferred(secondaryPorts, eagerComplete)
}
/**
@ -134,13 +135,13 @@ object MergePreferred {
*
* '''Backpressures when''' downstream backpressures
*
* '''Completes when''' all upstreams complete (eagerClose=false) or one upstream completes (eagerClose=true)
* '''Completes when''' all upstreams complete (eagerComplete=false) or one upstream completes (eagerComplete=true), default value is `false`
*
* '''Cancels when''' downstream cancels
*
* A `Broadcast` has one `in` port and 2 or more `out` ports.
*/
final class MergePreferred[T] private (val secondaryPorts: Int, val eagerClose: Boolean) extends GraphStage[MergePreferred.MergePreferredShape[T]] {
final class MergePreferred[T] private (val secondaryPorts: Int, val eagerComplete: Boolean) extends GraphStage[MergePreferred.MergePreferredShape[T]] {
require(secondaryPorts >= 1, "A MergePreferred must have more than 0 secondary input ports")
override def initialAttributes = Attributes.name("MergePreferred")
@ -155,7 +156,7 @@ final class MergePreferred[T] private (val secondaryPorts: Int, val eagerClose:
var openInputs = secondaryPorts + 1
def onComplete(): Unit = {
openInputs -= 1
if (eagerClose || openInputs == 0) completeStage()
if (eagerComplete || openInputs == 0) completeStage()
}
setHandler(out, new OutHandler {