В моем потоке есть поток, выходные данные которого являются объектами List[Any]. Я хочу, чтобы за mapAsync следовали некоторые другие этапы, каждый из которых обрабатывал отдельный элемент вместо списка. Как я могу это сделать?
Фактически я хочу подключить вывод
Flow[Any].map { msg =>
someListDerivedFrom(msg)
}
потребляться -
Flow[Any].mapAsyncUnordered(4) { listElement =>
actorRef ? listElement
}.someOtherStuff
Как мне это сделать?
Я думаю, что комбинатор, который вы ищете, это mapConcat. Этот комбинатор примет входной аргумент и вернет что-то, что является Iterable. Простой пример будет следующим:
implicit val system = ActorSystem()
implicit val mater = ActorMaterializer()
val source = Source(List(List(1,2,3), List(4,5,6)))
val sink = Sink.foreach[Int](println)
val graph =
source.
mapConcat(identity).
to(sink)
graph.run
Здесь мой Source выдает List элементов, а мой Sink принимает базовый тип того, что содержится в этих List. Я не могу соединить их напрямую вместе, так как типы разные. Но если я применю mapConcat между ними, они могут быть связаны, поскольку этот комбинатор сгладит эти List элементы, отправив их отдельные элементы (Int) вниз по течению. Поскольку элемент ввода для mapConcat уже является Iterable, вам нужно только использовать функцию identify в теле mapConcat, чтобы все заработало.
