C.linesC works unreliably on streaming input

[jstranik]

bug:: ConduitT () Void ResIO ()
bug = do
  C.yieldMany ["a", "b", "c", "d", "e\nf\n"]
  C..|  C.peekForeverE do
          C.lineC do
            C.await >>= mapM_ \s -> do
              liftIO.putStrLn $ "-" <> s
  C..|  C.sinkNull

{-
Bug -- above code prints:
  - a
  - f

Expected:
  - abcde
  - f


Possible fix in takeExactlyUntilE:
takeExactlyUntilE2 ::forall m seq o r. (Monad m, Seq.IsSequence seq)
                  => (Element seq -> Bool)
                  -> ConduitT seq o m r
                  -> ConduitT seq o m r
takeExactlyUntilE2 f inner =
    loop emptySeq  .| do
        x <- inner
        sinkNull
        return x
  where
    emptySeq = Seq.fromList []
    loop x = await >>= mapM_ (go x)
    go r t =
        if onull y
            -- then C.yield x >> loop
            then loop x -- wait until f fires
            else do
                unless (onull x) $ C.yield x
                let y' = Seq.drop 1 y
                unless (onull y') $ leftover y'
      where
        (x', y) = Seq.break f t
        x = r <> x'

line2 :: (Monad m, Seq.IsSequence seq, Element seq ~ Char)
     => ConduitT seq o m r
     -> ConduitT seq o m r
line2 = takeExactlyUntilE2 (== '\n')

[snoyberg]

On my phone, but it looks like the example is buggy. The usage of await will ensure that only the first chunk in the line is printed, which corresponds with the behavior reported.