Scala Akka Streams Merging Filter And Map

Question

I wonder if there is any way to optimize the following Scala code because it doesn't look very efficient. Basically, I just want to remove any object which is not a Tweet from the flow and map it to Tweet instead of Any.

val tweetsFlow = Flow[Any].filter({
  case _: Tweet => true
  case _ => false
}).map({
  case tweet: Tweet => tweet
})

Answer 1

You might use collect method, some like this

val tws = Vector(
  "foo",
  Tweet(Author("foo"), tms, "foo #akka bar"),
  1000L,
  Tweet(Author("bar"), tms, "foo #spray bar"),
  Tweet(Author("baz"), tms, "foo bar"),
  1
)

val tflow = Flow[Any].collect {
  case x: Tweet => x
}
Source(tws).via(tflow)

Answer 2

Since you're only interested in filtering on a specific type you can use collectType:

// case class Tweet(title: String)
// val mixedSource: Source[Any, NotUsed] = Source(List(Tweet("1"), 3, Tweet("2")))
val tweetsSource: Source[Tweet, NotUsed] = mixedSource.collectType[Tweet]
// tweetsSource.runForeach(println)
// Tweet("1")
// Tweet("2")