0

我很难得到一个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] 然后将其用于类型F和C.SubFlow

斯卡拉版本：2.12.12

4

1 回答 1

0

Try to use higher-kinded type parameter as in the definition of SubFlow

implicit class SideEffectfulSubFlowOps[+Out, +Mat, +FOps[+_], C](val enrichedSubFlow: SubFlow[Out, Mat, FOps, C]) extends AnyVal
于 2021-10-25T16:04:02.050 回答