Акка, объединяющая Мойки без доступа к Потокам

Question

Я использую API, который принимает одну раковину AKKA и заполняет ее данными:

def fillSink(sink:Sink[String, _])

Есть ли способ, не углубляясь в глубины akka, обрабатывать вывод с двумя приемниками вместоодного?

Например,

val mySink1:Sink = ...
val mySink2:Sink = ...
//something
fillSink( bothSinks )

Если бы у меня был доступ к потоку, используемому методом fillSink, я мог бы использовать flow.alsoTo(mySink1).to(mySink2), но поток не был открыт.

Единственный обходной путь на данный момент - передать один Sink, который обрабатывает строки и передает его двум StringBuilder для замены mySink1/mySink2, но кажется, что это побеждает точку AKKA.Не потратив пару дней на изучение AKKA, я не могу сказать, есть ли способ разделить вывод из приемников.

Спасибо!

Answer 1

Оператор combine Sink, который объединяет два или более Sinks с использованием предоставленной функции Int => Graph[UniformFanOutShape[T, U], NotUsed]], может быть тем, что вы ищете:

def combine[T, U](first: Sink[U, _], second: Sink[U, _], rest: Sink[U, _]*)(strategy: Int => Graph[UniformFanOutShape[T, U], NotUsed]): Sink[T, NotUsed]

Тривиализированный пример:

val doubleSink = Sink.foreach[Int](i => println(s"Doubler: ${i*2}"))
val tripleSink = Sink.foreach[Int](i => println(s" Triper: ${i*3}"))

val combinedSink = Sink.combine(doubleSink, tripleSink)(Broadcast[Int](_))

Source(List(1, 2, 3)).runWith(combinedSink)

// Doubler: 2
//  Triper: 3
// Doubler: 4
//  Triper: 6
// Doubler: 6
//  Triper: 9