[Haskell-cafe] conduit: consume part of input and push all other input downstream

[Qingbo Liu]

Dear Cafe,

I am using conduit to parse a file in a streaming fashion. In the middle of the conduit pipeline, I have a conduit that parses a value, returns the value, and then passes all other input downstream. Is there any way to express this? I looked up functions provided by the conduit and it seems that there is no such a function that provides such semantics.

My conduit function looks as follows:

-- | Run the given `Get` only once and push unconsumed bytes from upstream to
-- downstream
conduitParseOnce :: (MonadThrow m, MonadResource m) =>
                        Get a -> ConduitT BS.ByteString BS.ByteString m a
conduitParseOnce g = go0
    where
      go0 = do x <- await
               case x of
                 Nothing -> undefined -- throwM SomeException
                 Just bs -> go (runGetChunk g Nothing bs)

      go (Fail msg bs) = throwM (DecodeError bs msg "conduitParseOnce")
      go (Partial f)   = await >>= maybe (go $ f mempty) (go . f)
      go (Done v bs)   = leftover bs >> return v

The downstream of this conduit will take all remaining inputs and parses them. The issue with this implementation is that the downstream will only be able to get the first chunk of data.

Best Regards,
Qingbo Liu
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <http://mail.haskell.org/pipermail/haskell-cafe/attachments/20190108/5ce4cce5/attachment.html>