Передача элемента в поток

Question

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

У меня есть что-то вроде этого:

Stream.flow(file1, file2, file3)
.via(processFileFlow)
.via(archiveFlow)

ProcessFileFlow выглядит следующим образом:

Flow.create().map(file1 -> Files.lines(file1)).map(line -> {send(line); return line;})

Я могу создать пару файлов и строку файла, чтобы передать файл в поток, но затем метод архива вызывается слишком много раз, потому что файлдублируется столько раз, сколько строк в этом файле.Я хочу запустить поток архива только один раз для файла.

Каков наилучший способ сделать это?

Answer 1

Посмотрите на метод сгиба [T] (ноль: T) (f: (T, Out) ⇒ T), который позволил бы вам снова сократить (файл, строка), например, последовательность файлов. перед вызовом шага архивирования.