如何将 Sink 变成 Conduit？

Question

我正在尝试使用attoparsec 解析器编写Conduit。具体来说，给定parseOne :: Parser T，我想构造一个Conduit ByteString m T，它重复地将解析器应用于输入并流式传输结果。

attoparsec-conduit 提供sinkParser 将Parser 转换为Sink，但是我如何将这个Sink 转换为Conduit？我正在寻找的是这样的功能：

conduitSink :: (Resource m) => Sink a m b -> Conduit a m b

它反复将数据输入Sink，并生成每个结果。看起来它可以很容易地编写为手动循环，但我想知道是否有更好的方法。

管道库中缺少这个看似显而易见的功能让我觉得我可能做错了什么；有没有更好的方法来做到这一点？用例是将原始字节转换为基于消息的网络协议的解析形式，以供管道的后续阶段处理。感谢blaze-builder-conduit，我已经有了相反的方向（即Conduit T m ByteString），所以这似乎是构建事物的最自然的方式。

Answer 1

为此您需要使用SequencedSink 系统；它使用 sink 和 tracked state 从 sink producer 的重复应用中产生一个管道。

您创建的接收器已针对增量解析一个值进行了优化，这将是管道序列结束时的结果。

但是，由于您希望它成为管道管道的一部分，并且传入 ByteString 的每个块可能会或可能不会与您的解析器匹配一次或多次，因此您需要注意获得更细粒度的控制解析过程，在接收器的每个应用程序之间传递不完整解析的状态。

例如，假设您的解析器解析 [--] 或 [----] 等，而 T 是 Int 表示解析的破折号数量，您需要跟踪解析器的状态，如下所示：

Input chunk    Sink result - Data.Conduit.SequencedSinkResponse
[--][---]      Emit Nothing [2, 3]
[---][---      Emit (Just #func) [3]
---------      Emit (Just #func) []
]              Emit Nothing [12]
               Stop

在这种情况下，我使用Maybe (ByteString -> Data.Attoparsec.ByteString.Result) 作为传递状态；根据具体情况，不同的数据类型可能更合适。

需要这种显式的流处理来维持管道的管道性质；让解析器管道成为“瓶颈”，总是等待足够的数据来分块地满足解析器，这将是一个主要的性能下降。

使用可用的ResourceT monad 接口，所需接收器的实现应该相当简单。

编辑：简单地在循环中应用你的接收器确实是最简单的解决方案，但如果你的解析器解析通常结束于边界的短 sn-ps，它的性能特征会略有不同字节块。