Akka Stream 源类型中的#Repr[A]答案

【问题标题】：#Repr[A] in Akka Stream source typeAkka Stream 源类型中的#Repr[A]
【发布时间】：2016-08-29 13:59:11
【问题描述】：

我有以下代码创建输出MarketFactors 的源。但是，我希望能够切换随机生成器的类型或 toMarketFactors 函数的作用。

但是，由于初始源上的所有转换都属于类型，因此我无法将其转换为类实现的特征中的泛型类型 (Option[Source[MarketFactors, NotUsed]])，我需要做什么？

我正在附加 Sink 并稍后实现，这是一种不好的做法吗？

override def factors: Option[Source[CorrelatedRandomVectorGenerator, NotUsed]#Repr[MarketFactors]] = {
    val source = for { g <- generator } yield Source.repeat(g)

    val marketFactors = Flow[CorrelatedRandomVectorGenerator].map(toMarketFactors)

    source.map(_.via(marketFactors))
}

【问题讨论】：

标签： scala stream akka

【解决方案1】：

在factors 方法上使用类型参数。

case class CorrelatedRandomVectorGenerator()
case class MarketFactors1()
case class MarketFactors2()

def factors[T](transformation: CorrelatedRandomVectorGenerator => T): Option[Source[T, NotUsed]] = {
  val generator = Option.empty[CorrelatedRandomVectorGenerator]
  val source = for { g <- generator } yield Source.repeat(g)

  val marketFactors = Flow[CorrelatedRandomVectorGenerator].map(transformation)
  source.map(_.via(marketFactors))
}

val toMarketFactors1: CorrelatedRandomVectorGenerator => MarketFactors1 = ???
val toMarketFactors2: CorrelatedRandomVectorGenerator => MarketFactors2 = ???

val source1: Option[Source[MarketFactors1, NotUsed]] = factors(toMarketFactors1)
val source2: Option[Source[MarketFactors2, NotUsed]] = factors(toMarketFactors2)

【讨论】：