Существует ли итерационная концепция, которая извлекает данные из нескольких источников?

Ответ 1

Чтобы сделать это с помощью труб, вы вставляете трансформатор для траншей труб внутри себя, один раз для каждого производителя, с которым вы хотите взаимодействовать. Например:

import Control.Monad
import Control.Monad.Trans
import Control.Pipe

producerA, producerB :: (Monad m) => Producer Int m ()
producerA = mapM_ yield [1,2,3]
producerB = mapM_ yield [4,5,6]

consumes2 :: (Show a, Show b) =>
    Consumer a (Consumer b IO) r
consumes2 = forever $ do
    a <- await       -- await from outer producer
    b <- lift await  -- await from inner producer
    lift $ lift $ print (a, b)

Точно так же, как функция Haskell с несколькими переменными, вы частично применяете ее к каждому источнику, используя композицию и runPipe:

consumes1 :: (Show b) => Consumer b IO ()
consumes1 = runPipe $ consumes2 <+< producerA

fullyApplied :: IO ()
fullyApplied = runPipe $ consumes1 <+< producerB

Вышеуказанная функция выводится при запуске:

>>> fullyApplied
(1, 4)
(2, 5)
(3, 6)

Этот трюк работает для того, чтобы уступить или ждать любого количества труб вверх или вниз по течению. Он также работает для прокси-серверов, двунаправленных аналогов для труб.

Изменить: Обратите внимание, что это также работает для любой библиотеки iteratee, а не только pipes. Фактически, Джон Миликин и Олег были оригинальными сторонниками такого подхода, и я просто украл их у них.

Ответ 2

Мы используем Machines в Scala, чтобы вытащить не только два, но и произвольное количество источников.

Два примера бинарных объединений предоставляются самой библиотекой в ​​модуле Tee: mergeOuterJoin и hashJoin. Вот как выглядит код для hashJoin (он предполагает, что оба потока отсортированы):

/**
 * A natural hash join according to keys of type `K`.
 */
def hashJoin[A, B, K](f: A => K, g: B => K): Tee[A, B, (A, B)] = {
  def build(m: Map[K, A]): Plan[T[A, B], Nothing, Map[K, A]] = (for {
    a  <- awaits(left[A])
    mp <- build(m + (f(a) -> a))
  } yield mp) orElse Return(m)
  for {
    m <- build(Map())
    r <- (awaits(right[B]) flatMap (b => {
      val k = g(b)
      if (m contains k) emit(m(k) -> b) else Return(())
    })) repeatedly
  } yield r
}

Этот код создает a Plan, который "скомпилирован" для Machine с помощью метода repeatedly. Тип, создаваемый здесь, Tee[A, B, (A, B)], который является машиной с двумя входами. Вы запрашиваете входы слева и справа с помощью awaits(left) и awaits(right), и вы выводите с помощью emit.

Существует также версия Haskell для машин.

Ответ 3

Conduits (и, он может быть создан для Pipes, но этот код еще не выпущен) имеет примитив zip, который принимает два восходящих потока и объединяет их как поток кортежей.

Ответ 4

Посмотрите pipe library, где вертикальная конкатенация может делать то, что вы хотите. Например,

import Control.Pipe
import Control.Monad
import Control.Monad.State
import Data.Void

source0, source1 :: Producer Char IO ()
source0 = mapM_ yield "say"
source1 = mapM_ yield "what"

sink :: Show b => Consumer b IO ()
sink = forever $ await >>= \x -> lift $ print x

pipeline :: Pipe () Void IO ()
pipeline = sink <+< (source0 >> source1)

Оператор последовательности (>>) вертикально объединяет источники, получая выход (на runPipe)

's'
'a'
'y'
'w'
'h'
'a'
't'