conduit-algorithms 0.0.7.2 → 0.0.7.3
raw patch · 4 files changed
+259/−25 lines, 4 filesdep +monad-control
Dependencies added: monad-control
Files
- ChangeLog +4/−0
- Data/Conduit/Algorithms/Async.hs +131/−10
- conduit-algorithms.cabal +14/−4
- tests/Tests.hs +110/−11
ChangeLog view
@@ -1,3 +1,7 @@+Version 0.0.7.3 2018-03-12 by luispedro+ * Add xz support (patches by Renato Alves)+ * Add bzip2 support on Windows (patches by Renato Alves)+ Version 0.0.7.2 2018-02-19 by luispedro * Add handle information to error message to Handle in asyncGzipFrom
Data/Conduit/Algorithms/Async.hs view
@@ -10,12 +10,21 @@ module Data.Conduit.Algorithms.Async ( conduitPossiblyCompressedFile+ , conduitPossiblyCompressedToFile , asyncMapC , asyncMapEitherC , asyncGzipTo , asyncGzipToFile , asyncGzipFrom , asyncGzipFromFile+ , asyncBzip2To+ , asyncBzip2ToFile+ , asyncBzip2From+ , asyncBzip2FromFile+ , asyncXzTo+ , asyncXzToFile+ , asyncXzFrom+ , asyncXzFromFile , unorderedAsyncMapC ) where @@ -30,12 +39,11 @@ import qualified Data.Conduit.TQueue as CA import qualified Data.Conduit.List as CL import qualified Data.Conduit.Zlib as CZ+#ifndef WINDOWS import qualified Data.Conduit.Lzma as CX+#endif import qualified Data.Streaming.Zlib as SZ-#ifndef WINDOWS--- bzlib cannot compile on Windows (as of 2016/07/05) import qualified Data.Conduit.BZlib as CZ-#endif import qualified Data.Conduit as C import Data.Conduit ((.|)) @@ -229,6 +237,118 @@ hClose asyncGzipFrom +-- | A simple sink which performs bzip2 compression in a separate thread and+-- writes the results to `h`.+--+-- See also 'asyncGzipToFile'+asyncBzip2To :: forall m. (MonadIO m, MonadResource m, MonadBaseControl IO m) => Handle -> C.Sink B.ByteString 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++-- | Compresses the output and writes to the given file with compression being+-- performed in a separate thread.+--+-- See also 'asyncGzipTo'+asyncBzip2ToFile :: forall m. (MonadResource m, MonadBaseControl IO m) => FilePath -> C.Sink B.ByteString m ()+asyncBzip2ToFile fname = C.bracketP+ (openFile fname WriteMode)+ hClose+ 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+-- will probably be left at an undefined position in the file.+--+-- See also 'asyncGzipFromFile'+asyncBzip2From :: forall m. (MonadIO m, MonadResource m, MonadBaseControl IO m) => Handle -> C.Source m B.ByteString+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++-- | Open and read a bzip2 file with the uncompression being performed in a+-- separate thread.+--+-- See also 'asyncGzipFrom'+asyncBzip2FromFile :: forall m. (MonadResource m, MonadBaseControl IO m) => FilePath -> C.Source m B.ByteString+asyncBzip2FromFile fname = C.bracketP+ (openFile fname ReadMode)+ hClose+ asyncBzip2From++-- | A simple sink which performs lzma/xz compression in a separate thread and+-- writes the results to `h`.+--+-- See also 'asyncGzipToFile'+asyncXzTo :: forall m. (MonadIO m, MonadResource m, MonadBaseControl IO m) => Handle -> C.Sink B.ByteString m ()+asyncXzTo h = do+ let drain q = C.runConduit $+ CA.sourceTBQueue q+ .| untilNothing+ .| CL.map (B.concat . reverse)+#ifndef WINDOWS+ .| CX.compress Nothing+#else+ .| error "lzma/xz compression is not available on Windows"+#endif+ .| C.sinkHandle h+ bsConcatTo ((2 :: Int) ^ (15 :: Int))+ .| CA.drainTo 8 drain++-- | Compresses the output and writes to the given file with compression being+-- performed in a separate thread.+--+-- See also 'asyncGzipTo'+asyncXzToFile :: forall m. (MonadResource m, MonadBaseControl IO m) => FilePath -> C.Sink B.ByteString m ()+asyncXzToFile fname = C.bracketP+ (openFile fname WriteMode)+ hClose+ 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+-- will probably be left at an undefined position in the file.+--+-- See also 'asyncGzipFromFile'+asyncXzFrom :: forall m. (MonadIO m, MonadResource m, MonadBaseControl IO m) => Handle -> C.Source m B.ByteString+asyncXzFrom h = do+ let oneGBmembuffer = Just $ 1024 ^ (3 :: Integer)+ prod q = do+ C.runConduit $+ C.sourceHandle h+#ifndef WINDOWS+ .| CZ.multiple (CX.decompress oneGBmembuffer)+#else+ .| error "lzma/xz decompression is not available on Windows"+#endif+ .| CL.map Just+ .| CA.sinkTBQueue q+ liftIO $ atomically (TQ.writeTBQueue q Nothing)+ CA.gatherFrom 8 prod .| untilNothing+++-- | Open and read a lzma/xz file with the uncompression being performed in a+-- separate thread.+--+-- See also 'asyncXzFrom'+asyncXzFromFile :: forall m. (MonadResource m, MonadBaseControl IO m) => FilePath -> C.Source m B.ByteString+asyncXzFromFile fname = C.bracketP+ (openFile fname ReadMode)+ hClose+ asyncXzFrom+ -- | If the filename indicates a gzipped file (or, on Unix, also a bz2 file), -- then it reads it and uncompresses it. --@@ -238,12 +358,13 @@ conduitPossiblyCompressedFile :: (MonadBaseControl IO m, MonadResource m) => FilePath -> C.Source m B.ByteString conduitPossiblyCompressedFile fname | ".gz" `isSuffixOf` fname = asyncGzipFromFile fname- | ".xz" `isSuffixOf` fname = C.sourceFile fname .| CX.decompress oneGBmembuffer-#ifndef WINDOWS- | ".bz2" `isSuffixOf` fname = C.sourceFile fname .| CZ.bunzip2-#else- | ".bz2" `isSuffixOf` fname = error "bzip2 decompression is not available on Windows"-#endif+ | ".xz" `isSuffixOf` fname = asyncXzFromFile fname+ | ".bz2" `isSuffixOf` fname = asyncBzip2FromFile fname | otherwise = C.sourceFile fname- where oneGBmembuffer = Just $ 1024 * 1024 * 1024 +conduitPossiblyCompressedToFile :: (MonadBaseControl IO m, MonadResource m) => FilePath -> C.Sink B.ByteString m ()+conduitPossiblyCompressedToFile fname+ | ".gz" `isSuffixOf` fname = asyncGzipToFile fname+ | ".xz" `isSuffixOf` fname = asyncXzToFile fname+ | ".bz2" `isSuffixOf` fname = asyncBzip2ToFile fname+ | otherwise = C.sinkFile fname
conduit-algorithms.cabal view
@@ -2,10 +2,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: d2ff0dcb74b83abcf9bd574764e0d2acf0134cba6ae1d4a221fe6abf3d90b2ea+-- hash: 3a46059f9e3513ce4b3dce9af08750249e1263b0cfac24b7294a5d5269f6344e name: conduit-algorithms-version: 0.0.7.2+version: 0.0.7.3 synopsis: Conduit-based algorithms description: Algorithms on Conduits, including higher level asynchronous processing and some other utilities. category: Conduit@@ -39,7 +39,7 @@ , conduit-extra , containers , deepseq- , lzma-conduit+ , monad-control , mtl , resourcet , stm@@ -47,6 +47,11 @@ , streaming-commons , transformers , vector+ if os(windows)+ cpp-options: -DWINDOWS+ else+ build-depends:+ lzma-conduit exposed-modules: Data.Conduit.Algorithms Data.Conduit.Algorithms.Utils@@ -75,7 +80,7 @@ , containers , deepseq , directory- , lzma-conduit+ , monad-control , mtl , resourcet , stm@@ -86,6 +91,11 @@ , test-framework-th , transformers , vector+ if os(windows)+ cpp-options: -DWINDOWS+ else+ build-depends:+ lzma-conduit other-modules: Paths_conduit_algorithms default-language: Haskell2010
tests/Tests.hs view
@@ -1,7 +1,7 @@ {- Copyright 2017-2018 Luis Pedro Coelho - License: MIT -}-{-# LANGUAGE TemplateHaskell, QuasiQuotes, FlexibleContexts, OverloadedStrings #-}+{-# LANGUAGE TemplateHaskell, CPP, QuasiQuotes, FlexibleContexts, OverloadedStrings #-} module Main where import Test.Framework.TH@@ -14,11 +14,15 @@ import qualified Data.Conduit.Combinators as CC import qualified Data.Conduit.Binary as CB import qualified Data.Conduit.List as CL+import qualified Data.Functor.Identity as FID import qualified Data.Vector.Storable as VS import Data.Conduit ((.|)) import Data.List (sort) import System.Directory (removeFile)+import Control.Exception (catch, ErrorCall) import Control.Monad (forM_)+import Control.Monad.Trans.Resource.Internal (ResourceT)+import Control.Monad.Trans.Control (MonadBaseControl) import qualified Data.Conduit.Algorithms as CAlg import qualified Data.Conduit.Algorithms.Storable as CAlg@@ -33,17 +37,43 @@ testingFileNameGZ = "file_just_for_testing_delete_me_please.gz" testingFileNameGZ2 :: FilePath testingFileNameGZ2 = "file_just_for_testing_delete_me_please_2.gz"+testingFileNameBZ2 :: FilePath+testingFileNameBZ2 = "file_just_for_testing_delete_me_please.bz2"+testingFileNameBZ22 :: FilePath+testingFileNameBZ22 = "file_just_for_testing_delete_me_please_2.bz2"+testingFileNameXZ :: FilePath+testingFileNameXZ = "file_just_for_testing_delete_me_please.xz"+testingFileNameXZ2 :: FilePath+testingFileNameXZ2 = "file_just_for_testing_delete_me_please_2.xz" +extract :: C.ConduitM () b FID.Identity () -> [b] extract c = C.runConduitPure (c .| CC.sinkList) +extractIO :: MonadBaseControl IO m => C.ConduitM () b (ResourceT m) () -> m [b] extractIO c = C.runConduitRes (c .| CC.sinkList) +shouldProduce :: (Show b, Eq b) => [b] -> C.ConduitM () b FID.Identity () -> Assertion shouldProduce values cond = extract cond @?= values++shouldProduceIO :: (Show b, Eq b) => [b] -> C.ConduitM () b (ResourceT IO) () -> IO () shouldProduceIO values cond = do p <- extractIO cond p @?= values +assertError :: IO f -> IO ()+assertError f = do+ errored <- catch (f >> pure False) handler+ if errored then+ pure ()+ else+ assertFailure "Expected error but none was triggered"+ where+ handler :: ErrorCall -> IO Bool+ handler _ = pure True++case_uniqueC :: Assertion case_uniqueC = extract (CC.yieldMany [1,2,3,1,1,2,3] .| CAlg.uniqueC) @=? [1,2,3 :: Int]+case_mergeC :: Assertion case_mergeC = shouldProduce expected $ CAlg.mergeC [ CC.yieldMany i1@@ -57,6 +87,7 @@ i2 = [ 1, 4, 4, 5] i3 = [-1, 0, 7] +case_mergeCmonad :: Assertion case_mergeCmonad = shouldProduce expected $ CAlg.mergeC [ mYield i1@@ -67,24 +98,30 @@ expected = sort (concat [i1, i2, i3]) mYield lst = do let lst' = map return lst- forM_ lst' $ \elem -> do- elem' <- elem- C.yield elem'+ forM_ lst' $ \elemnt -> do+ elemnt' <- elemnt+ C.yield elemnt' i1 = [ 0, 2, 4 :: Int] i2 = [ 1, 3, 4, 5] i3 = [-1, 0, 7] +case_mergeCempty :: Assertion+case_mergeCempty = shouldProduce ([] :: [Int]) $ CAlg.mergeC []++case_mergeC2 :: Assertion case_mergeC2 = shouldProduce [0, 1, 1, 2, 3, 5 :: Int] $ CAlg.mergeC2 (CC.yieldMany [0, 1, 2]) (CC.yieldMany [1, 3, 5]) +case_mergeC2same :: Assertion case_mergeC2same = shouldProduce [0, 0, 1, 1, 2, 2 :: Int] $ CAlg.mergeC2 (CC.yieldMany [0, 1, 2]) (CC.yieldMany [0, 1, 2]) +case_mergeC2monad :: Assertion case_mergeC2monad = shouldProduce [0, 1, 2, 2, 3, 4 :: Int] $ do CAlg.mergeC2 (CC.yieldMany [0, 2])@@ -92,33 +129,54 @@ CC.yieldMany [3] CC.yieldMany [4] +case_groupC :: Assertion case_groupC = shouldProduce [[0,1,2], [3,4,5], [6,7,8], [9, 10 :: Int]] $ CC.yieldMany [0..10] .| CAlg.groupC 3 +case_enumerateC :: Assertion case_enumerateC = shouldProduce [(0,'z'), (1,'o'), (2,'t')] $- CC.yieldMany ("zot" :: [Char]) .| CAlg.enumerateC+ CC.yieldMany ("zot" :: String) .| CAlg.enumerateC +case_removeRepeatsC :: Assertion case_removeRepeatsC = shouldProduce [0,1,2,3,4,5,6,7,8,9, 10 :: Int] $ CC.yieldMany [0,0,0,1,1,1,2,2,3,4,5,6,6,6,6,7,7,8,9,10,10] .| CAlg.removeRepeatsC case_asyncMap :: IO () case_asyncMap = do vals <- extractIO (CC.yieldMany [0..10] .| CAlg.asyncMapC 3 (+ (1:: Int)))- (vals @?= [1..11])+ vals @?= [1..11] case_unorderedAsyncMapC :: IO () case_unorderedAsyncMapC = do vals <- extractIO (CC.yieldMany [0..10] .| CAlg.unorderedAsyncMapC 3 (+ (1:: Int)))- (sort vals @?= [1..11])+ sort vals @?= [1..11] case_asyncGzip :: IO () case_asyncGzip = do C.runConduitRes (CC.yieldMany ["Hello", " ", "World"] .| CAlg.asyncGzipToFile testingFileNameGZ)- r <- B.concat <$> (extractIO (CAlg.asyncGzipFromFile testingFileNameGZ))+ r <- B.concat <$> extractIO (CAlg.asyncGzipFromFile testingFileNameGZ) r @?= "Hello World" removeFile testingFileNameGZ +case_asyncBzip2 :: IO ()+case_asyncBzip2 = do+ C.runConduitRes (CC.yieldMany ["Hello", " ", "World"] .| CAlg.asyncBzip2ToFile testingFileNameBZ2)+ r <- B.concat <$> extractIO (CAlg.asyncBzip2FromFile testingFileNameBZ2)+ r @?= "Hello World"+ removeFile testingFileNameBZ2 +case_asyncXz :: IO ()+#ifdef WINDOWS+case_asyncXz = assertError $ do+#else+case_asyncXz = do+#endif+ C.runConduitRes (CC.yieldMany ["Hello", " ", "World"] .| CAlg.asyncXzToFile testingFileNameXZ)+ r <- B.concat <$> extractIO (CAlg.asyncXzFromFile testingFileNameXZ)+ r @?= "Hello World"+ removeFile testingFileNameXZ++case_async_gzip_to_from :: IO () case_async_gzip_to_from = do let testdata = [0 :: Int .. 12] C.runConduitRes $@@ -135,15 +193,56 @@ removeFile testingFileNameGZ removeFile testingFileNameGZ2 +case_async_bzip2_to_from :: IO ()+case_async_bzip2_to_from = do+ let testdata = [0 :: Int .. 12]+ C.runConduitRes $+ CC.yieldMany testdata+ .| CL.map (B8.pack . (\n -> show n ++ "\n"))+ .| CAlg.asyncBzip2ToFile testingFileNameBZ2+ C.runConduitRes $+ CAlg.asyncBzip2FromFile testingFileNameBZ2+ .| CAlg.asyncBzip2ToFile testingFileNameBZ22+ shouldProduceIO testdata $+ CAlg.asyncBzip2FromFile testingFileNameBZ22+ .| CB.lines+ .| CL.map (read . B8.unpack)+ removeFile testingFileNameBZ2+ removeFile testingFileNameBZ22++case_async_xz_to_from :: IO ()+#ifdef WINDOWS+case_async_xz_to_from = assertError $ do+#else+case_async_xz_to_from = do+#endif+ let testdata = [0 :: Int .. 12]+ C.runConduitRes $+ CC.yieldMany testdata+ .| CL.map (B8.pack . (\n -> show n ++ "\n"))+ .| CAlg.asyncXzToFile testingFileNameXZ+ C.runConduitRes $+ CAlg.asyncXzFromFile testingFileNameXZ+ .| CAlg.asyncXzToFile testingFileNameXZ2+ shouldProduceIO testdata $+ CAlg.asyncXzFromFile testingFileNameXZ2+ .| CB.lines+ .| CL.map (read . B8.unpack)+ removeFile testingFileNameXZ+ removeFile testingFileNameXZ2++case_asyncFilterLines :: IO () case_asyncFilterLines = do vals <- extractIO (CC.yieldMany ["This is\nMy data\nBut"," sometimes","\nit is split,\n","in weird ways."] .| CAlg.asyncFilterLinesC 2 (B8.notElem ','))- (vals @?= ["This is", "My data", "But sometimes", "in weird ways."])+ vals @?= ["This is", "My data", "But sometimes", "in weird ways."] +case_asyncFilterLinesAllTrue :: IO () case_asyncFilterLinesAllTrue = do vals <- extractIO (CC.yieldMany ["This is\nMy data\nBut"," sometimes","\nit is split,\n","in weird ways."] .| CAlg.asyncFilterLinesC 2 (const True))- (vals @?= ["This is", "My data", "But sometimes", "it is split,", "in weird ways."])+ vals @?= ["This is", "My data", "But sometimes", "it is split,", "in weird ways."] +case_storableVector :: IO () case_storableVector = do let v = VS.fromList [0:: Int, 1, 2, 4, 6, 12] vals <- extractIO (CC.yieldMany [v,v,v] .| CAlg.writeStorableV .| CAlg.readStorableV 3)- (VS.concat vals @=? VS.concat [v,v,v])+ VS.concat vals @=? VS.concat [v,v,v]