如何使用 Conduit 组合器实现类似 takeWhile 的函数？

我正在尝试实现一个结合了以下想法的功能groupBy and takeWhile，以及在内部使用后者。具体来说，它将对连续返回的所有元素进行分组True将当前谓词作为列表，然后它将继续对下一个谓词执行相同的操作，依此类推：

   takeWhileGrouped :: (Monad m, MonoFoldable mono) =>
      ([Element mono -> Bool]) -> ConduitT (Element mono) [Element mono] m ()
    takeWhileGrouped preds = go preds
      where
        go (pred:nextPreds) = yield (goIter pred) >> go nextPreds
        goIter pred = takeWhile pred .| sinkList

这个实现很可能会遇到其他问题，但在这个阶段我收到一个编译错误，我不知道如何继续（为什么不能mono0被认同为mono？）；这是由于缺乏使用某些语言扩展造成的，还是还有其他问题？

    • Couldn't match type ‘Element mono0’ with ‘Element mono’
      Expected type: [Element mono -> Bool]
                     -> ConduitT (Element mono) [Element mono] m ()
        Actual type: [Element mono0 -> Bool]
                     -> ConduitT (Element mono0) [Element mono0] m ()
      NB: ‘Element’ is a non-injective type family
      The type variable ‘mono0’ is ambiguous
    • In the ambiguity check for ‘takeWhileGrouped’
      To defer the ambiguity check to use sites, enable AllowAmbiguousTypes
      In the type signature:
        takeWhileGrouped :: (Monad m, MonoFoldable mono) =>
                            ([Element mono -> Bool])
                            -> ConduitT (Element mono) [Element mono] m ()
    |
140 | takeWhileGrouped :: (Monad m, MonoFoldable mono) =>
    |                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^...

Update 1

启用AllowAmbiguousTypes确实会使错误消失，但我会注意到组合器库 https://hackage.haskell.org/package/conduit-1.3.1.1/docs/src/Data.Conduit.Combinators.html这似乎是不必要的。

现在真正的问题显现出来了：

    • Couldn't match type ‘ConduitT a0 c0 m0 [b0]’
                     with ‘[Element mono]’
      Expected type: ConduitT (Element mono) [Element mono] m ()
        Actual type: ConduitT
                       (Element seq0) (ConduitM a0 c0 m0 [b0]) m ()
    • In the expression: go preds
      In an equation for ‘takeWhileGrouped’:
          takeWhileGrouped preds
            = go preds
            where
                go (pred : nextPreds) = yield (goIter pred) >> go nextPreds
                goIter pred = takeWhile pred .| sinkList
    • Relevant bindings include
        preds :: [Element mono -> Bool]
          (bound at src/FDS/Data/Conduits.hs:143:18)
        takeWhileGrouped :: [Element mono -> Bool]
                            -> ConduitT (Element mono) [Element mono] m ()
          (bound at src/FDS/Data/Conduits.hs:143:1)
    |
143 | takeWhileGrouped preds = go preds
    |                          ^^^^^^^^


    • Couldn't match type ‘seq -> seq’ with ‘ConduitT a b m1 ()’
      Expected type: ConduitM a b m1 ()
        Actual type: seq -> seq
    • Probable cause: ‘takeWhile’ is applied to too few arguments
      In the first argument of ‘(.|)’, namely ‘takeWhile pred’
      In the expression: takeWhile pred .| sinkList
      In an equation for ‘goIter’:
          goIter pred = takeWhile pred .| sinkList
    • Relevant bindings include
        pred :: Element seq -> Bool
          (bound at src/FDS/Data/Conduits.hs:146:12)
        goIter :: (Element seq -> Bool) -> ConduitM a c m1 [b]
          (bound at src/FDS/Data/Conduits.hs:146:5)
    |
146 |     goIter pred = takeWhile pred .| sinkList
    |                   ^^^^^^^^^^^^^^

Update 2

