标签: conduit

首先,我要完成的任务的简化版本:我有几个大文件(相当于30GB),我想修剪重复的条目.为此,我建立了一个数据散列数据库,逐个打开文件,散列每个项目,并将其记录在数据库和输出文件中,如果它的散列不在数据库中.

我知道如何使用迭代器,枚举器,我想尝试管道.我也知道如何用管道来做,但现在我想使用管道和持久性.我遇到了类型问题,可能还有整个概念ResourceT.

这里有一些伪代码来说明问题:

withSqlConn "foo.db" $ runSqlConn $ runResourceT $ 
     sourceFile "in" $= parseBytes $= dbAction $= serialize $$ sinkFile "out"

问题在于dbAction功能.我想自然地在这里访问数据库.由于它所做的动作基本上只是一个过滤器,我首先想到这样写:

dbAction = CL.mapMaybeM p
     where p :: (MonadIO m, MonadBaseControl IO (SqlPersist m)) => DataType -> m (Maybe DataType)
           p = lift $ putStrLn "foo" -- fine
           insert $ undefined -- type error!
           return undefined

我得到的具体错误是:

Could not deduce (m ~ b0 m0)
from the context (MonadIO m, MonadBaseControl IO (SqlPersist m))
  bound by …

我有一个管道管道处理一个长文件.我想每1000条记录为用户打印一份进度报告,所以我写了这样的:

-- | Every n records, perform the IO action.
-- Used for progress reports to the user.
progress :: (MonadIO m) => Int -> (Int -> i -> IO ()) -> Conduit i m i
progress n act = skipN n 1
   where
      skipN c t = do
         mv <- await
         case mv of
            Nothing -> return ()
            Just v ->
               if c <= 1
                  then do
                     liftIO $ act t v
                     yield v
                     skipN n (succ …

我试图理解管道概念的不同实现之间的差异.导管和管道之间的区别之一是它们如何将管道熔合在一起.管道有

(>+>) :: Monad m
      => Pipe l a b r0 m r1 -> Pipe Void b c r1 m r2 -> Pipe l a c r0 m r2

而管有

(>->) :: (Monad m, Proxy p)
      => (b' -> p a' a b' b m r) -> (c' -> p b' b c' c m r) -> c' -> p a' a c' c m r

如果我理解正确,使用管道,当两个管道的任何管道停止时,返回其结果而停止另一个.使用导管,如果左侧管道完成,其结果将向下游发送到右侧管道.

我想知道, …

为什么这么简单的代码不起作用？

import Network.HTTP.Conduit
import qualified Data.ByteString.Lazy as L

main :: IO ()
main = simpleHttp "http://www.dir.bg/" >>= L.putStr

它会导致以下错误:

TestConduit.exe:InternalIOException getAddrInfo:不存在(错误10093)

我正在编写应用程序,它将通过HTTP下载一些文件.到目前为止,我使用以下代码片段来下载页面正文:

import network.HTTP
simpleHTTP (getRequest "http://www.haskell.org/") >>= getResponseBody

它工作正常,但无法通过HTTPS协议建立连接.所以为了解决这个问题,我已经切换到HTTP-Conduit,现在我正在使用以下代码:

simpleHttp' :: Manager -> String -> IO (C.Response LBS.ByteString)
simpleHttp' manager url = do
     request <- parseUrl url
     runResourceT $ httpLbs request manager

它可以连接到HTTPS,但出现了令人沮丧的新问题.大约每五个连接失败,异常:

getpics.hs: FailedConnectionException "i.imgur.com" 80

我确信这是HTTP-Conduit问题,因为network.HTTP在同一组页面上工作正常(不包括https页面).

有没有人遇到过这样的问题并且知道解决方案或更好(并且简单,因为这是一个简单的任务,不应该只需要几行代码)替代Conduit库？

来自network-conduit的函数runTCPClient具有以下签名:

runTCPClient :: (MonadIO m, MonadBaseControl IO m)
             => ClientSettings m -> Application m -> m ()

MonadIO m 提供

liftIO :: IO a -> m a

并MonadBaseControl IO m提供

liftBase :: IO a -> m a

没有明显的区别.它们提供相同的功能吗？如果是,为什么类型签名中的重复？如果没有,有什么区别？

我有一个简单的任务 - 从文件中读取一堆行并对其中的每一行执行一些操作.除了第一个 - 这是一些可以忽略的标题.

所以我想我会试试管道.

printFile src = runResourceT $ CB.sourceFile src =$= 
    CT.decode CT.utf8 =$= CT.lines =$= CL.mapM_ putStrLn

凉.

所以现在我只想放弃第一行...而且似乎有一个功能 -

printFile src = runResourceT $ CB.sourceFile src =$= 
    CT.decode CT.utf8 =$= CT.lines =$= drop 1 =$= CL.mapM_ putStrLn

嗯 - 但现在我注意到drop有类型签名Sink a m ().有人向我建议我可以使用Monad实例来管道并使用drop来有效地删除一些元素 - 所以我尝试了这个:

drop' :: Int -> Pipe a a m ()
drop' n = do
  CL.drop n
  x <- await
  case x of 
    Just v -> yield v
    Nothing -> …

我正在尝试创建一个可以使用多个输入流的管道.我需要能够在没有特定顺序(例如,不交替)的情况下等待输入流中的一个或另一个,使得zip无用.这里没有任何平行或不确定的东西:我等待一个流或另一个流.我希望能够编写类似于以下的代码(分别在第一个或第二个输入流上的where awaitA和awaitBawait):

do
  _ <- awaitA
  x <- awaitA
  y <- awaitB
  yield (x,y)
  _ <- awaitB
  _ <- awaitB
  y' <- awaitB
  yield (x,y')

我所拥有的最好的解决方案是使内部monad成为另一个导管,例如

foo :: Sink i1 (ConduitM i2 o m) ()

然后允许

awaitA = await
awaitB = lift await

这主要是有效的.不幸的是,这似乎使得在外导管完全连接之前很难熔化到内导管.我尝试的第一件事是:

fuseInner :: Monad m =>
                Conduit i2' m i2 -> 
                Sink i1 (ConduitM i2 o m) () -> 
                Sink i1 (ConduitM i2' o m) ()
fuseInner x = transPipe (x =$=)

但这不起作用,至少在x …

使用相同的导管值执行多个操作是否安全？就像是

do
  let sink = sinkSocket sock

  something $$ sink
  somethingElse $$ sink

我记得在导管的早期版本中有一些脏的黑客使得这不安全.现状是什么？

(注意sinkSocket不要关闭套接字.)

我正在尝试构造一个Conduit接收为输入ByteStrings(每个块大小约1kb)并生成ByteString512kb块的输出连接.

这看起来应该很简单,但是我遇到了很多麻烦,我尝试过的大多数策略只能成功地将块分成更小的块,我没有成功连接更大的块.

我开始尝试isolate,然后takeExactlyE最终conduitVector,但无济于事.最终我决定这个:

import qualified Data.Conduit               as C
import qualified Data.Conduit.Combinators   as C
import qualified Data.ByteString            as B
import qualified Data.ByteString.Lazy       as BL

chunksOfAtLeast :: Monad m => Int -> C.Conduit B.ByteString m BL.ByteString
chunksOfAtLeast chunkSize = loop BL.empty chunkSize
  where 
    loop buffer n = do
      mchunk <- C.await
      case mchunk of 
        Nothing -> 
          -- Yield last remaining bytes
          when (n < chunkSize) (C.yield buffer)
        Just chunk -> …