标签: haskell-pipes

我使用以下方法编写了以下解析代码attoparsec:

data Test = Test {
  a :: Int,
  b :: Int
  } deriving (Show)

testParser :: Parser Test
testParser = do
  a <- decimal
  tab
  b <- decimal
  return $ Test a b

tParser :: Parser [Test]
tParser =  many' $ testParser <* endOfLine

这适用于小型文件,我这样执行:

main :: IO ()
main = do
  text <- TL.readFile "./testFile"
  let (Right a) = parseOnly (manyTill anyChar endOfLine *> tParser) text
  print a  

但是当文件大小超过70MB时,它会占用大量内存.作为解决方案,我想我会用attoparsec-conduit.在完成他们的API之后,我不确定如何让它们一起工作.我的解析器有类型,Parser Test但它 …

我正在使用管道库,需要String使用ASCII编码将ByteString流转换为线流(即).我知道还有其他库(Pipes.Text和Pipes.Prelude)可能让我更容易从文本文件中产生行,但由于其他一些代码,我需要能够String从Producer中获取行ByteString.

更正式地说,我需要将a转换Producer ByteString IO ()为a Producer String IO (),从而产生线条.

我相信,这一定是一个班轮对于有经验的管道,程序员,但我至今没能成功地通过所有的黑客FreeT,并Lens在管道-字节串-trickery.

任何帮助深表感谢!

斯蒂芬

我正在尝试从一个管道中读取一组最多50个项目,并在一次IO操作中处理它们.(这个用例就是我试图将数据插入到数据库中,我想在一个事务中执行整个批处理,因为它的效率要高得多).这是我到目前为止的简化版本:

type ExampleType = Int

doSomething :: [ExampleType] -> IO ()
doSomething = undefined

inGroupsOf50 :: Monad m => Producer ExampleType m () -> m ()
inGroupsOf50 input =
    runEffect $ input >-> loop
        where loop = do entries <- replicateM 50 await
                    lift $ doSomething entries  --Insert a bunch all in one transaction
                    loop

问题是我可以告诉的,除非要插入的项目数量除以50,否则我会错过一些.replicateM 50 await如果输入结束但是我无法弄清楚如何编写它,那么我真正想要的东西会给我多达50个项目或更少.

我一直在想管道解析可能是正确的库.draw看起来有一个很有希望的签名...但到目前为止,所有的位都没有在我脑海中融合在一起.我有一个producer,我正在写一个consumer,我真的不知道这与a的概念有什么关系parser.

在reddit上有一个存档的线程，该线程表示管道基本上不能是箭头b / c箭头需要同步。该线程链接在这里https://www.reddit.com/r/haskell/comments/rq1q5/conduitssinks_and_refactoring_arrows/

我看不到“同步”出现的地方，因为这不是箭头定义的一部分。另外，我在github https://github.com/cmahon/interactive-brokers上偶然发现了这个项目，该项目将管道明确地视为箭头。为了方便起见，我在此处粘贴实例def。我在这里想念什么？

-- The code in this module was provided by Gabriel Gonzalez

