如何制作“分支”管道？答案

【问题标题】：How do I make a "branched" Conduit?如何制作“分支”管道？
【发布时间】：2014-07-09 18:34:37
【问题描述】：

我希望将相同的数据分成两个“分支”分别处理，然后“加入”...

                                +----------+
                +---------+  -->| doublber |---   +--------+
   +--------+   |         |--   +----------+   -->|        |   +------+
   | source |-->| splitter|                       | summer |-->| sink |
   +--------+   |         |--   +----------+   -->|        |   +------+
                +---------+  -->| delayer  |---   +--------+
                                +----------+

我应该怎么做？

我的尝试：

import Data.Conduit
import Control.Monad.IO.Class
import qualified Data.Conduit.List as CL
-- import Data.Conduit.Internal (zipSources)
import Control.Arrow ((>>>))

source :: Source IO Int
source = do
    x <- liftIO $ getLine
    yield (read x)
    source

splitter :: Conduit Int IO (Int, Int)
splitter = CL.map $ \x -> (x,x)

doubler = CL.map (* 2)

delayer :: Conduit Int IO Int
delayer = do
    yield 0
    CL.map id

twoConduitBranches :: Monad m => Conduit a m b -> Conduit c m d -> Conduit (a,b) m (c,d)
twoConduitBranches q w = awaitForever $ \(x, y) -> do
    out1 <- undefined q x
    out2 <- undefined w y
    yield (out1, out2)


summer :: Conduit (Int,Int) IO Int
summer = CL.map $ \(x,y) -> x + y

sink :: Sink Int IO ()
sink = CL.mapM_ (show >>> putStrLn) 

-- combosrc = zipSources (source $= delayer) (source $= doubler)
main = source $= splitter $= twoConduitBranches doubler delayer $= summer $$ sink

我应该写什么来代替undefineds？

【问题讨论】：

标签： haskell conduit

【解决方案1】：

你可以这样做，但它很丑，希望实现能够清楚为什么它很丑，而不是管道的内置功能：

twoConduitBranches :: Monad m => Conduit a m c -> Conduit b m d -> Conduit (a,b) m (c,d)
twoConduitBranches q w = getZipConduit
      (ZipConduit (CL.map fst =$= q =$= CL.map Left)
    <* ZipConduit (CL.map snd =$= w =$= CL.map Right)) =$= collapse
  where
    collapse = do
        v1 <- await
        case v1 of
            Nothing -> return ()
            Just (Left _) -> error "out of sequence 1"
            Just (Right d) -> do
                v2 <- await
                case v2 of
                    Nothing -> error "mismatched count"
                    Just (Right _) -> error "out of sequence 2"
                    Just (Left c) -> do
                        yield (c, d)
                        collapse

（注意：我稍微调整了你的类型签名，我认为这是你真正想要的类型签名。）

方法如下：将q 转换为Conduit，它从每个传入的元组中获取第一个值，然后用Left 包装其输出。同样，我们从每个传入的元组中取出第二个值并将其传递给w，然后用Right 包装输出。

现在这些Conduits 具有相同的类型（它们接受相同的输入元组，并生成相同的 Either 值），我们使用ZipConduit 将它们组合起来，它在所有组件之间共享输入并将输出合并到单个流。

此流是Either c d 的流，而不是所需的(c, d)。为了进行最终转换，我们使用collapse。它会弹出一个 Right 和 Left 值，然后将它们放在一个它产生的单个元组中。

此函数假定输出值序列将始终是来自w 的一个值，然后是来自q 的一个值。如果发生其他任何事情，它将抛出异常。问题是：管道中没有任何内容暗示它们实际上会以相同的速率产生输出。事实上，管道是专门为避免这种假设而设计的！

因此，如果您知道您的两个组件管道将始终以相同的速率产生输出，则此功能将起作用。但一般来说这不是真的。

【讨论】：