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

=str #23398 groupedWeightedWithin zero cost bugfix (#23399)

Browse source 
* added a failing test to demonstrate a bug in groupedWeightedWithin with zero cost function

* fixed the bug by adding a flag indicating if there are elements in buffer, and using it instead of totalWeight in case onTimer is called
This commit is contained in:
Gilad Hoch 2017-08-08 14:30:32 +03:00 • committed by Patrik Nordwall
parent f623d10522
commit 0bcf32fccb
2 changed files with 30 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

23
akka-stream-tests/src/test/scala/akka/stream/scaladsl/FlowGroupedWithinSpec.scala
View file

@ -6,12 +6,14 @@ package akka.stream.scaladsl
import scala.collection.immutable
import scala.concurrent.duration._
import java.util.concurrent.ThreadLocalRandom.{ current ⇒ random }
import akka.stream.{ ActorMaterializer, ActorMaterializerSettings, ThrottleMode }
import akka.stream.testkit._
import akka.stream.testkit.Utils._
import scala.concurrent.Await
import akka.testkit.TimingTest
import akka.util.ConstantFun
class FlowGroupedWithinSpec extends StreamSpec with ScriptedTest {
@ -252,5 +254,26 @@ class FlowGroupedWithinSpec extends StreamSpec with ScriptedTest {
וupstream.sendComplete()
downstream.expectComplete()
}
"handle zero cost function to get only timed based grouping without limit" taggedAs TimingTest in {
val upstream = TestPublisher.probe[String]()
val downstream = TestSubscriber.probe[immutable.Seq[String]]()
Source
.fromPublisher(upstream)
.groupedWeightedWithin(1, 100.millis)(ConstantFun.zeroLong)
.to(Sink.fromSubscriber(downstream))
.run()
downstream.ensureSubscription()
downstream.request(1)
upstream.sendNext("333")
upstream.sendNext("22")
upstream.sendNext("333")
upstream.sendNext("22")
downstream.expectNoMsg(50.millis)
downstream.expectNext(Vector("333", "22", "333", "22"): immutable.Seq[String])
upstream.sendComplete()
downstream.expectComplete()
}
}
}

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

@ -1435,6 +1435,7 @@ private[stream] object Collect {
private var groupEmitted = true
private var finished = false
private var totalWeight = 0L
private var hasElements = false
override def preStart() = {
schedulePeriodically(GroupedWeightedWithin.groupedWeightedWithinTimer, interval)
@ -1444,8 +1445,9 @@ private[stream] object Collect {
private def nextElement(elem: T): Unit = {
groupEmitted = false
val cost = costFn(elem)
if (cost < 0) failStage(new IllegalArgumentException(s"Negative weight [$cost] for element [$elem] is not allowed"))
if (cost < 0L) failStage(new IllegalArgumentException(s"Negative weight [$cost] for element [$elem] is not allowed"))
else {
hasElements = true
if (totalWeight + cost <= maxWeight) {
buf += elem
totalWeight += cost
@ -1466,6 +1468,7 @@ private[stream] object Collect {
}
}
} else {
//we have a single heavy element that weighs more than the limit
if (totalWeight == 0L) {
buf += elem
totalWeight += cost
@ -1502,7 +1505,8 @@ private[stream] object Collect {
pending = null.asInstanceOf[T]
groupEmitted = false
} else {
totalWeight = 0
totalWeight = 0L
hasElements = false
}
pushEagerly = false
if (isAvailable(in)) nextElement(grab(in))
@ -1521,7 +1525,7 @@ private[stream] object Collect {
else tryCloseGroup()
}
override protected def onTimer(timerKey: Any) = if (totalWeight > 0) {
override protected def onTimer(timerKey: Any) = if (hasElements) {
if (isAvailable(out)) emitGroup()
else pushEagerly = true
}