akka流SubFlow的隐式类(泛型路径依赖类型)

Question

我很难得到一个implicit class为akka.stream.scaladsl.SubFlow编译。

我的测试代码:

val subFlow = Source(List("1", "2", "3"))
.groupBy(1, f)
val richSubFlow = new SideEffectfulSubFlowOps(subFlow)
val got = richSubFlow
.withSideEffect((elem: String) => recordedItems.add(elem))
.mergeSubstreams
.to(Sink.seq)
/* In the end I would like to write it like this:
val got = Source(List("1", "2", "3"))
.groupBy(1, f)
.withSideEffect((elem: String) => recordedItems.add(elem))
.mergeSubstreams
.to(Sink.seq)
*/ 

目前为止的隐式类。

implicit class SideEffectfulSubFlowOps[+Out, +Mat, FOps <: FlowOps[Out, Mat], C](val enrichedSubFlow: SubFlow[Out, Mat, FOps#Repr, C]) extends AnyVal {
/** Perform a side effect without mutating the stream's element.
*  Unlike [[SubFlow.alsoTo]] and [[SubFlow.wireTap]], this operation has the same semantics as [[SubFlow.map]] regarding backpressure, and concurrency */
def withSideEffect(f: Out => Unit): enrichedSubFlow.Repr[Out] = {
enrichedSubFlow.map { o =>
f(o)
o
}
}
}

不幸的是，我不能找出适当的泛型类型来定义隐式类。

编译错误:

[error] SubFlowExtensionsSpec.scala:21:43: type mismatch;
[error]  found   : akka.stream.scaladsl.SubFlow[String,akka.NotUsed,[+O]akka.stream.scaladsl.Source[O,akka.NotUsed],akka.stream.scaladsl.RunnableGraph[akka.NotUsed]]
[error]  required: akka.stream.scaladsl.SubFlow[?,?,?#Repr,?]
[error]       val x = new SideEffectfulSubFlowOps(subFlow)

查看子流的定义:trait SubFlow[+Out, +Mat, +F[+_], C] extends FlowOps[Out, Mat]我不明白我需要如何在我的隐式类上定义泛型类型，然后用于SubFlow的F和C类型。

Scala版本:2.12.12

Answer 1

尝试使用SubFlow

定义中的高级类型参数

implicit class SideEffectfulSubFlowOps[+Out, +Mat, +FOps[+_], C](val enrichedSubFlow: SubFlow[Out, Mat, FOps, C]) extends AnyVal