如何将partialProcessConduitInMemory改为使用Conduit Sink替代DList？

目标函数签名 #

我希望实现一个签名如下的函数：

partialProcessConduit :: forall m a b r. Monad m 
  => (a -> (b, Maybe (ConduitT () a m ()) )) 
  -> ConduitT b Void m r 
  -> ConduitT () a m () 
  -> m (r, ConduitT () a m ())
partialProcessConduit splitFunc consumingConduit sourceConduit

核心逻辑 #

• 从sourceConduit中反复取出类型为a的值；
• 对该值应用splitFunc函数；
• 将splitFunc返回的b值传入consumingConduit；
• 若splitFunc返回的二元组第二个元素为Just（某个conduit），则：
  • 关闭consumingConduit并获取结果r；
  • 返回包含sourceConduit剩余部分的conduit，并将Just中的conduit前置。

现有实现（内存收集版） #

我已实现类似函数partialProcessConduitInMemory（命名欠佳），代码如下：

{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE ScopedTypeVariables #-}

import Conduit (ConduitT, SealedConduitT, unsealConduitT, ($$+), await)
import Data.Void (Void)
import qualified Control.Arrow as Arrow
import Data.DList (DList)

partialProcessConduitInMemory :: forall m a b r. Monad m 
  => (a -> (b, Maybe (ConduitT () a m ()))) 
  -> (DList b -> r) 
  -> ConduitT () a m () 
  -> m (r, ConduitT () a m ())
partialProcessConduitInMemory splitFunc collapseDList sourceConduit = do
  (sc :: SealedConduitT () a m (), (result :: r, leftOver :: ConduitT () a m ())) <- x
  pure (result, leftOver >> unsealConduitT sc)
  where
    x :: m (SealedConduitT () a m (), (r, ConduitT () a m ()))
    x = sourceConduit $$+ g
    g :: ConduitT a Void m (r, ConduitT () a m ())
    g = Arrow.first collapseDList <$> go mempty
    go :: DList b -> ConduitT a Void m (DList b, ConduitT () a m ())
    go blockList = await >>= \case
      Nothing -> pure (blockList, pure ())
      Just block -> case splitFunc block of
        (transformedBlock, Nothing) -> go $ blockList <> pure transformedBlock
        (transformedBlock, Just leftOver) -> pure (blockList <> pure transformedBlock, leftOver)

问题与需求 #

该函数已接近需求，但第二个参数是通过DList收集元素后处理，而非用于消费元素的conduit sink。我希望改为用conduit sink消费sourceConduit前半部分，而非先收集所有元素到列表中。请问是否可行？若可行需做哪些调整？我曾尝试在go循环中将元素传入sink再通过runConduit获取结果r，但类型匹配出现问题，恳请帮助。

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