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

+str #16610 minor fixes

Browse source
This commit is contained in:
Gilad Hoch 2016-01-20 19:29:50 +02:00
parent fde0c11fff
commit 7e488001cc
3 changed files with 9 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

4
akka-stream/src/main/scala/akka/stream/impl/ConstantFun.scala
View file

@ -21,7 +21,7 @@ private[akka] object ConstantFun {
def scalaIdentityFunction[T]: T ⇒ T = conforms
def returnZero[T](t: T): Long = 0L
val zeroLong = (_: Any) ⇒ 0L
def returnOne[T](t: T): Long = 1L
val oneLong = (_: Any) ⇒ 1L
}

3
akka-stream/src/main/scala/akka/stream/impl/fusing/Ops.scala
View file

@ -478,6 +478,9 @@ private[akka] final case class Conflate[In, Out](seed: In ⇒ Out, aggregate: (O
override def restart(): Conflate[In, Out] = copy()
}
/**
* INTERNAL API
*/
private[akka] final case class Batch[In, Out](max: Long, costFn: In ⇒ Long, seed: In ⇒ Out, aggregate: (Out, In) ⇒ Out)
extends GraphStage[FlowShape[In, Out]] {

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

@ -906,8 +906,9 @@ trait FlowOps[+Out, +Mat] {
*
* See also [[FlowOps.limit]], [[FlowOps.limitWeighted]] [[FlowOps.batch]] [[FlowOps.batchWeighted]]
*/
def conflate[S](seed: Out ⇒ S)(aggregate: (S, Out) ⇒ S): Repr[S] = //andThen(Conflate(seed, aggregate))
via(Batch(1L, ConstantFun.returnZero[Out], seed, aggregate).withAttributes(DefaultAttributes.conflate))
def conflate[S](seed: Out ⇒ S)(aggregate: (S, Out) ⇒ S): Repr[S] = andThen(Conflate(seed, aggregate))
//FIXME: conflate can be expressed as a batch
//via(Batch(1L, ConstantFun.zeroLong, seed, aggregate).withAttributes(DefaultAttributes.conflate))
/**
* Allows a faster upstream to progress independently of a slower subscriber by aggregating elements into batches
@ -932,7 +933,7 @@ trait FlowOps[+Out, +Mat] {
* @param aggregate Takes the currently batched value and the current pending element to produce a new aggregate
*/
def batch[S](max: Long, seed: Out ⇒ S)(aggregate: (S, Out) ⇒ S): Repr[S] =
via(Batch(max, ConstantFun.returnOne[Out], seed, aggregate).withAttributes(DefaultAttributes.batch))
via(Batch(max, ConstantFun.oneLong, seed, aggregate).withAttributes(DefaultAttributes.batch))
/**
* Allows a faster upstream to progress independently of a slower subscriber by aggregating elements into batches