cereal-conduit 0.7.2.5 → 0.7.3
raw patch · 4 files changed
+96/−10 lines, 4 filesPVP ok
version bump matches the API change (PVP)
API changes (from Hackage documentation)
+ Data.Conduit.Cereal: conduitGet2 :: MonadThrow m => Get o -> Conduit ByteString m o
Files
- ChangeLog.md +4/−0
- Data/Conduit/Cereal.hs +59/−8
- Test/Main.hs +32/−1
- cereal-conduit.cabal +1/−1
ChangeLog.md view
@@ -1,3 +1,7 @@+## 0.7.3++* Provide `conduitGet2`+ ## 0.7.2.5 * Support cereal 0.5
Data/Conduit/Cereal.hs view
@@ -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
Test/Main.hs view
@@ -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
cereal-conduit.cabal view
@@ -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 <myles.maxfield@gmail.com>