【发布时间】:2021-12-10 23:09:13
【问题描述】:
我很难获得 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
【问题讨论】:
-
如果这不仅仅是一个示例,并且您想在子流中通过传递产生副作用,为什么不直接使用
wireTap?subFlow.wireTap(recordedItems.add(_)).mergeSubstreams -
您是否尝试过从
ìmplicit class中删除差异? -
@LeviRamsey 这是一个有意识的决定,因为我们想要与
map相同的流语义(关于流控制)。
标签: scala generics akka-stream implicit implicit-class