Элегантный способ повторного использования потоков akka-stream

Question

Я ищу способ легко использовать потоки akka-stream.

Я лечить Flow я намерен повторно использовать как функцию, так что я хотел бы сохранить свою подпись, как:

Flow[Input, Output, NotUsed]

Теперь, когда я использую этот поток, я хотел бы быть в состоянии «вызов «этот поток и сохранить результат в сторону для дальнейшей обработки.

Итак, я хочу начать с потока, испускающего [Input], применить мой поток и продолжить с испусканием потока [(Input, Output)].

пример:

val s: Source[Int, NotUsed] = Source(1 to 10) 

val stringIfEven = Flow[Int].filter(_ % 2 == 0).map(_.toString) 

val via: Source[(Int, String), NotUsed] = ??? 

Теперь это невозможно прямым путем, поскольку комбинируя поток с .via() даст мне Проточная излучающий только [Output]

val via: Source[String, NotUsed] = s.via(stringIfEven) 

Альтернатива, чтобы сделать мой многоразовый поток испускают [(Input, Output)] но это требует, чтобы каждый поток вводил свой вход через все этапы и делал мой код плохой.

Так что я придумал объединитель, как это:

def tupledFlow[In,Out](flow: Flow[In, Out, _]):Flow[In, (In,Out), NotUsed] = { 
    Flow.fromGraph(GraphDSL.create() { implicit b => 
    import GraphDSL.Implicits._ 

    val broadcast = b.add(Broadcast[In](2)) 
    val zip = b.add(Zip[In, Out]) 

    broadcast.out(0) ~> zip.in0 
    broadcast.out(1) ~> flow ~> zip.in1 

    FlowShape(broadcast.in, zip.out) 
}) 

}

, вещающие вход в поток и, а также в параллельной линии непосредственно -> как к «Zip» где я присоединяю значения к кортежу. Это может быть элегантно применен:

val tupled: Source[(Int, String), NotUsed] = s.via(tupledFlow(stringIfEven)) 

Все замечательно, но когда данный поток делает «фильтр» операцию - это объединитель застревает и прекращает обработку дальнейших событий.

Я предполагаю, что это связано с поведением «Zip», которое требует, чтобы все субпотоки выполняли то же самое - в моем случае одна ветвь передает данный объект напрямую, так что другой подпоток не может игнорировать этот элемент. filter(), а так как это происходит - поток останавливается, потому что Zip ждет нажатия.

Есть ли лучший способ достичь состава потока? Есть ли что-нибудь, что я могу сделать в моем tupledFlow, чтобы получить желаемое поведение, когда «поток» игнорирует элементы с помощью «фильтра»?

Answer 1

два возможных подхода - с дискуссионной элегантность - это:

1) избегать использования ступеней фильтрации, мутирует свой фильтр в Flow[Int, Option[Int], NotUsed]. Таким образом, вы можете применить свою оберточную ленту вокруг всего графика, как и ваш первоначальный план. Тем не менее, код выглядит более испорченным, и добавляется накладные расходы, проходя вокруг None с.

val stringIfEvenOrNone = Flow[Int].map{ 
    case x if x % 2 == 0 => Some(x.toString) 
    case _ => None 
} 

val tupled: Source[(Int, String), NotUsed] = s.via(tupledFlow(stringIfEvenOrNone)).collect{ 
    case (num, Some(str)) => (num,str) 
} 

2) отделяет фильтрацию и трансформирующие стадии, и применять те фильтрации перед вашими проносясь обертками. Вероятно, более легкий и лучший компромисс.

val filterEven = Flow[Int].filter(_ % 2 == 0) 

val toString = Flow[Int].map(_.toString) 

val tupled: Source[(Int, String), NotUsed] = s.via(filterEven).via(tupledFlow(toString)) 

EDIT

3) Проводка другого решения здесь для ясности, как в дискуссии в комментариях.

Эта обтекатель потоков позволяет испускать каждый элемент из заданного потока в сочетании с исходным элементом ввода, который сгенерировал его. Он работает для любого внутреннего потока (излучая 0, 1 или более элементов для каждого входа).

def tupledFlow[In,Out](flow: Flow[In, Out, _]): Flow[In, (In,Out), NotUsed] = 
    Flow[In].flatMapConcat(in => Source.single(in).via(flow).map(out => in -> out)) 

Answer 2

я пришел с реализацией TupledFlow, который работает, когда обернутый Flow использует filter() или mapAsync() и когда обернутый Flow испускает 0,1 или N элементов для каждого входа:

def tupledFlow[In,Out](flow: Flow[In, Out, _])(implicit materializer: Materializer, executionContext: ExecutionContext):Flow[In, (In,Out), NotUsed] = { 
    val v:Flow[In, Seq[(In, Out)], NotUsed] = Flow[In].mapAsync(4) { in: In => 
    val outFuture: Future[Seq[Out]] = Source.single(in).via(flow).runWith(Sink.seq) 
    val bothFuture: Future[Seq[(In,Out)]] = outFuture.map(seqOfOut => seqOfOut.map((in,_))) 
    bothFuture 
    } 
    val onlyDefined: Flow[In, (In, Out), NotUsed] = v.mapConcat[(In, Out)](seq => seq.to[scala.collection.immutable.Iterable]) 
    onlyDefined 
} 

единственным недостатком я см. здесь, что я создаю и материализую поток для одного объекта - просто чтобы получить понятие «вызов потока как функции».

Я не проводил никаких проверок эффективности, однако, поскольку тяжелая работа выполняется в обернутом Flow, который выполнен в будущем, я считаю, что это будет нормально.

Эта реализация проходит все тесты из https://gist.github.com/kretes/8d5f2925de55b2a274148b69f79e55ac#file-tupledflowspec-scala