conduit-algorithms 0.0.8.2 → 0.0.9.0
raw patch · 4 files changed
+129/−78 lines, 4 filesdep +conduit-zstd
Dependencies added: conduit-zstd
Files
- ChangeLog +4/−0
- Data/Conduit/Algorithms/Async.hs +91/−72
- conduit-algorithms.cabal +8/−5
- tests/Tests.hs +26/−1
ChangeLog view
@@ -1,3 +1,7 @@+Version 0.0.9.0 2019-01-09 by luispedro+ * Add Zstandard compression+ * Refactor internals to avoid code repetition+ Version 0.0.8.2 2018-09-27 by luispedro * Add withPossiblyCompressedFile function for prompt resource deallocation
Data/Conduit/Algorithms/Async.hs view
@@ -1,6 +1,6 @@ {-| Module : Data.Conduit.Algorithms.Async-Copyright : 2013-2018 Luis Pedro Coelho+Copyright : 2013-2019 Luis Pedro Coelho License : MIT Maintainer : luis@luispedro.org @@ -26,6 +26,10 @@ , asyncXzToFile , asyncXzFrom , asyncXzFromFile+ , asyncZstdTo+ , asyncZstdToFile+ , asyncZstdFrom+ , asyncZstdFromFile , unorderedAsyncMapC ) where @@ -42,7 +46,9 @@ import qualified Data.Conduit.Zlib as CZ import qualified Data.Conduit.Lzma as CX import qualified Data.Streaming.Zlib as SZ-import qualified Data.Conduit.BZlib as CZ+import qualified Data.Conduit.BZlib as CBZ+import qualified Data.Conduit.Zstd as CZstd+import qualified Control.Monad.Trans.Resource as R import qualified Data.Conduit as C import Data.Conduit ((.|)) @@ -185,30 +191,59 @@ untilNothing _ -> return () --- | A simple sink which performs gzip compression in a separate thread and--- writes the results to `h`.------ See also 'asyncGzipToFile'-asyncGzipTo :: forall m. (MonadIO m, MonadUnliftIO m) => Handle -> C.ConduitT B.ByteString C.Void m ()-asyncGzipTo h = do- let drain q = liftIO . C.runConduit $++genericAsyncFrom :: forall m. (MonadIO m, MonadUnliftIO m) => C.ConduitT B.ByteString B.ByteString m () -> Handle -> C.ConduitT () B.ByteString m ()+genericAsyncFrom transform h = do+ let prod q = do+ C.runConduit $+ C.sourceHandle h+ .| transform+ .| CL.map Just+ .| CA.sinkTBQueue q+ liftIO $ atomically (TQ.writeTBQueue q Nothing)+ CA.gatherFrom 8 prod .| untilNothing++genericAsyncTo :: forall m. (MonadIO m, MonadUnliftIO m) => C.ConduitT B.ByteString B.ByteString (R.ResourceT IO) () -> Handle -> C.ConduitT B.ByteString C.Void m ()+genericAsyncTo tranform h = do+ let drain q = liftIO . C.runConduitRes $ CA.sourceTBQueue q .| untilNothing .| CL.map (B.concat . reverse)- .| CZ.gzip+ .| tranform .| C.sinkHandle h bsConcatTo ((2 :: Int) ^ (15 :: Int)) .| CA.drainTo 8 drain ++genericFromFile :: forall m. (MonadResource m, MonadUnliftIO m) => (Handle -> C.ConduitT () B.ByteString m ()) -> FilePath -> C.ConduitT () B.ByteString m ()+genericFromFile from fname = C.bracketP+ (openFile fname ReadMode)+ hClose+ from++genericToFile :: forall m. (MonadResource m, MonadUnliftIO m) => (Handle -> C.ConduitT B.ByteString C.Void m ()) -> FilePath -> C.ConduitT B.ByteString C.Void m ()+genericToFile to fname = C.bracketP+ (openFile fname WriteMode)+ hClose+ to++-- | A simple sink which performs gzip compression in a separate thread and+-- writes the results to `h`.+--+-- See also 'asyncGzipToFile'+asyncGzipTo :: forall m. (MonadIO m, MonadUnliftIO m) => Handle -> C.ConduitT B.ByteString C.Void m ()+asyncGzipTo h = genericAsyncTo gz h+ where+ gz = CZ.gzip `C.catchC` handleZLibException+ handleZLibException = \(e :: SZ.ZlibException) ->+ liftIO . ioError $ mkIOError userErrorType ("Error compressing gzip stream: "++displayException e) (Just h) Nothing+ -- | Compresses the output and writes to the given file with compression being -- performed in a separate thread. -- -- See also 'asyncGzipTo' asyncGzipToFile :: forall m. (MonadResource m, MonadUnliftIO m) => FilePath -> C.ConduitT B.ByteString C.Void m ()-asyncGzipToFile fname = C.bracketP- (openFile fname WriteMode)- hClose- asyncGzipTo+asyncGzipToFile = genericToFile asyncGzipTo -- | A source which produces the ungzipped content from the the given handle. -- Note that this "reads ahead" so if you do not use all the input, the Handle@@ -233,35 +268,21 @@ -- -- See also 'asyncGzipFrom' asyncGzipFromFile :: forall m. (MonadResource m, MonadUnliftIO m) => FilePath -> C.ConduitT () B.ByteString m ()-asyncGzipFromFile fname = C.bracketP- (openFile fname ReadMode)- hClose- asyncGzipFrom+asyncGzipFromFile = genericFromFile asyncGzipFrom -- | A simple sink which performs bzip2 compression in a separate thread and -- writes the results to `h`. -- -- See also 'asyncBzip2ToFile'-asyncBzip2To :: forall m. (MonadIO m, MonadResource m, MonadUnliftIO m) => Handle -> C.ConduitT B.ByteString C.Void m ()-asyncBzip2To h = do- let drain q = C.runConduit $- CA.sourceTBQueue q- .| untilNothing- .| CL.map (B.concat . reverse)- .| CZ.bzip2- .| C.sinkHandle h- bsConcatTo ((2 :: Int) ^ (15 :: Int))- .| CA.drainTo 8 drain+asyncBzip2To :: forall m. (MonadIO m, MonadUnliftIO m) => Handle -> C.ConduitT B.ByteString C.Void m ()+asyncBzip2To = genericAsyncTo CBZ.bzip2 -- | Compresses the output and writes to the given file with compression being -- performed in a separate thread. -- -- See also 'asyncBzip2To' asyncBzip2ToFile :: forall m. (MonadResource m, MonadUnliftIO m) => FilePath -> C.ConduitT B.ByteString C.Void m ()-asyncBzip2ToFile fname = C.bracketP- (openFile fname WriteMode)- hClose- asyncBzip2To+asyncBzip2ToFile = genericToFile asyncBzip2To -- | A source which produces the bzipped2 content from the the given handle. -- Note that this "reads ahead" so if you do not use all the input, the Handle@@ -269,50 +290,28 @@ -- -- See also 'asyncBzip2FromFile' asyncBzip2From :: forall m. (MonadIO m, MonadResource m, MonadUnliftIO m) => Handle -> C.ConduitT () B.ByteString m ()-asyncBzip2From h = do- let prod q = do- C.runConduit $- C.sourceHandle h- .| CZ.multiple CZ.bunzip2- .| CL.map Just- .| CA.sinkTBQueue q- liftIO $ atomically (TQ.writeTBQueue q Nothing)- CA.gatherFrom 8 prod .| untilNothing+asyncBzip2From = genericAsyncFrom (CZ.multiple CBZ.bunzip2) -- | Open and read a bzip2 file with the uncompression being performed in a -- separate thread. -- -- See also 'asyncBzip2From' asyncBzip2FromFile :: forall m. (MonadResource m, MonadUnliftIO m) => FilePath -> C.ConduitT () B.ByteString m ()-asyncBzip2FromFile fname = C.bracketP- (openFile fname ReadMode)- hClose- asyncBzip2From+asyncBzip2FromFile = genericFromFile asyncBzip2From -- | A simple sink which performs lzma/xz compression in a separate thread and -- writes the results to `h`. -- -- See also 'asyncXzToFile' asyncXzTo :: forall m. (MonadIO m, MonadResource m, MonadUnliftIO m) => Handle -> C.ConduitT B.ByteString C.Void m ()-asyncXzTo h = do- let drain q = C.runConduit $- CA.sourceTBQueue q- .| untilNothing- .| CL.map (B.concat . reverse)- .| CX.compress Nothing- .| C.sinkHandle h- bsConcatTo ((2 :: Int) ^ (15 :: Int))- .| CA.drainTo 8 drain+asyncXzTo = genericAsyncTo (CX.compress Nothing) -- | Compresses the output and writes to the given file with compression being -- performed in a separate thread. -- -- See also 'asyncXzTo' asyncXzToFile :: forall m. (MonadResource m, MonadUnliftIO m) => FilePath -> C.ConduitT B.ByteString C.Void m ()-asyncXzToFile fname = C.bracketP- (openFile fname WriteMode)- hClose- asyncXzTo+asyncXzToFile = genericToFile asyncXzTo -- | A source which produces the unxzipped content from the the given handle. -- Note that this "reads ahead" so if you do not use all the input, the Handle@@ -320,29 +319,46 @@ -- -- See also 'asyncXzFromFile' asyncXzFrom :: forall m. (MonadIO m, MonadResource m, MonadUnliftIO m, MonadThrow m) => Handle -> C.ConduitT () B.ByteString m ()-asyncXzFrom h = do+asyncXzFrom = let oneGBmembuffer = Just $ 1024 ^ (3 :: Integer)- prod q = do- C.runConduit $- C.sourceHandle h- .| CZ.multiple (CX.decompress oneGBmembuffer)- .| CL.map Just- .| CA.sinkTBQueue q- liftIO $ atomically (TQ.writeTBQueue q Nothing)- CA.gatherFrom 8 prod .| untilNothing-+ in genericAsyncFrom (CX.decompress oneGBmembuffer) -- | Open and read a lzma/xz file with the uncompression being performed in a -- separate thread. -- -- See also 'asyncXzFrom' asyncXzFromFile :: forall m. (MonadResource m, MonadUnliftIO m, MonadThrow m) => FilePath -> C.ConduitT () B.ByteString m ()-asyncXzFromFile fname = C.bracketP- (openFile fname ReadMode)- hClose- asyncXzFrom+asyncXzFromFile = genericFromFile asyncXzFrom +-- | Decompress ZStd format using a separate thread+--+-- See also 'asyncZstdFromFile'+asyncZstdFrom :: forall m. (MonadIO m, MonadUnliftIO m) => Handle -> C.ConduitT () B.ByteString m ()+asyncZstdFrom = genericAsyncFrom CZstd.decompress +-- | Compress in ZStd format using a separate thread and write to a file+-- See also 'asyncZstdFrom'+asyncZstdFromFile :: forall m. (MonadResource m, MonadUnliftIO m, MonadThrow m) => FilePath -> C.ConduitT () B.ByteString m ()+asyncZstdFromFile = genericFromFile asyncZstdFrom+++-- | Compress in Zstd format using a separate thread+-- +-- See also 'asyncZstdToFile'+asyncZstdTo :: forall m. (MonadIO m, MonadUnliftIO m) =>+ Int -- ^ compression level+ -> Handle -> C.ConduitT B.ByteString C.Void m ()+asyncZstdTo clevel = genericAsyncTo (CZstd.compress clevel)+++-- | Compress in ZStd format using a separate thread and write to a file+--+-- This will use compression level 3 as this is the default in the ZStd C API+--+-- See also 'asyncZstdTo'+asyncZstdToFile :: forall m. (MonadResource m, MonadUnliftIO m) => FilePath -> C.ConduitT B.ByteString C.Void m ()+asyncZstdToFile = genericToFile (asyncZstdTo 3)+ -- | If the filename indicates a supported compressed file (gzip, xz, and, on -- Unix, bzip2), then it reads it and uncompresses it. --@@ -368,6 +384,7 @@ | ".gz" `isSuffixOf` fname = asyncGzipFrom | ".xz" `isSuffixOf` fname = asyncXzFrom | ".bz2" `isSuffixOf` fname = asyncBzip2From+ | ".zstd" `isSuffixOf` fname = asyncZstdFrom | otherwise = C.sourceHandle @@ -384,6 +401,7 @@ | ".gz" `isSuffixOf` fname = asyncGzipFromFile fname | ".xz" `isSuffixOf` fname = asyncXzFromFile fname | ".bz2" `isSuffixOf` fname = asyncBzip2FromFile fname+ | ".zstd" `isSuffixOf` fname = asyncZstdFromFile fname | otherwise = C.sourceFile fname -- | If the filename indicates a gzipped file (or, on Unix, also a bz2 file),@@ -395,4 +413,5 @@ | ".gz" `isSuffixOf` fname = asyncGzipToFile fname | ".xz" `isSuffixOf` fname = asyncXzToFile fname | ".bz2" `isSuffixOf` fname = asyncBzip2ToFile fname+ | ".zstd" `isSuffixOf` fname = asyncZstdToFile fname | otherwise = C.sinkFile fname
conduit-algorithms.cabal view
@@ -2,10 +2,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: 0fe3779c7b8153d9edb2c1061a1ea1ce629a6e6474a0c02fd622a91b5c237a31+-- hash: 568a611355a3ef3c59150cc32f207c38a0d6d802487517f30c8e437c8c839ab8 name: conduit-algorithms-version: 0.0.8.2+version: 0.0.9.0 synopsis: Conduit-based algorithms description: Algorithms on Conduits, including higher level asynchronous processing and some other utilities. category: Conduit@@ -33,7 +33,7 @@ Data.Conduit.Algorithms.Async.ByteString Data.Conduit.Algorithms.Storable default-extensions: BangPatterns OverloadedStrings LambdaCase TupleSections- ghc-options: -Wall -O2+ ghc-options: -Wall build-depends: async , base >4.8 && <5@@ -42,6 +42,7 @@ , conduit >=1.3 , conduit-combinators >=1.1.2 , conduit-extra+ , conduit-zstd , containers , deepseq , exceptions@@ -68,7 +69,7 @@ hs-source-dirs: ./tests default-extensions: BangPatterns OverloadedStrings LambdaCase TupleSections- ghc-options: -Wall -O2+ ghc-options: -Wall build-depends: HUnit , async@@ -79,6 +80,7 @@ , conduit-algorithms , conduit-combinators >=1.1.2 , conduit-extra+ , conduit-zstd , containers , deepseq , directory@@ -107,7 +109,7 @@ hs-source-dirs: ./bench default-extensions: BangPatterns OverloadedStrings LambdaCase TupleSections- ghc-options: -Wall -O2 -Wall -fwarn-tabs -fno-warn-missing-signatures -threaded -rtsopts "-with-rtsopts=-A64m -n4m -H"+ ghc-options: -Wall -Wall -fwarn-tabs -fno-warn-missing-signatures -threaded -rtsopts "-with-rtsopts=-A64m -n4m -H" build-depends: async , base >4.8 && <5@@ -117,6 +119,7 @@ , conduit-algorithms , conduit-combinators >=1.1.2 , conduit-extra+ , conduit-zstd , containers , criterion , deepseq
tests/Tests.hs view
@@ -1,4 +1,4 @@-{- Copyright 2017-2018 Luis Pedro Coelho+{- Copyright 2017-2019 Luis Pedro Coelho - License: MIT -} {-# LANGUAGE TemplateHaskell, CPP, QuasiQuotes, FlexibleContexts, OverloadedStrings #-}@@ -46,6 +46,8 @@ testingFileNameXZ = "file_just_for_testing_delete_me_please.xz" testingFileNameXZ2 :: FilePath testingFileNameXZ2 = "file_just_for_testing_delete_me_please_2.xz"+testingFileNameZstd :: FilePath+testingFileNameZstd = "file_just_for_testing_delete_me_please.zstd" extract :: C.ConduitM () b FID.Identity () -> [b] extract c = C.runConduitPure (c .| CC.sinkList)@@ -173,6 +175,22 @@ r @?= "Hello World" removeFile testingFileNameGZ +case_asyncGzipLarge :: IO ()+case_asyncGzipLarge = do+ let repeats = 16384 -- large enough to cause the input to be split into blocks inside asyncGzipToFile+ C.runConduitRes (CC.yieldMany (concat $ replicate repeats ["Hello" :: B.ByteString, " ", "World\n"]) .| CAlg.asyncGzipToFile testingFileNameGZ)+ let checkHello !n = C.await >>= \case+ Nothing -> return n+ Just "Hello World\n" -> checkHello (n + 1)+ _ -> return (-1)+ r <- C.runConduitRes $+ CAlg.asyncGzipFromFile testingFileNameGZ+ .| CC.chunksOfE 12+ .| checkHello (0 :: Int)++ removeFile testingFileNameGZ+ r @?= repeats+ case_asyncBzip2 :: IO () case_asyncBzip2 = do C.runConduitRes (CC.yieldMany ["Hello", " ", "World"] .| CAlg.asyncBzip2ToFile testingFileNameBZ2)@@ -250,6 +268,13 @@ .| CL.map (read . B8.unpack) removeFile testingFileNameXZ removeFile testingFileNameXZ2++case_async_zstd_to_from :: IO ()+case_async_zstd_to_from = do+ C.runConduitRes (CC.yieldMany ["Hello", " ", "World"] .| CAlg.asyncZstdToFile testingFileNameZstd)+ r <- B.concat <$> extractIO (CAlg.asyncZstdFromFile testingFileNameZstd)+ r @?= "Hello World"+ removeFile testingFileNameZstd case_asyncFilterLines :: IO () case_asyncFilterLines = do