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

Support primitives in Flow#collectType (#1490)

Browse source
This commit is contained in:
João Costa 2024-09-20 21:05:18 +01:00 • committed by GitHub
parent 22a7385113
commit 6ceb59ba26
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 13 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

13
stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowCollectTypeSpec.scala
View file

@ -25,7 +25,7 @@ class FlowCollectTypeSpec extends StreamSpec {
"A CollectType" must {
"collectType" in {
"collectType with references" in {
val fruit = Source(List(Orange, Apple, Apple, Orange))
val apples = fruit.collectType[Apple].runWith(Sink.seq).futureValue
@ -36,6 +36,17 @@ class FlowCollectTypeSpec extends StreamSpec {
all should equal(List(Orange, Apple, Apple, Orange))
}
"collectType with primitives" in {
val numbers = Source(List[Int](1, 2, 3) ++ List[Double](1.5))
val integers = numbers.collectType[Int].runWith(Sink.seq).futureValue
integers should equal(List(1, 2, 3))
val doubles = numbers.collectType[Double].runWith(Sink.seq).futureValue
doubles should equal(List(1.5))
val all = numbers.collectType[Any].runWith(Sink.seq).futureValue
all should equal(List(1, 2, 3, 1.5))
}
}
}

2
stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala
View file

@ -1686,7 +1686,7 @@ trait FlowOps[+Out, +Mat] {
* '''Cancels when''' downstream cancels
*/
def collectType[T](implicit tag: ClassTag[T]): Repr[T] =
collect { case c if tag.runtimeClass.isInstance(c) => c.asInstanceOf[T] }
collect { case tag(c) => c }
/**
* Chunk up this stream into groups of the given size, with the last group