我用错了takeWhile，现在使用CC.takeWhile来自 Conduit Cominators，我目前只剩下：

    • Couldn't match type ‘ConduitT
                             (Element mono) c0 m0 [Element mono]’
                     with ‘[Element mono]’
      Expected type: ConduitT (Element mono) [Element mono] m ()
        Actual type: ConduitT
                       (Element mono) (ConduitM (Element mono) c0 m0 [Element mono]) m ()
    • In the expression: go preds
      In an equation for ‘takeWhileGrouped’:
          takeWhileGrouped preds
            = go preds
            where
                go (pred : nextPreds) = yield (goIter pred) >> go nextPreds
                goIter pred = CM.takeWhile pred .| sinkList
    • Relevant bindings include
        preds :: [Element mono -> Bool]
          (bound at src/FDS/Data/Conduits.hs:144:18)
        takeWhileGrouped :: [Element mono -> Bool]
                            -> ConduitT (Element mono) [Element mono] m ()
          (bound at src/FDS/Data/Conduits.hs:144:1)
    |
144 | takeWhileGrouped preds = go preds
    |                          ^^^^^^^^

Update 3

有一些组合器 API 问题需要修复，但至少仍然存在一个问题：

takeWhileGrouped :: forall m mono. (Monad m, MonoFoldable mono) =>
  ([Element mono -> Bool]) -> ConduitT (Element mono) [Element mono] m ()
takeWhileGrouped preds = go preds
  where
    go (pred:nextPreds) = yieldM (goIter pred) >> go nextPreds
    go [] = yield []
    goIter :: (Element mono -> Bool) -> m ([Element mono])
    goIter pred = (CC.takeWhile pred) .| sinkList & runConduitRes

没想到我有一个()在输入中弹出takeWhile:

    • Couldn't match type ‘Element mono’ with ‘()’
      Expected type: () -> Bool
        Actual type: Element mono -> Bool
    • In the first argument of ‘CC.takeWhile’, namely ‘pred’
      In the first argument of ‘(.|)’, namely ‘(CC.takeWhile pred)’
      In the first argument of ‘(&)’, namely
        ‘(CC.takeWhile pred) .| sinkList’
    • Relevant bindings include
        pred :: Element mono -> Bool
          (bound at src/FDS/Data/Conduits.hs:148:12)
        goIter :: (Element mono -> Bool) -> m [Element mono]
          (bound at src/FDS/Data/Conduits.hs:148:5)
        preds :: [Element mono -> Bool]
          (bound at src/FDS/Data/Conduits.hs:144:18)
        takeWhileGrouped :: [Element mono -> Bool]
                            -> ConduitT (Element mono) [Element mono] m ()
          (bound at src/FDS/Data/Conduits.hs:144:1)
    |
148 |     goIter pred = (CC.takeWhile pred) .| sinkList & runConduitRes
    |                                 ^^^^

Update 4

修复了更多逻辑和类型错误后，添加内部类型注释确实有所帮助(CC.takeWhile curPred :: ConduitT (Element mono) (Element mono) m ())，我有一些可以编译的东西，但仍需要测试：

takeWhileGrouped :: forall m mono. (Monad m, MonoFoldable mono) =>
  ([Element mono -> Bool])
  -> ConduitT () (Element mono) m ()
  -> ConduitT (Element mono) [Element mono] m ()
takeWhileGrouped preds conIn = go preds
  where
    go (curPred:nextPreds) = yieldM (goIter curPred) >> go nextPreds
    go [] = yield []
    goIter :: (Element mono -> Bool) -> m ([Element mono])
    goIter curPred = conIn .| CC.takeWhile curPred .| sinkList & runConduit

我最终得出的结论是，如果我想实现这个组件，就组件构成而言我无法做到 https://stackoverflow.com/questions/58107971/why-is-there-an-unexpected-expected-type-of-in-this-conduit-composition-fu/58124544#58124544（融合）。

这是一个似乎有效的实现：

takeWhileGrouped :: forall m mono. Monad m =>
  [Element mono -> Bool] -> ConduitT (Element mono) [Element mono] m ()
takeWhileGrouped preds = start
  where
    start = await >>= maybe (return ()) (loop preds [])
    loop :: [(Element mono) -> Bool] -> [Element mono] -> (Element mono)
      -> ConduitT (Element mono) [Element mono] m ()
    loop [] _ _ = yield []
    loop curPreds@(predF:predRest) accum x =
        await >>= maybe (yield accumX) go
      where
        accumX = if predF x then x:accum else accum
        go y = if predF y then loop curPreds accumX y
               else yield accumX >> loop predRest [] y

这个设计受到我所知道的最相似的函数的影响（尽管来自一个已弃用的库），groupBy https://hackage.haskell.org/package/conduit-1.3.1.1/docs/src/Data.Conduit.List.html#groupBy.