如何将Either类型函数转为支持快速失败的ConduitT管道？

你可以通过手动递归循环或者利用Conduit的组合子来实现需求，核心是流式处理输入，遇到Left b时立即终止并返回错误值，处理完所有输入则返回Nothing。

手动递归实现（直观版） #

这个版本直接通过await和递归循环控制流的处理逻辑，清晰易懂：

import Conduit

eitherConduit :: Monad m => (a -> Either b c) -> ConduitT a c m (Maybe b)
eitherConduit f = loop
  where
    loop = do
      -- 获取下一个输入元素
      nextInput <- await
      case nextInput of
        -- 流结束，返回Nothing表示无错误
        Nothing -> pure Nothing
        Just a -> case f a of
          -- 处理成功，输出c并继续循环
          Right c -> yield c >> loop
          -- 遇到错误，立即返回Just b终止管道
          Left b -> pure (Just b)

组合子实现（简洁版） #

利用mapAccumM组合子携带错误状态，避免手动递归，更符合Conduit的组合式编程风格：

import Conduit

eitherConduit' :: Monad m => (a -> Either b c) -> ConduitT a c m (Maybe b)
eitherConduit' f = snd <$> mapAccumM step Nothing
  where
    step errState input = case errState of
      -- 已遇到错误，跳过后续所有输入，不产生输出
      Just _ -> pure (errState, Nothing)
      Nothing -> case f input of
        -- 处理成功，输出c，保持无错误状态
        Right c -> pure (Nothing, Just c)
        -- 遇到错误，更新状态为Just b，不产生输出
        Left b -> pure (Just b, Nothing)

逻辑说明 #

• 两种实现都严格遵循需求：流式处理输入，每处理一个a就输出对应的c，一旦遇到Left b立即终止管道，不再处理后续输入。
• 和列表版本sequence . map f的区别：列表版本是一次性处理所有元素，而Conduit版本是流式处理，无需等待所有输入加载完成，适合处理大型甚至无限流。
• 最终返回值Maybe b：Just b表示处理过程中遇到错误并终止，Nothing表示所有输入处理完成且无错误。

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