使用 Conduit 分块数据答案

【问题标题】：Chunk data with Conduit使用 Conduit 分块数据
【发布时间】：2018-03-27 08:46:17
【问题描述】：

这是一个 conduit 组合子的示例，当从上游接收到完整消息时，它应该在下游 yield：

import qualified Data.ByteString as BS
import Data.Conduit
import Data.Conduit.Combinators
import Data.Conduit.Network

message :: Monad m => ConduitT BS.ByteString BS.ByteString m ()
message = loop
  where
    loop = await >>= maybe (return ()) go
    go x = if (BS.isSuffixOf "|" x)
        then yield (BS.init x) >> loop
        else leftover x

服务器代码本身如下所示：

main :: IO ()
main = do
  runTCPServer (serverSettings 5000 "!4") $ \ appData -> runConduit $
    (appSource appData)
    .| message
    .| (appSink appData)

由于某种原因telnet 127.0.0.1 5000 在发送任何消息后断开连接：

telnet 127.0.0.1 5000
Trying 127.0.0.1...
Connected to 127.0.0.1.
Escape character is '^]'.
123|
Connection closed by foreign host.

请指教，我在这里做错了什么？

更新

更重要的是我在这里尝试做的是等待完成信号|，然后yield 下游的完整消息。这是message组合器的演变：

message :: Monad m => ConduitT BS.ByteString BS.ByteString m ()
message = do
  minput <- await
  case minput of
    Nothing    -> return ()
    Just input -> do
      case BS.breakSubstring "|" input of
        ("", "")  -> return ()
        ("", "|") -> return ()
        ("", xs)  -> leftover $ BS.tail xs
        (x, "")   -> leftover x -- problem is in this leftover
        (x, xs)   -> do
          yield x
          leftover $ BS.tail xs
      message

我的想法是，如果上游没有任何东西来自组合器，则必须等到有东西出现，这样它才能向下游发送完整的消息。但在上述message 组合器中的leftover 调用上，conduit 开始在 CPU 上旋转很多。

【问题讨论】：

标签： haskell conduit network-conduit

【解决方案1】：

最后发现在基本情况下需要await 而不是leftover。以下是message 组合器的工作方式：

message :: Monad m => ConduitT BS.ByteString BS.ByteString m ()
message = do
  minput <- await
  case minput of
    Nothing    -> return ()
    Just input -> process input >> message
  where
    process input =
      case BS.breakSubstring "|" input of
        ("", "")  -> return ()
        ("", "|") -> return ()
        ("", xs)  -> leftover $ BS.tail xs
        (x, "")   -> do
          minput <- await
          case minput of
            Nothing -> return ()
            Just newInput -> process $ BS.concat [x, newInput]
        (x, xs)   -> do
          yield x
          leftover $ BS.tail xs

可能可以清理一些样板，但它可以工作。

【讨论】：

【解决方案2】：

在go 中打印x 以进行调试。

  ...
  go x = do
    liftIO (Prelude.print x)
    if ...

socket 接收到一个以\r\n 结尾的字节串，所以你转到else 分支，它会终止会话。

【讨论】：

确实\r\n 可能会妨碍您，但最终它以某种方式连接到leftover 调用。我更新了原始问题。