如何通过子类型 B<:A 过滤类似 Stream[Future[A]] 的类？答案

【问题标题】：How to filter a Stream[Future[A]]-like class by sub-type B<:A?如何通过子类型 B<:A 过滤类似 Stream[Future[A]] 的类？
【发布时间】：2016-02-11 09:57:24
【问题描述】：

我有这个类来管理异步消息流的接收。它有类型参数，比如A，我喜欢实现一个按类型过滤这些消息的函数，比如B<:A

但由于类型擦除，我的第一个实现无法正常工作（参见下面的示例）。有什么好的方法吗？

这是我的问题的简化示例：

package test.filterByType

import scala.concurrent.Await
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration._
import scala.concurrent.Future

trait MessageStream[A]{ self =>

  def next(): Future[A]

  def filter[B<:A]: MessageStream[B] = new MessageStream[B]{
    def next(): Future[B] = self.next.flatMap{
      case b: B => Future.successful(b)
      case _    => next()
      }
  }
}

case class MessageStreamImpl[A](msg: IndexedSeq[A]) extends MessageStream[A]{
  private var index = 0
  def next() = {
    index += 1
    Future.successful(msg(index-1))
  }
}

object Main{
  trait A
  case class B(i: Int) extends A
  case class C(i: Int) extends A

  def main(args: Array[String]){
    val msg: IndexedSeq[A] = (1 to 10).map{ i => if(i%2==0) B(i) else C(i) }
    val streamOfA = MessageStreamImpl(msg)
    val streamOfB = streamOfA.filter[B]

    val b: B = Await.result(streamOfB.next(), 1 second)
  }
}

在编译时，我得到warning: abstract type pattern B is unchecked since it is eliminated by erasure，确实代码不起作用。如果我执行 Main，我会收到错误： lang.ClassCastException: test.filterByType.Main$C cannot be cast to test.filterByType.Main$B

这是因为它没有过滤掉第一个列表项C(1)

【问题讨论】：

标签： scala type-erasure type-parameter

【解决方案1】：

这个小调整可以解决问题

import scala.reflect.ClassTag
def filter[B<:A](implicit C: ClassTag[B]): MessageStream[B] = new MessageStream[B]{
    def next(): Future[B] = self.next.flatMap{
        case b: B => Future.successful(b)
        case _    => next()
    }
}

【讨论】：

我以前见过ClassTag 这个东西。你能解释一下为什么它会起作用吗？
A ClassTag[T] 存储给定类型 T 的已擦除类。所以本质上它是为了解决你面临的问题。

【解决方案2】：

我找到了一个可行的替代方案：

def collect[B<:A](f: PartialFunction[A,B]) = new MessageStream[B] {
  override def next(): Future[B] = self.next().flatMap { m =>
    if (f.isDefinedAt(m)) Future.successful(f(m))
    else next()
  }
}

【讨论】：