我正在使用streaming-utils streaming-utils来传输HTTP响应主体。我想跟踪类似于bytestring-progress如何允许懒惰的ByteString
s的进度。我怀疑像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 ()
:
ByteString
s的传入长度并将更新打印到控制台。我们可以在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
这将打印ByteString
s以及累计长度:
main :: IO ()
main = S.mapM_ B.putStr . track $ S.each ["aa","bb","c"]