diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,3 +1,7 @@ +## 0.7.3 + +* Provide `conduitGet2` + ## 0.7.2.5 * Support cereal 0.5 diff --git a/Data/Conduit/Cereal.hs b/Data/Conduit/Cereal.hs --- a/Data/Conduit/Cereal.hs +++ b/Data/Conduit/Cereal.hs @@ -13,12 +13,12 @@ module Data.Conduit.Cereal ( GetException , sinkGet , conduitGet + , conduitGet2 , sourcePut , conduitPut ) where import Control.Exception.Base -import Control.Monad.Trans.Class (MonadTrans, lift) import Control.Monad.Trans.Resource (MonadThrow, monadThrow) import qualified Data.ByteString as BS import qualified Data.ByteString.Lazy as LBS @@ -37,7 +37,8 @@ -- | Run a 'Get' repeatedly on the input stream, producing an output stream of whatever the 'Get' outputs. conduitGet :: MonadThrow m => Get o -> C.Conduit BS.ByteString m o conduitGet = mkConduitGet errorHandler - where errorHandler msg = pipeError $ GetException msg + where errorHandler msg = monadThrow $ GetException msg +{-# DEPRECATED conduitGet "Please switch to conduitGet2, see comment on that function" #-} -- | Convert a 'Get' into a 'Sink'. The 'Get' will be streamed bytes until it returns 'Done' or 'Fail'. -- @@ -45,14 +46,11 @@ -- If the 'Get' fails due to deserialization error or early termination of the input stream it raise an error. sinkGet :: MonadThrow m => Get r -> C.Consumer BS.ByteString m r sinkGet = mkSinkGet errorHandler terminationHandler - where errorHandler msg = pipeError $ GetException msg + where errorHandler msg = monadThrow $ GetException msg terminationHandler f = case f BS.empty of - Fail msg _ -> pipeError $ GetException msg + Fail msg _ -> monadThrow $ GetException msg Done r lo -> C.leftover lo >> return r - Partial _ -> pipeError $ GetException "Failed reading: Internal error: unexpected Partial." - -pipeError :: (MonadThrow m, MonadTrans t, Exception e) => e -> t m a -pipeError e = lift $ monadThrow e + Partial _ -> monadThrow $ GetException "Failed reading: Internal error: unexpected Partial." -- | Convert a 'Put' into a 'Source'. Runs in constant memory. sourcePut :: Monad m => Put -> C.Producer m BS.ByteString @@ -61,3 +59,56 @@ -- | Run a 'Putter' repeatedly on the input stream, producing a concatenated 'ByteString' stream. conduitPut :: Monad m => Putter a -> C.Conduit a m BS.ByteString conduitPut p = CL.map $ runPut . p + +-- | Reapply @Get o@ to a stream of bytes as long as more data is available, +-- and yielding each new value downstream. This has a few differences from +-- @conduitGet@: +-- +-- * If there is a parse failure, the bytes consumed so far by this will not be +-- returned as leftovers. The reason for this is that the only way to guarantee +-- the leftovers will be returned correctly is to hold onto all consumed +-- @ByteString@s, which leads to non-constant memory usage. +-- +-- * This function will properly terminate a @Get@ function at end of stream, +-- see https://github.com/snoyberg/conduit/issues/246. +-- +-- * @conduitGet@ will pass empty @ByteString@s from the stream directly to +-- cereal, which will trigger cereal to think that the stream has been closed. +-- This breaks the normal abstraction in conduit of ignoring how data is +-- chunked. In @conduitGet2@, all empty @ByteString@s are filtered out and not +-- passed to cereal. +-- +-- * After @conduitGet2@ successfully returns, we are guaranteed that there is +-- no data left to be consumed in the stream. +-- +-- @since 0.7.3 +conduitGet2 :: MonadThrow m => Get o -> C.Conduit BS.ByteString m o +conduitGet2 get = + awaitNE >>= start + where + -- Get the next chunk of data, only returning an empty ByteString at the + -- end of the stream. + awaitNE = + loop + where + loop = C.await >>= maybe (return BS.empty) check + check bs + | BS.null bs = loop + | otherwise = return bs + + start bs + | BS.null bs = return () + | otherwise = result (runGetPartial get bs) + + result (Fail msg _) = monadThrow (GetException msg) + -- This will feed an empty ByteString into f at end of stream, which is how + -- we indicate to cereal that there is no data left. If we wanted to be + -- more pedantic, we could ensure that cereal only ever consumes a single + -- ByteString to avoid a loop, but that is the contract that cereal is + -- giving us anyway. + result (Partial f) = awaitNE >>= result . f + result (Done x rest) = do + C.yield x + if BS.null rest + then awaitNE >>= start + else start rest diff --git a/Test/Main.hs b/Test/Main.hs --- a/Test/Main.hs +++ b/Test/Main.hs @@ -1,8 +1,9 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE Rank2Types #-} +{-# LANGUAGE OverloadedStrings #-} -import Control.Applicative (many) +import Control.Applicative (many, optional) import Control.Exception.Base import Control.Monad.Identity import Control.Monad.Trans.Resource @@ -213,6 +214,35 @@ recurse where recurse = C.await >>= maybe (return ()) (\ x -> C.yield (Left x) >> recurse) +-- see https://github.com/snoyberg/conduit/issues/246 +conduittest18 :: Test +conduittest18 = TestCase $ assertEqual "Deals with Get that consumes everything" + (Right ["hello"]) + (runIdentity $ runExceptionT + $ (C.yield "hello" + C.$= conduitGet2 slurp + C.$$ CL.consume)) + +slurp :: Get BS.ByteString +slurp = + loop id + where + loop front = do + x <- remaining + if x == 0 + then do + mbs <- optional $ lookAhead $ getBytes 1 + case mbs of + Nothing -> do + let bs = BS.concat $ front [] + if BS.null bs + then fail "no bytes remaining" + else return bs + Just _ -> loop front + else do + bs <- getBytes $ fromIntegral x + loop (front . (bs:)) + puttest1 :: Test puttest1 = TestCase (assertEqual "conduitPut works" [BS.pack [0, 1]] @@ -256,6 +286,7 @@ , conduittest15 , conduittest16 , conduittest17 + , conduittest18 ] puttests = TestList [ puttest1 diff --git a/cereal-conduit.cabal b/cereal-conduit.cabal --- a/cereal-conduit.cabal +++ b/cereal-conduit.cabal @@ -1,5 +1,5 @@ name: cereal-conduit -version: 0.7.2.5 +version: 0.7.3 license: BSD3 license-file: LICENSE author: Myles C. Maxfield