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.