Akka Stream: Ausgang zu FlowOps

Question

Ist es möglich, einen Outlet[A] in FlowOps[A, _] zu heben? Das ist, wenn ich diese:

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] 
} 

Um sicherzustellen, dass grouped Anruf Arbeit zu machen, wie es wäre, wenn in ist entweder ein Source oder ein Flow?

Answer 1

Dies war nur der Import der implicits fehlt, wird dann Outlet für Strömungs ops verwendet werden:

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 
    }