使用组合器将导管重新组装成更大的块

Question

我正在尝试构造一个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 -> do
          -- Yield when the buffer has been filled and start over
          let buffer' = buffer <> BL.fromStrict chunk
              l       = B.length chunk
          if n <= l
          then C.yield buffer' >> loop BL.empty chunkSize
          else loop buffer' (n - l)

_{PS我决定不为此功能拆分更大的块,但这只是一个方便的简化.}

然而,这似乎与分块非常冗长鉴于处理的所有导管功能^{[ 1,2,3,4 ]}.请帮忙!使用组合器肯定有更好的方法可以做到这一点,但我缺少一些直觉!

_{PPS是否可以像我一样使用延迟字节串作为缓冲区？我有点不清楚bytestring的内部表示以及这是否会有所帮助,特别是因为我正在使用BL.length它,我猜可能会评估thunk？}

---

结论

为了详细阐述迈克尔的答案和评论,我最终得到了这个渠道:

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

-- | "Strict" rechunk of a chunked conduit
chunksOfE' :: (MonadBase base m, PrimMonad base) 
         => Int 
         -> C.Conduit ByteString m ByteString
chunksOfE' chunkSize = C.vectorBuilder chunkSize C.mapM_E =$= C.map fromByteVector

我的理解是,vectorBuilder将支付早期连接较小块的成本,将聚合块作为严格的字节串.

据我所知,当聚合的块非常大和/或馈送到像网络套接字这样的自然流接口时,可能需要产生延迟字节串块(即"分块块")的替代实现.这是我对"lazy bytestring"版本的最佳尝试:

import qualified Data.Sequences.Lazy        as SL
import qualified Data.Sequences             as S
import qualified Data.Conduit.List          as CL

-- | "Lazy" rechunk of a chunked conduit
chunksOfE :: (Monad m, SL.LazySequence lazy strict)
          => S.Index lazy
          -> C.Conduit strict m lazy
chunksOfE chunkSize = CL.sequence C.sinkLazy =$= C.takeE chunkSize

Answer 1

这个怎么样？

{-# LANGUAGE NoImplicitPrelude #-}
{-# LANGUAGE OverloadedStrings #-}
import ClassyPrelude.Conduit

chunksOfAtLeast :: Monad m => Int -> Conduit ByteString m LByteString
chunksOfAtLeast chunkSize =
    loop
  where
    loop = do
        lbs <- takeCE chunkSize =$= sinkLazy
        unless (null lbs) $ do
            yield lbs
            loop

main :: IO ()
main =
    yieldMany ["hello", "there", "world!"]
        $$ chunksOfAtLeast 3
        =$ mapM_C print

根据您的目标,您可以采取许多其他方法.如果你想拥有一个严格的缓冲区,那么使用vectorBuilder的blaze-builder会很有意义.但是这保持了你已经拥有的相同类型签名.