[SOLVED] Conduit parser is interrupted prematurely

Conduit parser is interrupted prematurely

I'm working on a program where I need to parse data from a USB connection and write it to a circular buffer.

The issue I keep running into is that the parser is not consuming the input. Do I need to use another operator to connect the parser conduit to the stream of ByteStrings?

If I remove the two 'filters' dropC and dropWhileC, the parser consumes some of the input but fails, because the first few chunks of data are junk.

Clearly my mental model of how the library works is off somehow (I've never used it before).

I'd also like to know if empty strings are treated as EOF markers by conduitParser, as with attoparsecs own parseWith.

-- |
sourcePort :: SerialPort -> Producer IO ByteString
sourcePort port = repeatMC (recv port 16)


-- |
parseSerialStream :: (Vector v (Int, Int)) => RingBuffer v (Int, Int) -> SerialPort -> IO () -- ConduitM a c IO ()
parseSerialStream buffer port = sourcePort port
                             .| dropC 4
                             .| dropWhileC B.null
                             .| conduitParser (parseMeasurement <* endOfLine)
                             $$ mapM_C (\item -> print item >> RB.append (snd item) buffer)

Solution

@danidiaz has explained in the comments that dropC et.al. do not yield any values, which is why you must use monadic composition instead of piping (.|).

To anyone who stumbles upon this question, see this answer for details.

This code works as expected:

parseSerialStream :: RingBuffer VU.Vector (Int, Int) -> SerialPort -> IO () -- ConduitM a c IO ()
parseSerialStream buffer port = sourcePort port
                             .| (dropC 4
                             >> dropWhileC B.null
                             >> conduitParser (parseMeasurement <* endOfLine))
                             $$ mapM_C (\item -> RB.append (snd item) buffer)

I liked how the pipes looked though. Oh well...