haskell - 将两个消费者连接成一个返回多个值的消费者？

Question

我一直在尝试新的 Pipes-http 包，并且有了一个想法。我有两个用于网页的解析器，一个返回行项目，另一个返回页面其他位置的数字。当我抓取页面时，最好将这些解析器串在一起并同时从同一个字节串生成器获取它们的结果，而不是两次获取页面或将所有 html 获取到内存中并解析两次。 p>

换句话说，假设您有两个消费者:

c1 :: Consumer a m r1
c2 :: Consumer a m r2

是否可以创建这样的函数:

combineConsumers :: Consumer a m r1 -> Consumer a m r2 -> Consumer a m (r1, r2)
combineConsumers = undefined

我已经尝试了一些方法，但我无法弄清楚。我理解如果不可能的话，但是会很方便。

编辑:

很抱歉，我对 Pipes-attoparsec 做出了假设，因为我对管道 attoparsec 的经验导致我问了错误的问题。当我假设它会返回一个管道 Consumer 时，Pipes-attoparsec 将 attoparsec 转换为管道解析器。这意味着我实际上无法将两个 attoparsec 解析器转变为获取文本并返回结果的消费者，然后将它们与普通的旧管道生态系统一起使用。抱歉，我只是不明白管道解析。

尽管这对我没有帮助，但 Arthur 的回答几乎就是我提出问题时的设想，而且我将来可能最终会使用他的解决方案。与此同时，我将使用管道。

Answer 1

如果结果是“monoidal”，您可以将 Pipes prelude 中的 tee 函数与 WriterT 结合使用。

{-# LANGUAGE OverloadedStrings #-}

import Data.Monoid
import Control.Monad
import Control.Monad.Writer
import Control.Monad.Writer.Class
import Pipes
import qualified Pipes.Prelude as P
import qualified Data.Text as T

textSource :: Producer T.Text IO ()
textSource = yield "foo" >> yield "bar" >> yield "foo" >> yield "nah"

counter :: Monoid w => T.Text 
                    -> (T.Text -> w) 
                    -> Consumer T.Text (WriterT w IO) ()
counter word inject = P.filter (==word) >-> P.mapM (tell . inject) >-> P.drain

main :: IO ()
main = do
    result <-runWriterT $ runEffect $ 
        hoist lift textSource >-> 
        P.tee (counter "foo" inject1) >-> (counter "bar" inject2)
    putStrLn . show $ result
    where
    inject1 _ = (,) (Sum 1) mempty
    inject2 _ = (,) mempty (Sum 1)

更新:正如评论中提到的，我看到的真正问题是在管道中解析器不是消费者。如果两个解析器对于剩余部分有不同的行为，如何同时运行它们？如果其中一个解析器想要“取消绘制”某些文本而另一个解析器不想，会发生什么？

一种可能的解决方案是在不同的线程中以真正并发的方式运行解析器。 Pipes-concurrency 包中的原语允许您通过将相同的数据写入两个不同的邮箱来“复制”Producer。然后每个解析器可以用它自己的生产者副本做任何它想做的事情。下面是一个还使用 pipes-parse、pipes-attoparsec 和 async 包的示例:

{-# LANGUAGE OverloadedStrings #-}

import Data.Monoid
import qualified Data.Text as T
import Data.Attoparsec.Text hiding (takeWhile)
import Data.Attoparsec.Combinator
import Control.Applicative
import Control.Monad
import Control.Monad.State.Strict
import Pipes
import qualified Pipes.Prelude as P
import qualified Pipes.Attoparsec as P
import qualified Pipes.Concurrent as P
import qualified Control.Concurrent.Async as A

parseChars :: Char -> Parser [Char] 
parseChars c = fmap mconcat $ 
    many (notChar c) *> many1 (some (char c) <* many (notChar c))

textSource :: Producer T.Text IO ()
textSource = yield "foo" >> yield "bar" >> yield "foo" >> yield "nah"

parseConc :: Producer T.Text IO () 
          -> Parser a 
          -> Parser b 
          -> IO (Either P.ParsingError a,Either P.ParsingError b)
parseConc producer parser1 parser2 = do
    (outbox1,inbox1,seal1) <- P.spawn' P.Unbounded
    (outbox2,inbox2,seal2) <- P.spawn' P.Unbounded
    feeding <- A.async $ runEffect $ producer >-> P.tee (P.toOutput outbox1) 
                                              >->        P.toOutput outbox2
    sealing <- A.async $ A.wait feeding >> P.atomically seal1 >> P.atomically seal2
    r <- A.runConcurrently $ 
        (,) <$> (A.Concurrently $ parseInbox parser1 inbox1)
            <*> (A.Concurrently $ parseInbox parser2 inbox2)
    A.wait sealing
    return r 
    where
    parseInbox parser inbox = evalStateT (P.parse parser) (P.fromInput inbox)

main :: IO ()
main = do
    (Right a, Right b) <- parseConc textSource (parseChars 'o')  (parseChars 'a')
    putStrLn . show $ (a,b) 

结果是:

("oooo","aa")

我不确定这种方法会带来多少开销。