Обработка нескольких файлов параллельно с потоками scalaz

Question

Я пытаюсь использовать scalaz-stream для обработки нескольких файлов одновременно, применяя одну функцию к каждой строке в файлах по всем файлам. Для конкретности, предположим, что у меня есть функция, которая принимает список строк

def f(lines: Seq[String]): Something = ??? 

и пару файлов, первый:

foo1 
foo2 
foo3 

второй:

bar1 
bar2 
bar3 

Результат всего процесса:

List(
    f(Seq("foo1", "bar1")), 
    f(Seq("foo2", "bar2")), 
    f(Seq("foo3", "bar3")) 
) 

(или, скорее всего, написан непосредственно в каком-либо другом файле)

Количество файлов неизвестно заранее, а количество строк может варьироваться в зависимости от разных файлов, но я в порядке с заполнением (во время выполнения) концы более коротких файлов со значениями по умолчанию или вырезание более длинных файлов.

По существу, мне нужен способ объединить Seq[Process[Task, String]] (полученный через что-то вроде io.linesR) в один Process[Task, Seq[String]].

Какой будет самый простой способ достичь этого?

Или, в общем, как я могу объединить n экземпляры Process[F, I] в одном экземпляре Process[F, Seq[I]]?

Я уверен, что есть какая-то стандартная комбинатор для этой цели, но я не был в состоянии понять это ...

Спасибо.

Answer 1

Этот комбинатор еще не существует, но вы можете добавить его. Я думаю, что это будет что-то вроде:

def zipN[F[_], A](xs: Seq[Process[F,A]]): Process[F,Seq[A]] = 
    if (xs.isEmpty) Process.halt 
    else xs.map(_ map (Vector(_))).reduceLeft(_.zipWith(_)(_ ++ _)) 

Можно также добавить zipAllN, который принимает значение раздуть последовательности с (и который использует zipAll и alignN, что позволяет потоки «выпасть» на выходе (так что выходная последовательность может быть короче.)

Я предлагаю вам реализовать его как «сбалансированное» сокращение, а не уменьшать слева или справа, поскольку оно будет более эффективным таким образом.

Пожалуйста, подайте запрос на растяжение + тесты, если вы в конечном итоге выполните это серьезно!