如何通过流式ByteString跟踪进度？

Posted 2021-04-07

我正在使用streaming-utils streaming-utils来传输HTTP响应主体。我想跟踪类似于bytestring-progress如何允许懒惰的ByteStrings的进度。我怀疑像toChunks这样的东西是必要的，然后减少一些累积的字节读取并返回未经修改的原始流。但我无法弄清楚，streaming文档是非常无益的，大多数充满了与替代库的宏伟比较。

到目前为止，这是我最努力的代码。它还没有包括计数，只是尝试在流过（并且不编译）时打印块的大小。

download :: ByteString -> FilePath -> IO ()
download i file = do
  req <- parseRequest . C.unpack $ i
  m <- newHttpClientManager
  runResourceT $ do
    resp <- http req m
    lift . traceIO $ "downloading " <> file
    let body = SBS.fromChunks $ mapsM step $ SBS.toChunks $ responseBody resp
    SBS.writeFile file body

step bs = do
  traceIO $ "got " <> show (C.length bs) <> " bytes"
  return bs

答案

我们想要的是以两种方式穿越Stream (Of ByteString) IO ()：

• 一个累积ByteStrings的传入长度并将更新打印到控制台。
• 将流写入文件的人。

我们可以在copy函数的帮助下做到这一点，它具有以下类型：

copy :: Monad m => Stream (Of a) m r -> Stream (Of a) (Stream (Of a) m) r

copy获取流并将其复制到两个不同的monadic层中，其中原始流的每个元素由新分离的流的两个层发出。

（请注意，我们正在更改基础monad，而不是functor。将functor更改为另一个Stream的做法是将delimit groups放在一个流中，我们对此不感兴趣。）

以下函数获取流，复制它，使用S.scan，prints them累积传入字符串的长度，并返回另一个仍可使用的流，例如将其写入文件：

{-# LANGUAGE OverloadedStrings #-}
import Streaming
import qualified Streaming.Prelude as S
import qualified Data.ByteString as B

track :: Stream (Of B.ByteString) IO r -> Stream (Of B.ByteString) IO r
track stream =
      S.mapM_ (liftIO . print) -- brings us back to the base monad, here another stream
    . S.scan (s b -> s + B.length b) (0::Int) id
    $ S.copy stream

这将打印ByteStrings以及累计长度：

main :: IO ()
main = S.mapM_ B.putStr . track $ S.each ["aa","bb","c"]