cereal-conduit 0.5 → 0.6

raw patch · 4 files changed

+194/−142 lines, 4 filesdep +transformersdep −voiddep ~conduitPVP ok

version bump matches the API change (PVP)

Dependencies added: transformers

Dependencies removed: void

Dependency ranges changed: conduit

API changes (from Hackage documentation)

Files

• Data/Conduit/Cereal.hs +59/−55
• Data/Conduit/Cereal/Internal.hs +64/−62
• Test/Main.hs +66/−20
• cereal-conduit.cabal +5/−5

@@ -1,55 +1,59 @@-{-# LANGUAGE FlexibleContexts #-}
-{-# LANGUAGE DeriveDataTypeable #-}
-
--- | Turn a 'Get' into a 'Sink' and a 'Put' into a 'Source'
--- These functions are built upno the Data.Conduit.Cereal.Internal functions with default
--- implementations of 'ErrorHandler' and 'TerminationHandler'
---
--- The default 'ErrorHandler' and 'TerminationHandler' both throw a 'GetException'.
-
-module Data.Conduit.Cereal ( GetException
-                           , sinkGet
-                           , conduitGet
-                           , sourcePut
-                           , conduitPut
-                           ) where
-
-import           Control.Exception.Base
-import qualified Data.ByteString as BS
-import qualified Data.ByteString.Lazy as LBS
-import qualified Data.Conduit as C
-import qualified Data.Conduit.List as CL
-import           Data.Serialize hiding (get, put)
-import           Data.Typeable
-
-import           Data.Conduit.Cereal.Internal
-
-data GetException = GetException String
-  deriving (Show, Typeable)
-
-instance Exception GetException
-
--- | Run a 'Get' repeatedly on the input stream, producing an output stream of whatever the 'Get' outputs.
-conduitGet :: C.MonadThrow m => Get output -> C.Conduit BS.ByteString m output
-conduitGet = mkConduitGet errorHandler 
-  where errorHandler msg _ = pipeError $ GetException msg 
-
--- | Convert a 'Get' into a 'Sink'. The 'Get' will be streamed bytes until it returns 'Done' or 'Fail'.
---
--- If 'Get' succeed it will return the data read and unconsumed part of the input stream.
--- If the 'Get' fails due to deserialization error or early termination of the input stream it raise an error.
-sinkGet :: C.MonadThrow m => Get r -> C.Sink BS.ByteString m r
-sinkGet = mkSinkGet errorHandler terminationHandler 
-  where errorHandler msg _ = pipeError $ GetException msg 
-        terminationHandler f _ = let Fail msg = f BS.empty in pipeError $ GetException msg 
-
-pipeError :: (C.MonadThrow m, Exception e) => e -> C.Pipe i o m r
-pipeError e = C.PipeM (C.monadThrow e) undefined 
-
--- | Convert a 'Put' into a 'Source'. Runs in constant memory.
-sourcePut :: Monad m => Put -> C.Source m BS.ByteString
-sourcePut put = CL.sourceList $ LBS.toChunks $ runPutLazy put
-
--- | Run a 'Putter' repeatedly on the input stream, producing a concatenated 'ByteString' stream.
-conduitPut :: Monad m => Putter input -> C.Conduit input m BS.ByteString
-conduitPut p = CL.map $ runPut . p
+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE NoMonomorphismRestriction #-}+{-# LANGUAGE Rank2Types #-}+{-# LANGUAGE RankNTypes #-}++-- | Turn a 'Get' into a 'Sink' and a 'Put' into a 'Source'+-- These functions are built upno the Data.Conduit.Cereal.Internal functions with default+-- implementations of 'ErrorHandler' and 'TerminationHandler'+--+-- The default 'ErrorHandler' and 'TerminationHandler' both throw a 'GetException'.++module Data.Conduit.Cereal ( GetException+                           , sinkGet+                           , conduitGet+                           , sourcePut+                           , conduitPut+                           ) where++import           Control.Exception.Base+import           Control.Monad.Trans.Class (MonadTrans, lift)+import qualified Data.ByteString as BS+import qualified Data.ByteString.Lazy as LBS+import qualified Data.Conduit as C+import qualified Data.Conduit.List as CL+import           Data.Serialize hiding (get, put)+import           Data.Typeable++import           Data.Conduit.Cereal.Internal++data GetException = GetException String+  deriving (Show, Typeable)++instance Exception GetException++-- | Run a 'Get' repeatedly on the input stream, producing an output stream of whatever the 'Get' outputs.+conduitGet :: C.MonadThrow m => Get o -> C.GLConduit BS.ByteString m o+conduitGet = mkConduitGet errorHandler+  where errorHandler msg = pipeError $ GetException msg++-- | Convert a 'Get' into a 'Sink'. The 'Get' will be streamed bytes until it returns 'Done' or 'Fail'.+--+-- If 'Get' succeed it will return the data read and unconsumed part of the input stream.+-- If the 'Get' fails due to deserialization error or early termination of the input stream it raise an error.+sinkGet :: C.MonadThrow m => Get r -> C.GLSink BS.ByteString m r+sinkGet = mkSinkGet errorHandler terminationHandler+  where errorHandler msg = pipeError $ GetException msg+        terminationHandler f = let Fail msg = f BS.empty in pipeError $ GetException msg ++pipeError :: (C.MonadThrow m, MonadTrans t, Exception e) => e -> t m a+pipeError e = lift $ C.monadThrow e++-- | Convert a 'Put' into a 'Source'. Runs in constant memory.+sourcePut :: Monad m => Put -> C.GSource m BS.ByteString+sourcePut put = CL.sourceList $ LBS.toChunks $ runPutLazy put++-- | Run a 'Putter' repeatedly on the input stream, producing a concatenated 'ByteString' stream.+conduitPut :: Monad m => Putter a -> C.GInfConduit a m BS.ByteString+conduitPut p = CL.map $ runPut . p

@@ -1,62 +1,64 @@-module Data.Conduit.Cereal.Internal 
-  ( ErrorHandler
-  , TerminationHandler
-
-  , mkConduitGet
-  , mkSinkGet
-  ) where
-
-import qualified Data.ByteString as BS
-import qualified Data.Conduit as C
-import           Data.Serialize hiding (get, put)
-import           Data.Void 
-
--- | What should we do if the Get fails?
-type ErrorHandler i o m r = String -> Maybe BS.ByteString -> C.Pipe i o m r
-
--- | What should we do if the stream is done before the Get is done?
-type TerminationHandler i o m r = (BS.ByteString -> Result r) -> Maybe BS.ByteString -> C.Pipe i o m r
-
--- | Construct a conduitGet with the specified 'ErrorHandler'
-mkConduitGet :: Monad m 
-             => ErrorHandler BS.ByteString o m ()
-             -> Get o
-             -> C.Conduit BS.ByteString m o
-mkConduitGet errorHandler get = consume True (runGetPartial get) [] BS.empty 
-  where push f b s | BS.null s = C.NeedInput (push f b) (close b)
-                   | otherwise = consume False f b s
-        consume initial f b s = case f s of
-          Fail msg  -> errorHandler msg (chunkedStreamToMaybe consumed)
-          Partial p -> C.NeedInput (push p consumed) (close consumed)
-          Done a s' -> case initial of
-                         True  -> infiniteSequence a
-                         False -> C.HaveOutput (push (runGetPartial get) [] s') (return ()) a
-          where consumed = s : b
-                infiniteSequence r = C.HaveOutput (infiniteSequence r) (return ()) r
-                -- infinteSequence only works because the Get will either _always_ consume no input, or _never_ consume no input.
-        close b = C.Done (chunkedStreamToMaybe b) ()
-
--- | Construct a sinkGet with the specified 'ErrorHandler' and 'TerminationHandler'
-mkSinkGet :: Monad m 
-          => ErrorHandler BS.ByteString Void m r
-          -> TerminationHandler BS.ByteString Void m r 
-          -> Get r
-          -> C.Sink BS.ByteString m r
-mkSinkGet errorHandler terminationHandler get = consume (runGetPartial get) [] BS.empty
-  where push f b s
-          | BS.null s = C.NeedInput (push f b) (close f b)
-          | otherwise = consume f b s
-        consume f b s = case f s of
-          Fail msg  -> errorHandler msg (chunkedStreamToMaybe consumed)
-          Partial p -> C.NeedInput (push p consumed) (close p consumed)
-          Done r s' -> C.Done (streamToMaybe s') r
-          where consumed = s : b
-        close f = terminationHandler f . chunkedStreamToMaybe
-
-chunkedStreamToMaybe :: [BS.ByteString] -> Maybe BS.ByteString
-chunkedStreamToMaybe = streamToMaybe . BS.concat . reverse
-
-streamToMaybe :: BS.ByteString -> Maybe BS.ByteString
-streamToMaybe s = if BS.null s
-                    then Nothing
-                    else Just s
+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE Rank2Types #-}++module Data.Conduit.Cereal.Internal+  ( ConduitErrorHandler+  , SinkErrorHandler+  , SinkTerminationHandler++  , mkConduitGet+  , mkSinkGet+  ) where++import           Control.Monad (when)+import qualified Data.ByteString as BS+import qualified Data.Conduit as C+import           Data.Serialize hiding (get, put)++-- | What should we do if the Get fails?+type ConduitErrorHandler m o = String -> C.GLConduit BS.ByteString m o+type SinkErrorHandler m r = String -> C.GLSink BS.ByteString m r++-- | What should we do if the stream is done before the Get is done?+type SinkTerminationHandler m r = (BS.ByteString -> Result r) -> C.GLSink BS.ByteString m r++-- | Construct a conduitGet with the specified 'ErrorHandler'+mkConduitGet :: C.MonadThrow m+             => ConduitErrorHandler m o+             -> Get o+             -> C.GLConduit BS.ByteString m o+mkConduitGet errorHandler get = consume True (runGetPartial get) [] BS.empty+  where pull f b s+          | BS.null s = C.await >>= maybe (when (not $ null b) (C.leftover $ BS.concat $ reverse b)) (pull f b)+          | otherwise = consume False f b s+        consume initial f b s = case f s of+          Fail msg  -> do+            when (not $ null b) (C.leftover $ BS.concat $ reverse consumed)+            errorHandler msg+          Partial p -> pull p consumed BS.empty+          Done a s' -> case initial of+                         -- this only works because the Get will either _always_ consume no input, or _never_ consume no input.+                         True  -> sequence_ $ repeat $ C.yield a+                         False -> C.yield a >> pull (runGetPartial get) [] s'+--                         False -> C.yield a >> C.leftover s' >> mkConduitGet errorHandler get+          where consumed = s : b++-- | Construct a sinkGet with the specified 'ErrorHandler' and 'TerminationHandler'+mkSinkGet :: Monad m+          => SinkErrorHandler m r+          -> SinkTerminationHandler m r+          -> Get r+          -> C.GLSink BS.ByteString m r+mkSinkGet errorHandler terminationHandler get = consume (runGetPartial get) [] BS.empty+  where pull f b s+          | BS.null s = C.await >>= \ x -> case x of+                          Nothing -> when (not $ null b) (C.leftover $ BS.concat $ reverse b) >> terminationHandler f+                          Just a -> pull f b a+          | otherwise = consume f b s+        consume f b s = case f s of+          Fail msg  -> do+            when (not $ null b) (C.leftover $ BS.concat $ reverse consumed)+            errorHandler msg+          Partial p -> pull p consumed BS.empty+          Done r s' -> when (not $ BS.null s') (C.leftover s') >> return r+          where consumed = s : b

@@ -1,4 +1,6 @@ {-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE Rank2Types #-}  import           Control.Exception.Base import           Control.Monad.Identity@@ -26,6 +28,13 @@   y <- getWord8   return $ x + y +threeItemGet :: Get Word8+threeItemGet = do+  x <- getWord8+  y <- getWord8+  z <- getWord8+  return $ x + y + z+ putter :: Putter Char putter c = put x >> put (x + 1)   where x = (fromIntegral $ (fromEnum c) - (fromEnum 'a') :: Word8)@@ -70,21 +79,27 @@ -- Current sink implementation will terminate the pipe in case of error.  -- One may need non-terminating version like one defined below to get access to Leftovers -sinkGetMaybe :: Monad m => Get output -> C.Sink BS.ByteString m (Maybe output)-sinkGetMaybe = mkSinkGet errorHandler terminationHandler . fmap Just-  where errorHandler     msg s = C.Done s Nothing-        terminationHandler f s = C.Done s Nothing+sinkGetMaybe :: Get Word8 -> C.GLSink BS.ByteString (ExceptionT Identity) Word8+sinkGetMaybe = mkSinkGet errorHandler terminationHandler+  where errorHandler       _ = return 34+        terminationHandler _ = return 114  sinktest7 :: Test-sinktest7 = TestCase (assertBool "Leftover input with failure works"-  (case runIdentity $ do-     (CL.sourceList [BS.pack [1, 2]]) C.$$ (do-       output <- sinkGetMaybe (getWord8 >> fail "" :: Get Word8)-       output' <- CL.consume-       return (output, BS.concat output')) of-     (Nothing, bs) -> bs == BS.pack [1, 2]-     otherwise -> False))+sinktest7 = TestCase (assertEqual "Leftover input with failure works"+  (Right (34, BS.pack [1, 2]))+  (runIdentity $ runExceptionT $ (CL.sourceList [BS.pack [1, 2]]) C.$$ (do+    output <- sinkGetMaybe (getWord8 >> fail "" :: Get Word8)+    output' <- CL.consume+    return (output, BS.concat output')))) +sinktest8 :: Test+sinktest8 = TestCase (assertEqual "Leftover with incomplete input works"+  (Right (114, BS.singleton 1))+  (runIdentity $ runExceptionT $ (CL.sourceList [BS.singleton 1]) C.$$ (do+    output <- sinkGetMaybe twoItemGet+    output' <- CL.consume+    return (output, BS.concat output'))))+ conduittest1 :: Test conduittest1 = TestCase (assertEqual "Handles starting with empty bytestring"   (Right [])@@ -148,23 +163,50 @@     Left _ -> True     Right _ -> False)) --- This test CAN'T work because of the type of HaveOutput. conduittest13 :: Test conduittest13 = TestCase (assertEqual "Leftover success conduit input works"-  (Right ([12], BS.pack [3, 4, 5]))-  (runIdentity $ runExceptionT $ (CL.sourceList [BS.pack [10, 2, 3], BS.pack [4, 5]]) C.$$ (do-    output <- (conduitGet twoItemGet) C.=$ (CL.take 1)-    output' <- CL.consume-    return (output, BS.concat output'))))+  (Right [Right 12, Right 7, Left (BS.pack [5])])+  (runIdentity $ runExceptionT $ (CL.sourceList [BS.pack [10, 2, 3], BS.pack [4, 5]]) C.$= fancyConduit C.$$ CL.consume))+  where fancyConduit = do+          conduitGet twoItemGet C.=$= CL.map (\ x -> Right x)+          recurse+          where recurse = C.await >>= maybe (return ()) (\ x -> C.yield (Left x) >> recurse)  conduittest14 :: Test-conduittest14 = TestCase (assertEqual "Leftover failure conduit input works"+conduittest14 = TestCase (assertEqual "Leftover coercing works"+  (Right [Left (BS.pack [10, 2])])+  (runIdentity $ runExceptionT $ (CL.sourceList [BS.pack [10], BS.pack [2]]) C.$= fancyConduit C.$$ CL.consume))+  where fancyConduit = do+          conduitGet threeItemGet C.=$= CL.map (\ x -> Right x)+          recurse+          where recurse = C.await >>= maybe (return ()) (\ x -> C.yield (Left x) >> recurse)++conduittest15 :: Test+conduittest15 = TestCase (assertEqual "Leftover premature end conduit input works"   (Right ([], BS.singleton 1))   (runIdentity $ runExceptionT $ (CL.sourceList [BS.singleton 1]) C.$$ (do     output <- (conduitGet twoItemGet) C.=$ (CL.take 1)     output' <- CL.consume     return (output, BS.concat output')))) +conduittest16 :: Test+conduittest16 = TestCase (assertEqual "Leftover failure conduit input works"+  (Right [Left $ BS.pack [10, 11], Left $ BS.singleton 2])+  (runIdentity $ runExceptionT $ (CL.sourceList [BS.pack [10, 11], BS.pack [2]]) C.$= fancyConduit C.$$ CL.consume))+  where fancyConduit = do+          mkConduitGet (const $ return ()) (getWord8 >> fail "asdf" :: Get Word8) C.=$= CL.map (\ x -> Right x)+          recurse+          where recurse = C.await >>= maybe (return ()) (\ x -> C.yield (Left x) >> recurse)++conduittest17 :: Test+conduittest17 = TestCase (assertEqual "Leftover failure conduit with broken input works"+  (Right [Left $ BS.pack [10, 11], Left $ BS.singleton 12])+  (runIdentity $ runExceptionT $ (CL.sourceList [BS.singleton 10, BS.singleton 11, BS.singleton 12]) C.$= fancyConduit C.$$ CL.consume))+  where fancyConduit = do+          mkConduitGet (const $ return ()) (twoItemGet >> fail "asdf" :: Get Word8) C.=$= CL.map (\ x -> Right x)+          recurse+          where recurse = C.await >>= maybe (return ()) (\ x -> C.yield (Left x) >> recurse)+ puttest1 :: Test puttest1 = TestCase (assertEqual "conduitPut works"   [BS.pack [0, 1]]@@ -187,6 +229,7 @@                      , sinktest5                      , sinktest6                      , sinktest7+                     , sinktest8                      ]  conduittests = TestList [ conduittest1@@ -201,8 +244,11 @@                         , conduittest10                         , conduittest11                         , conduittest12-                        --, conduittest13+                        , conduittest13                         , conduittest14+                        , conduittest15+                        , conduittest16+                        , conduittest17                         ]  puttests = TestList [ puttest1

@@ -1,5 +1,5 @@ name:            cereal-conduit-version:         0.5+version:         0.6 license:         BSD3 license-file:    LICENSE author:          Myles C. Maxfield <myles.maxfield@gmail.com>@@ -14,10 +14,10 @@  library     build-depends: base         >= 4       && < 5-                 , conduit      >= 0.4.0   && < 0.5.0+                 , conduit      >= 0.5.0   && < 0.6.0                  , cereal       >= 0.3.1.0                  , bytestring-                 , void+                 , transformers >= 0.2.0.0     exposed-modules: Data.Conduit.Cereal                    , Data.Conduit.Cereal.Internal     ghc-options:     -Wall@@ -26,14 +26,14 @@     type: exitcode-stdio-1.0     main-is: Test/Main.hs     build-depends: base >= 4 && < 5-                 , conduit >= 0.4.0 && < 0.5.0+                 , conduit >= 0.5.0 && < 0.6.0                  , cereal >= 0.3.1.0                  , bytestring                  --, test-framework-hunit                  , HUnit-                 , void                  , resourcet                  , mtl+                 , transformers >= 0.2.0.0  source-repository head   type:     git