diff --git a/ChangeLog b/ChangeLog --- a/ChangeLog +++ b/ChangeLog @@ -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 diff --git a/Data/Conduit/Algorithms/Async.hs b/Data/Conduit/Algorithms/Async.hs --- a/Data/Conduit/Algorithms/Async.hs +++ b/Data/Conduit/Algorithms/Async.hs @@ -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 diff --git a/conduit-algorithms.cabal b/conduit-algorithms.cabal --- a/conduit-algorithms.cabal +++ b/conduit-algorithms.cabal @@ -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 diff --git a/tests/Tests.hs b/tests/Tests.hs --- a/tests/Tests.hs +++ b/tests/Tests.hs @@ -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]