流式传输字节串作为 WAI HTTP 服务器响应正文

Question

我有一个值body :: BS.ByteString (ResourceT IO) ()，来自基于BS.readFile 的函数。我想将该值作为来自Wai Application 的响应正文进行流式传输。有一个帮助器streamingResponse，它采用Stream (Of ByteString) IO r 类型的值。我可以通过使用BS.toChunks 将我的BS.ByteString (ResourceT IO) () 转换为Stream (Of ByteString) (ResourceT IO) ()，但它包含一个额外的ResourceT monad 层。将body 传递给streamingResponse 给了我：

Couldn't match type ‘ResourceT IO’ with ‘IO’
  Expected type: Stream (Of ByteString) IO ()
    Actual type: Stream (Of ByteString) (ResourceT IO) ()

我尝试了各种方法，例如在 runResourceT 中包装东西、绑定和提升值等，但我真的不知道如何进行。如果需要额外的上下文，Here 是整个项目中的行。

更新0

hoist runResourceT body 似乎在进行类型检查。有人还向我推荐了 Haskell Pipes thread，这可能是一个非常相关的问题，并且可能暗示了解决方案。

Answer 1

如果我们想允许存在于ResourceT 中的Streams，我们可以不使用streaming-wai 中的函数（仅适用于基于IO 的Streams），而是构建在像responseStream 这样的函数来自network-wai：

import           Control.Monad.Trans.Resource
import           Network.Wai                     
import           Streaming                   
import qualified Streaming.Prelude               as S
import           Data.ByteString.Builder (byteString, Builder)

streamingResponseR :: Stream (Of ByteString) (ResourceT IO) r
                   -> Status
                   -> ResponseHeaders
                   -> Response
streamingResponseR stream status headers =
    responseStream status headers streamingBody
    where
    streamingBody writeBuilder flush =
        let writer a =
                do liftIO (writeBuilder (byteString a))
                    -- flushes for every produced bytestring, perhaps not optimal
                   liftIO flush
         in runResourceT $ void $ S.effects $ S.for stream writer

streamingBody 的类型为StreamingBody，它实际上是函数(Builder -> IO ()) -> IO () -> IO () 的类型同义词，该函数将写入回调和刷新回调作为参数，并使用它们使用某些数据源写入响应在适用范围。 （请注意，这些回调由 WAI 提供，而不是由用户提供。）

在我们的例子中，数据源是位于ResourceT 中的Stream。我们需要使用liftIO 解除写入和刷新回调（位于IO），还记得调用runResourceT 以在最后返回一个普通的IO 操作。

---

如果我们想仅在发出的字节串的累积长度达到某个限制后刷新响应怎么办？

我们需要一个函数（此处未实现）来在每次达到限制时创建一个部门：

breaks' :: Monad m 
        => Int 
        -> Stream (Of ByteString) m r 
        -> Stream (Stream (Of ByteString) m) m r
breaks' breakSize = undefined

然后我们可以在写入流之前使用intercalates 在每个组之间插入刷新操作：

streamingBodyFrom :: Stream (Of ByteString) (ResourceT IO) () 
                  -> Int 
                  -> StreamingBody
streamingBodyFrom stream breakSize writeBuilder flush =
    let writer a = liftIO (writeBuilder (byteString a))
        flusher = liftIO flush
        broken = breaks' breakSize stream
     in runResourceT . S.mapM_ writer . S.intercalates flusher $ broken

Answer 2

用withFile + hSetBinaryMode +Data.ByteString.Streaming.fromHandle 代替readFile 就够了吗？

fromHandle 产生一个ByteString IO ()，可以转换为streamingResponse 或streamingBody 可以接受的Stream (Of ByteString) IO ()。

存在将withFile 包围操作放在何处的问题。根据WAI documentation，您可以用它包装您的Application-building 函数的结果：

请注意，从 WAI 3.0 开始，这种类型的结构是延续的 传递样式以允许适当的安全资源处理。这是 过去通过其他方式（例如 ResourceT）处理过。

[...]

为了以异常安全的方式分配资源，您可以 在对 responseStream 的调用之外使用括号模式。

---

注意：streaming-bytestring 的文档说fromHandle 在到达 EOF 时会自动关闭句柄。查看实现，that doesn't seem to be the case。您需要withFile 才能正确关闭手柄。