conduit-algorithms 0.0.7.3 → 0.0.8.0
raw patch · 5 files changed
+47/−45 lines, 5 filesdep +exceptionsdep +unliftio-coredep ~conduitdep ~stm-conduit
Dependencies added: exceptions, unliftio-core
Dependency ranges changed: conduit, stm-conduit
Files
- ChangeLog +3/−0
- Data/Conduit/Algorithms.hs +12/−15
- Data/Conduit/Algorithms/Async.hs +20/−22
- conduit-algorithms.cabal +10/−6
- tests/Tests.hs +2/−2
ChangeLog view
@@ -1,3 +1,6 @@+Version 0.0.8.0 2018-03-12 by luispedro+ * Update for conduit 1.3 (issue #6)+ 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)
Data/Conduit/Algorithms.hs view
@@ -1,6 +1,6 @@ {-| Module : Data.Conduit.Algorithms-Copyright : 2013-2017 Luis Pedro Coelho+Copyright : 2013-2018 Luis Pedro Coelho License : MIT Maintainer : luis@luispedro.org @@ -80,10 +80,10 @@ mergeC :: (Ord a, Monad m) => [C.Source m a] -> C.Source m a mergeC [a] = a mergeC [a,b] = mergeC2 a b-mergeC cs = CI.ConduitM $ \rest -> let+mergeC cs = CI.ConduitT $ \rest -> let go [] = rest ()- go allc@(CI.HaveOutput c_next _ v:larger) =- CI.HaveOutput (norm1 c_next >>= go . insert1 larger) (finalizeAll allc) v+ go (CI.HaveOutput c_next v:larger) =+ CI.HaveOutput (norm1 c_next >>= go . insert1 larger) v go _ = error "This situation should have been impossible (mergeC/go)" insert1 larger CI.Done{} = larger insert1 larger c = insertBy compareHO c larger@@ -95,13 +95,10 @@ norm1 (CI.Leftover next ()) = norm1 next isHO CI.HaveOutput{} = True isHO _ = False- compareHO (CI.HaveOutput _ _ a) (CI.HaveOutput _ _ b) = compare a b+ compareHO (CI.HaveOutput _ a) (CI.HaveOutput _ b) = compare a b compareHO _ _ = error "This situation should have been impossible (mergeC/compareHO)"- finalizeAll [] = return ()- finalizeAll (CI.HaveOutput _ f _ : larger) = f >> finalizeAll larger- finalizeAll (_ :larger) = finalizeAll larger in do- let st = map (($ CI.Done) . CI.unConduitM) cs+ let st = map (($ CI.Done) . CI.unConduitT) cs st' <- mapM norm1 st go . sortBy compareHO . filter isHO $ st' @@ -110,12 +107,12 @@ -- -- See 'mergeC' mergeC2 :: (Ord a, Monad m) => C.Source m a -> C.Source m a -> C.Source m a-mergeC2 (CI.ConduitM s1) (CI.ConduitM s2) = CI.ConduitM $ \rest -> let- go right@(CI.HaveOutput s1' f1 v1) left@(CI.HaveOutput s2' f2 v2)- | v1 <= v2 = CI.HaveOutput (go s1' left) (f1 >> f2) v1- | otherwise = CI.HaveOutput (go right s2') (f1 >> f2) v2- go right@CI.Done{} (CI.HaveOutput s f v) = CI.HaveOutput (go right s) f v- go (CI.HaveOutput s f v) left@CI.Done{} = CI.HaveOutput (go s left) f v+mergeC2 (CI.ConduitT s1) (CI.ConduitT s2) = CI.ConduitT $ \rest -> let+ go right@(CI.HaveOutput s1' v1) left@(CI.HaveOutput s2' v2)+ | v1 <= v2 = CI.HaveOutput (go s1' left) v1+ | otherwise = CI.HaveOutput (go right s2') v2+ go right@CI.Done{} (CI.HaveOutput s v) = CI.HaveOutput (go right s) v+ go (CI.HaveOutput s v) left@CI.Done{} = CI.HaveOutput (go s left) v go CI.Done{} CI.Done{} = rest () go (CI.PipeM p) left = do next <- lift p
Data/Conduit/Algorithms/Async.hs view
@@ -1,6 +1,6 @@ {-| Module : Data.Conduit.Algorithms.Async-Copyright : 2013-2017 Luis Pedro Coelho+Copyright : 2013-2018 Luis Pedro Coelho License : MIT Maintainer : luis@luispedro.org @@ -50,10 +50,11 @@ import qualified Data.Sequence as Seq import Data.Sequence ((|>), ViewL(..)) import Data.Foldable (toList)-import Control.Monad (forM_) import Control.Monad.IO.Class (MonadIO, liftIO) import Control.Monad.Error.Class (MonadError(..))-import Control.Monad.Trans.Resource (MonadResource, MonadBaseControl)+import Control.Monad.IO.Unlift (MonadUnliftIO)+import Control.Monad.Trans.Resource (MonadResource)+import Control.Monad.Catch (MonadThrow) import Control.Exception (evaluate, displayException) import Control.DeepSeq (NFData, force) import System.IO.Error (mkIOError, userErrorType)@@ -121,7 +122,7 @@ yAll :: Seq.Seq (A.Async b) -> C.Conduit a m b yAll q = case Seq.viewl q of EmptyL -> return ()- v :< rest -> (liftIO (A.wait v) >>= yieldOrCleanup rest) >> yAll rest+ v :< rest -> (liftIO (A.wait v) >>= C.yield) >> yAll rest loop :: Seq.Seq (A.Async b) -> C.Conduit a m b loop q = C.await >>= \case@@ -129,11 +130,8 @@ Just v -> do v' <- sched v (r, q') <- liftIO $ retrieveResult q- yieldOrCleanup q' r+ C.yield r loop (q' |> v')- cleanup :: Seq.Seq (A.Async b) -> m ()- cleanup q = liftIO $ forM_ q A.cancel- yieldOrCleanup q = flip C.yieldOr (cleanup q) retrieveResult :: Seq.Seq (A.Async b) -> IO (b, Seq.Seq (A.Async b)) retrieveResult q@@ -188,7 +186,7 @@ -- writes the results to `h`. -- -- See also 'asyncGzipToFile'-asyncGzipTo :: forall m. (MonadIO m, MonadBaseControl IO m) => Handle -> C.Sink B.ByteString m ()+asyncGzipTo :: forall m. (MonadIO m, MonadUnliftIO m) => Handle -> C.Sink B.ByteString m () asyncGzipTo h = do let drain q = liftIO . C.runConduit $ CA.sourceTBQueue q@@ -203,7 +201,7 @@ -- performed in a separate thread. -- -- See also 'asyncGzipTo'-asyncGzipToFile :: forall m. (MonadResource m, MonadBaseControl IO m) => FilePath -> C.Sink B.ByteString m ()+asyncGzipToFile :: forall m. (MonadResource m, MonadUnliftIO m) => FilePath -> C.Sink B.ByteString m () asyncGzipToFile fname = C.bracketP (openFile fname WriteMode) hClose@@ -214,7 +212,7 @@ -- will probably be left at an undefined position in the file. -- -- See also 'asyncGzipFromFile'-asyncGzipFrom :: forall m. (MonadIO m, MonadResource m, MonadBaseControl IO m) => Handle -> C.Source m B.ByteString+asyncGzipFrom :: forall m. (MonadIO m, MonadResource m, MonadUnliftIO m) => Handle -> C.Source m B.ByteString asyncGzipFrom h = do let prod q = liftIO $ do C.runConduit $@@ -231,7 +229,7 @@ -- separate thread. -- -- See also 'asyncGzipFrom'-asyncGzipFromFile :: forall m. (MonadResource m, MonadBaseControl IO m) => FilePath -> C.Source m B.ByteString+asyncGzipFromFile :: forall m. (MonadResource m, MonadUnliftIO m) => FilePath -> C.Source m B.ByteString asyncGzipFromFile fname = C.bracketP (openFile fname ReadMode) hClose@@ -241,7 +239,7 @@ -- 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 :: forall m. (MonadIO m, MonadResource m, MonadUnliftIO m) => Handle -> C.Sink B.ByteString m () asyncBzip2To h = do let drain q = C.runConduit $ CA.sourceTBQueue q@@ -256,7 +254,7 @@ -- performed in a separate thread. -- -- See also 'asyncGzipTo'-asyncBzip2ToFile :: forall m. (MonadResource m, MonadBaseControl IO m) => FilePath -> C.Sink B.ByteString m ()+asyncBzip2ToFile :: forall m. (MonadResource m, MonadUnliftIO m) => FilePath -> C.Sink B.ByteString m () asyncBzip2ToFile fname = C.bracketP (openFile fname WriteMode) hClose@@ -267,7 +265,7 @@ -- 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 :: forall m. (MonadIO m, MonadResource m, MonadUnliftIO m) => Handle -> C.Source m B.ByteString asyncBzip2From h = do let prod q = do C.runConduit $@@ -282,7 +280,7 @@ -- separate thread. -- -- See also 'asyncGzipFrom'-asyncBzip2FromFile :: forall m. (MonadResource m, MonadBaseControl IO m) => FilePath -> C.Source m B.ByteString+asyncBzip2FromFile :: forall m. (MonadResource m, MonadUnliftIO m) => FilePath -> C.Source m B.ByteString asyncBzip2FromFile fname = C.bracketP (openFile fname ReadMode) hClose@@ -292,7 +290,7 @@ -- 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 :: forall m. (MonadIO m, MonadResource m, MonadUnliftIO m) => Handle -> C.Sink B.ByteString m () asyncXzTo h = do let drain q = C.runConduit $ CA.sourceTBQueue q@@ -311,7 +309,7 @@ -- performed in a separate thread. -- -- See also 'asyncGzipTo'-asyncXzToFile :: forall m. (MonadResource m, MonadBaseControl IO m) => FilePath -> C.Sink B.ByteString m ()+asyncXzToFile :: forall m. (MonadResource m, MonadUnliftIO m) => FilePath -> C.Sink B.ByteString m () asyncXzToFile fname = C.bracketP (openFile fname WriteMode) hClose@@ -322,7 +320,7 @@ -- 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 :: forall m. (MonadIO m, MonadResource m, MonadUnliftIO m, MonadThrow m) => Handle -> C.Source m B.ByteString asyncXzFrom h = do let oneGBmembuffer = Just $ 1024 ^ (3 :: Integer) prod q = do@@ -343,7 +341,7 @@ -- separate thread. -- -- See also 'asyncXzFrom'-asyncXzFromFile :: forall m. (MonadResource m, MonadBaseControl IO m) => FilePath -> C.Source m B.ByteString+asyncXzFromFile :: forall m. (MonadResource m, MonadUnliftIO m, MonadThrow m) => FilePath -> C.Source m B.ByteString asyncXzFromFile fname = C.bracketP (openFile fname ReadMode) hClose@@ -355,14 +353,14 @@ -- On Windows, attempting to read from a bzip2 file, results in 'error'. -- -- For the case of gzip, 'asyncGzipFromFile' is used.-conduitPossiblyCompressedFile :: (MonadBaseControl IO m, MonadResource m) => FilePath -> C.Source m B.ByteString+conduitPossiblyCompressedFile :: (MonadUnliftIO m, MonadResource m, MonadThrow m) => FilePath -> C.Source m B.ByteString conduitPossiblyCompressedFile fname | ".gz" `isSuffixOf` fname = asyncGzipFromFile fname | ".xz" `isSuffixOf` fname = asyncXzFromFile fname | ".bz2" `isSuffixOf` fname = asyncBzip2FromFile fname | otherwise = C.sourceFile fname -conduitPossiblyCompressedToFile :: (MonadBaseControl IO m, MonadResource m) => FilePath -> C.Sink B.ByteString m ()+conduitPossiblyCompressedToFile :: (MonadUnliftIO m, MonadResource m) => FilePath -> C.Sink B.ByteString m () conduitPossiblyCompressedToFile fname | ".gz" `isSuffixOf` fname = asyncGzipToFile fname | ".xz" `isSuffixOf` fname = asyncXzToFile fname
conduit-algorithms.cabal view
@@ -2,10 +2,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: 3a46059f9e3513ce4b3dce9af08750249e1263b0cfac24b7294a5d5269f6344e+-- hash: e408a4353cbeef37b0ef19abf3d1c40d0939cc1dc51ce2f917d36e620eb29a89 name: conduit-algorithms-version: 0.0.7.3+version: 0.0.8.0 synopsis: Conduit-based algorithms description: Algorithms on Conduits, including higher level asynchronous processing and some other utilities. category: Conduit@@ -34,18 +34,20 @@ , base >4.8 && <5 , bytestring , bzlib-conduit- , conduit >=1.0.11+ , conduit >=1.3 , conduit-combinators >=1.1.2 , conduit-extra , containers , deepseq+ , exceptions , monad-control , mtl , resourcet , stm- , stm-conduit >=2.7+ , stm-conduit >=4.0 , streaming-commons , transformers+ , unliftio-core , vector if os(windows) cpp-options: -DWINDOWS@@ -73,23 +75,25 @@ , base >4.8 && <5 , bytestring , bzlib-conduit- , conduit >=1.0.11+ , conduit >=1.3 , conduit-algorithms , conduit-combinators >=1.1.2 , conduit-extra , containers , deepseq , directory+ , exceptions , monad-control , mtl , resourcet , stm- , stm-conduit >=2.7+ , stm-conduit >=4.0 , streaming-commons , test-framework , test-framework-hunit , test-framework-th , transformers+ , unliftio-core , vector if os(windows) cpp-options: -DWINDOWS
tests/Tests.hs view
@@ -22,7 +22,7 @@ import Control.Exception (catch, ErrorCall) import Control.Monad (forM_) import Control.Monad.Trans.Resource.Internal (ResourceT)-import Control.Monad.Trans.Control (MonadBaseControl)+import Control.Monad.IO.Unlift (MonadUnliftIO) import qualified Data.Conduit.Algorithms as CAlg import qualified Data.Conduit.Algorithms.Storable as CAlg@@ -49,7 +49,7 @@ 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 :: MonadUnliftIO 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