How to implement a takeWhile-like function using Conduit combinators?

Question

I'm trying to implement a function that combines the ideas of groupBy and takeWhile, as well as uses the latter internally. Specifically, it will group all elements that consecutively return True on the current predicate as a list, then it will proceed to do the same with the next predicate, and so on:

   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

It's entirely likely this implementation suffers other issues, but at this stage I'm getting a compilation error that I'm not sure how to proceed with (why can't mono0 be identified with mono?); is this due to the lack of using some language extension, or is there another issue?

    • 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

Enabling AllowAmbiguousTypes does make the error go away, but I'll note that in the combinators library this appears to be unnecessary.

Now the real issues show themselves:

    • 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

I was using the wrong takeWhile, now using CC.takeWhile from Conduit Cominators, I'm currently left with:

    • 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

Had a few combinator API issues to fix, but at least one still remains:

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

Unexpectedly, I have a () popping up in the input to 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

After fixing some more logic and type errors, which was really aided by adding an internal type annotation (CC.takeWhile curPred :: ConduitT (Element mono) (Element mono) m ()), I have something that compiles, but still needs testing:

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

bbarker bbarker · Accepted Answer · 2019-09-26T20:52:33

I came to the eventual conclusion that, if I wanted to implement this component, I couldn't do it in terms of component composition (fusion).

Here's an implementation that seems to work:

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

This design was influenced by the most similar function I knew of (though from a deprecated library), groupBy.

How to implement a takeWhile-like function using Conduit combinators?

1 Answers