{-# LANGUAGE RankNTypes #-}

module Pipes.Edge where

import           Control.Arrow
import           Control.Category (Category((.), id))
import           Control.Monad ((>=>))
import           Control.Monad.Trans.State.Strict (get, put)
import           Pipes
import           Pipes.Core (request, respond, (\>\), (/>/), push, (>~>))
import           Pipes.Internal (unsafeHoist)
import           Pipes.Lift (evalStateP)
import           Prelude hiding ((.), id)

newtype Edge m r a b = Edge { unEdge :: a -> Pipe a b m r }

instance (Monad …

使用管道库,我想编写一个程序来从某个源读取数据并单独累积它(比如说Sum).最简单的方法是,

 import Control.Proxy as 
 import Data.Monoid (Sum)

 main = do
     let source = enumFromToS (0::Int) 5
     a <- runWriterT $ runProxy $ source >-> foldD Sum
     print a

当然,虽然这适用于小型源,但由于WriterT累加器的惰性,大输入会导致可怕的堆栈溢出.

值得庆幸的是,它似乎pipes预料到了这一点,提供了一个WriterP带有严格累加器的代理.不幸的是,围绕这个代理的文档非常稀少.经过一番探索(并简化问题而不是为每个下游元素累积1),我来到这个程序,

import Control.Proxy
import Control.Proxy.Trans.Writer
import Data.Monoid (Sum)

main = do
    let source = enumFromToS (0::Int) 5
    a <- runProxy $ runWriterK $ source >-> \x->tell (Sum 1::Sum Int)
    print a

当然,这个程序甚至没有正确地执行简化的任务,因为它累积到1而不是6.如果我没有弄错,这个行为可以解释为管道在终止之前只读取一个元素.要重复直到输入结束,我想出了以下内容,

import Control.Proxy
import Control.Proxy.Trans.Writer
import Data.Monoid (Sum)

main = do
    let …

例如,有......

consumer :: Proxy p => () -> Consumer p a (EitherT String IO) ()
producer :: Proxy p => () -> Producer p a (EitherT ByteString IO) r

......我该如何工作？

session :: EitherT ByteString (EitherT String IO) ()
session = runProxy $ producer >-> consumer

注:我读过混合基地单子在Control.Proxy.Tutorial.我得到了第一个例子,但无法理解人为的例子.更多的例子,不是那么明显但不那么做作,可能会澄清如何使用hoist和lift匹配任何基础monad的组合.

我很难写pipe这个签名:

toOneBigList :: (Monad m, Proxy p) => () -> Pipe p a [a] m r

它应该简单地a从上游获取所有s并将它们发送到下游列表中.

我的所有尝试都从根本上被打破了.

任何人都能指出我正确的方向吗？

这个问题有点像高尔夫球和很多新手.

我pipes在Haskell中使用了很棒的库,我想拆分管道以沿多个通道发送相同的数据(进行广播).本Pipes.Concurrent教程建议使用spawn创建邮箱,利用Output幺半群状态.例如,我们可能会这样做:

main = do
 (output1, input1) <- spawn Unbounded
 (output2, input2) <- spawn Unbounded
 let effect1 = fromInput input1 >-> pipe1
 let effect2 = fromInput input2 >-> pipe2
 let effect3 = P.stdinLn >-> toOutput (output1 <> output2)
 ...

通过邮箱的这种间接是否真的有必要？我们可以写这样的东西吗？

main = do
 let effect3 = P.stdinLn >-> (pipe1 <> pipe2)
 ...

以上不编译,因为Pipe没有Monoid实例.有这么好的理由吗？第一种方法真的是拆分管道最干净的方法吗？

我的类型:

data Test = Test {
 a :: Int,
 b :: Int
} deriving (Show)

我的解析器:

testParser :: Parser Test
testParser = do
  a <- decimal
  tab
  b <- decimal
  return $ Test a b

tab = char '\t'

现在为了跳过第一行,我做了这样的事情:

import qualified System.IO as IO    

parser :: Parser Test
parser = manyTill anyChar endOfLine *> testParser

main = IO.withFile testFile IO.ReadMode $ \testHandle -> runEffect $
         for (parsed (parser <* endOfLine) (fromHandle testHandle)) (lift . print)

但是上面的parser函数会使每个备用链接跳过(很明显).如何以与Pipes生态系统一起工作的方式跳过第一行(Producer …

我需要组合一个消费者和一个管道,以便消费者的输出将输入管道的输入.

我想这可以用这样的组合器来解决:

Consumer i m r -> (r -> Producer o m r') -> Pipe i o m r'

或这个:

Consumer i m i' -> Pipe i' o m r -> Pipe i o m r

或者如下所示的提升功能:

Consumer i m r -> Pipe i o m r

或者像这样:

Consumer i m o -> Pipe i o m r

我试过consumer >~ pipe没有成功.那么如何处理呢？