如何处理一系列 Akka 流源?

Question

我们有一个可以处理事件的Sink：

def parseEvent(): Sink[T, Future[akka.Done]] = {
Sink.foreach[T] { event => {
// Do stuff with the event
}}
}

这适用于单个Source：

val mySource: Source[T] = ...  
mySource.takeWhile( someCheck, true ).runWith(parseEvent)

如果你有：

val mySources: Seq[Source[T]] = ...

所有源都应并行运行，所有事件都应达到parseEvent。

Answer 1

以下内容应该符合要求：

import akka.NotUsed
import akka.stream.scaladsl.{ Concat, Merge, Source }
def sourceFromSources[T](sources: Seq[Source[T, NotUsed]]): Source[T, NotUsed] =
sources.size match {
case s if s < 1 => Source.empty[T]
case 1 => sources.head
case 2 => sources.head.merge(sources(1))
case _ => Source.combine(sources.head, sources(1), sources.drop(2): _*)(Merge(_))
}

合并策略"合并多个流，在元素从输入流到达时获取元素"，并随机选择多个流是否具有可用元素。 背压从下游传播到上游。