是否有类似Iteratee的概念从多个来源提取数据？

Question

可以使用流(懒惰列表)从一个数字(例如两个为了简单)中提取需求.Iteratees可用于处理来自单一来源的数据.

是否有类似Iteratee的功能概念来处理多个输入源？我可以想象一个Iteratee,它的状态信号来自它想要拉动的源.

Answer 1

要使用管道执行此操作,您可以将Pipe monad转换器嵌套在其自身内,对于您希望与之交互的每个生产者.例如:

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 curried函数一样,您可以使用composition和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.事实上,John Milikin和Oleg是这种方法的最初倡导者,我只是从他们那里偷走了这个想法.

Answer 2

我们在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
}

此代码构建Plan一个Machine使用该repeatedly方法"编译"到a 的代码.这里构建的类型是Tee[A, B, (A, B)]具有两个输入的机器.您在左侧和右侧请求输入awaits(left)和awaits(right),并输出emit.

还有一个Haskell版本的机器.