连续Either值的Conduit转为嵌套Conduit的可行性与实现方法

1. 类型ConduitT () a m (ConduitT () b m ())是否正确？ #

是的，这个类型完全匹配你的需求：

• 语义上，它明确表示先流式输出所有a（对应原始流的Left值），输出完毕后返回一个新的Conduit，该Conduit会流式输出所有b（对应原始流的Right值）。
• 类型系统强制了处理顺序：消费者必须先处理完所有a，才能获取到处理b的Conduit，完美契合API文档中“先连续Left，再连续Right”的连续性特性，从根源上避免误处理混合的Left/Right值。
• 支持实时流式处理：a值一到达就可以被消费，无需等待所有a全部生成，满足你“值到达时立即处理”的要求。

2. 如何从原始Conduit转换到该嵌套类型？ #

核心思路是利用SealedConduitT保存部分消费的流状态，遍历原始流输出Left值，直到遇到第一个Right，再将剩余流封装为返回的内层Conduit。具体实现如下：

import Data.Conduit
import qualified Data.Conduit.List as CL

-- 将原始Either流转换为分阶段嵌套Conduit
splitEitherConduit :: Monad m => ConduitT () (Either a b) m () -> ConduitT () a m (ConduitT () b m ())
splitEitherConduit originalSrc = do
  -- 将原始流封装为SealedConduitT，用于保存消费状态
  sealedSrc <- lift $ sealConduitT originalSrc
  processLefts sealedSrc
  where
    -- 递归处理Left值，直到遇到Right或流结束
    processLefts sealed = do
      -- 从SealedConduitT中取出一个元素，并保存剩余流的状态
      (mbElem, remainingSealed) <- lift $ unsealConduitT sealed $ \conduit -> do
        nextElem <- await
        return (nextElem, sealConduitT conduit)
      
      case mbElem of
        -- 流结束：返回空的b流
        Nothing -> return (return ())
        
        -- 遇到Left值：输出a，继续处理剩余流
        Just (Left a) -> do
          yield a
          processLefts remainingSealed
        
        -- 遇到Right值：构造内层Conduit，先输出当前b，再处理剩余流的所有Right值
        Just (Right b) -> do
          let bStream = do
                yield b
                -- 处理剩余流，只输出Right值（根据API约定，后续不会有Left）
                unsealConduitT remainingSealed $ \conduit ->
                  conduit .| CL.mapM_ (\case
                                        Right x -> yield x
                                        Left _ -> error "API违反约定：Left出现在Right之后")
          return bStream

代码说明： #

• SealedConduitT：用于封装Conduit，避免组合时丢失消费状态，确保我们可以安全暂停流消费、保存剩余状态，后续恢复消费。
• 实时处理：外层Conduit一收到Left a就yield，消费者可以立即处理，无需等待所有a生成。
• 错误处理：根据API文档，Right之后不会出现Left，因此遇到这种情况直接抛出错误，你也可以根据需求替换为自定义错误处理逻辑（比如返回Maybe或使用ExceptT）。

替代方案？ #

如果不需要强制类型化的处理顺序，原始ConduitT () (Either a b) m ()当然可以使用，但它无法利用类型系统保证API的连续性特性。其他方案比如m (ConduitT () a m (), ConduitT () b m ())需要先将所有a加载到内存中才能获取两个流，不符合实时处理的要求，因此你的初始选择是最优解。

内容的提问来源于stack exchange，提问作者Jules