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

Fix compilation of stream tests on Scala 2.12 (#31062)

Browse source
This commit is contained in:
Arnout Engelen 2022-01-14 12:13:42 +01:00 • committed by GitHub
parent 1f2e15d944
commit 9d0684d875
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 6 additions and 6 deletions
Download patch file Download diff file Expand all files Collapse all files

12
akka-stream-tests/src/test/scala/akka/stream/scaladsl/AggregateWithBoundarySpec.scala
View file

@ -24,7 +24,7 @@ class AggregateWithBoundarySpec extends StreamSpec {
val groupSize = 3 val groupSize = 3
val result = Source(stream) val result = Source(stream)
.aggregateWithBoundary(allocate = () => ListBuffer.empty[Int])(aggregate = (buffer, i) => { .aggregateWithBoundary(allocate = () => ListBuffer.empty[Int])(aggregate = (buffer, i) => {
buffer.addOne(i) buffer += i
(buffer, buffer.size >= groupSize) (buffer, buffer.size >= groupSize)
}, harvest = buffer => buffer.toSeq, emitOnTimer = None) }, harvest = buffer => buffer.toSeq, emitOnTimer = None)
.runWith(Sink.collection) .runWith(Sink.collection)
@ -39,7 +39,7 @@ class AggregateWithBoundarySpec extends StreamSpec {
val result = Source(stream) val result = Source(stream)
.aggregateWithBoundary(allocate = () => ListBuffer.empty[Int])( .aggregateWithBoundary(allocate = () => ListBuffer.empty[Int])(
aggregate = (buffer, i) => { aggregate = (buffer, i) => {
buffer.addOne(i) buffer += i
(buffer, buffer.size >= groupSize) (buffer, buffer.size >= groupSize)
}, },
harvest = buffer => buffer.toSeq :+ -1, // append -1 to output to demonstrate the effect of harvest harvest = buffer => buffer.toSeq :+ -1, // append -1 to output to demonstrate the effect of harvest
@ -56,7 +56,7 @@ class AggregateWithBoundarySpec extends StreamSpec {
val result = Source(stream) val result = Source(stream)
.aggregateWithBoundary(allocate = () => ListBuffer.empty[Int])(aggregate = (buffer, i) => { .aggregateWithBoundary(allocate = () => ListBuffer.empty[Int])(aggregate = (buffer, i) => {
buffer.addOne(i) buffer += i
(buffer, buffer.sum >= weight) (buffer, buffer.sum >= weight)
}, harvest = buffer => buffer.toSeq, emitOnTimer = None) }, harvest = buffer => buffer.toSeq, emitOnTimer = None)
.runWith(Sink.collection) .runWith(Sink.collection)
@ -145,7 +145,7 @@ class AggregateWithTimeBoundaryAndSimulatedTimeSpec extends AnyWordSpecLike with
.fromPublisher(p) .fromPublisher(p)
.aggregateWithTimeBoundary(allocate = ListBuffer.empty[Int])( .aggregateWithTimeBoundary(allocate = ListBuffer.empty[Int])(
aggregate = (buffer, i) => { aggregate = (buffer, i) => {
buffer.addOne(i) buffer += i
(buffer, false) (buffer, false)
}, },
harvest = seq => seq.toSeq, harvest = seq => seq.toSeq,
@ -189,7 +189,7 @@ class AggregateWithTimeBoundaryAndSimulatedTimeSpec extends AnyWordSpecLike with
.fromPublisher(p) .fromPublisher(p)
.aggregateWithTimeBoundary(allocate = ListBuffer.empty[Int])( .aggregateWithTimeBoundary(allocate = ListBuffer.empty[Int])(
aggregate = (buffer, i) => { aggregate = (buffer, i) => {
buffer.addOne(i) buffer += i
(buffer, false) (buffer, false)
}, },
harvest = seq => seq.toSeq, harvest = seq => seq.toSeq,
@ -230,7 +230,7 @@ class AggregateWithTimeBoundaryAndSimulatedTimeSpec extends AnyWordSpecLike with
.fromPublisher(upstream) .fromPublisher(upstream)
.aggregateWithTimeBoundary(allocate = ListBuffer.empty[Int])( .aggregateWithTimeBoundary(allocate = ListBuffer.empty[Int])(
aggregate = (buffer, i) => { aggregate = (buffer, i) => {
buffer.addOne(i) buffer += i
(buffer, false) (buffer, false)
}, },
harvest = buffer => buffer.toSeq, harvest = buffer => buffer.toSeq,