akka/akka-core

Improve Sink.combine to return materialized values of all Sinks

開放

#23,958 建立於 2017年11月9日

 (19 則留言) (2 個反應) (0 位負責人)Scala (3,547 個分叉)batch import
1 - triagedhelp wantedt:stream

倉庫指標

星標
 (13,277 顆星)
PR 合併指標
 (平均合併 8天 19小時) (30 天內合併 10 個 PR)

描述

The current implementation of Sink.combine materializes to NotUsed for quite obvious reasons: There is no way of getting the materialization types of the rest parameter.

Since Sink.combine is a super handy method to easily create FanOut/FanIn etc. streams, I kinda expected it to give me the materialized values of all sinks in a tuple.

In my specific case, I implemented that bit myself in:

def combine[T, U, M1, M2](first: Sink[U, M1], second: Sink[U, M2])(
  strategy: Int ⇒ Graph[UniformFanOutShape[T, U], NotUsed]): Sink[T, (M1, M2)] = {
  Sink.fromGraph(GraphDSL.create(first, second)((_, _)) { implicit b => (s1, s2) =>
    import GraphDSL.Implicits._
    val d = b.add(strategy(2))

    d ~> s1
    d ~> s2

    SinkShape(d.in)
  })
}

I can imagine the overrides one would need to be relatively low. combine20 would probably be overkill and arguably, having combine2 one can just reapply that method to get a huge stacked tuple back in the end.

In any case: Do you think this would be a useful thing to have?

貢獻者指南