scala - 阿卡流 : Outlet to FlowOps

Question

是否可以举起 Outlet[A]进入 FlowOps[A, _] ?那就是如果我有这个:

import akka.NotUsed
import akka.stream.Outlet
import akka.stream.scaladsl.{FlowOps, GraphDSL, Source}

def filter(in: Outlet[Double])
          (implicit b: GraphDSL.Builder[NotUsed]): Outlet[Double] = {
  val in0: FlowOps[Double, NotUsed] = ???
  val res = in0.grouped(8).statefulMapConcat[Double] { () =>
    seq => seq.reverse
  }
  res
  ??? : Outlet[Double]
}

为了做到这一点 grouped如果 in要么是 Source或 Flow ?

Answer 1

这只是缺少隐式的导入，然后 Outlet可用于流操作:

def filter(in: Outlet[Double])
            (implicit b: GraphDSL.Builder[NotUsed]): Outlet[Double] = {
    import GraphDSL.Implicits._
    import scala.collection.immutable.{Seq => ISeq}
    val grouped: PortOps[ISeq[Double]] = in.grouped(8)
    val flattened: PortOps[Double] = grouped.statefulMapConcat[Double] { () =>
      seq => seq.reverse
    }
    flattened.outlet
  }