conduit-algorithms 0.0.7.0 → 0.0.7.1

raw patch · 9 files changed

+759/−168 lines, 9 filesdep +conduit-algorithmsPVP: minor bump suggested

API additions: PVP suggests at least a minor version bump

Dependencies added: conduit-algorithms

API changes (from Hackage documentation)

Files

• ChangeLog +3/−0
• Data/Conduit/Algorithms.hs +130/−0
• Data/Conduit/Algorithms/Async.hs +246/−0
• Data/Conduit/Algorithms/Async/ByteString.hs +70/−0
• Data/Conduit/Algorithms/Storable.hs +68/−0
• Data/Conduit/Algorithms/Tests.hs +0/−149
• Data/Conduit/Algorithms/Utils.hs +63/−0
• conduit-algorithms.cabal +30/−19
• tests/Tests.hs +149/−0

@@ -1,3 +1,6 @@+Version 0.0.7.1 2018-01-18 by luispedro+	* Fix testing in newer stack/cabal/hpack (issue #3)+ Version 0.0.7.0 2018-01-17 by luispedro 	* Add unorderedAsyncMapC 	* Add Data.Conduit.Algorithms.Async.ByteString module

@@ -0,0 +1,130 @@+{-|+Module      : Data.Conduit.Algorithms+Copyright   : 2013-2017 Luis Pedro Coelho+License     : MIT+Maintainer  : luis@luispedro.org++Simple algorithms packaged as Conduits+-}+++{-# LANGUAGE Rank2Types #-}+module Data.Conduit.Algorithms+    ( uniqueOnC+    , uniqueC+    , removeRepeatsC+    , mergeC+    , mergeC2+    ) where++import qualified Data.Conduit as C+import qualified Data.Conduit.Internal as CI+import qualified Data.Set as S+import           Data.List (sortBy, insertBy)+import           Control.Monad.Trans.Class (lift)++import           Data.Conduit.Algorithms.Utils (awaitJust)+++-- | Unique conduit.+--+-- For each element, it checks its key (using the @a -> b@ key function) and+-- yields it if it has not seen it before.+--+-- Note that this conduit /does not/ assume that the input is sorted. Instead+-- it uses a 'Data.Set' to store previously seen elements. Thus, memory usage+-- is O(N) and time is O(N log N). If the input is sorted, you can use+-- 'removeRepeatsC'+uniqueOnC :: (Ord b, Monad m) => (a -> b) -> C.Conduit a m a+uniqueOnC f = checkU (S.empty :: S.Set b)+    where+        checkU cur = awaitJust $ \val ->+                        if f val `S.member` cur+                            then checkU cur+                            else do+                                C.yield val+                                checkU (S.insert (f val) cur)+-- | Unique conduit+--+-- See 'uniqueOnC' and 'removeRepeatsC'+uniqueC :: (Ord a, Monad m) => C.Conduit a m a+uniqueC = uniqueOnC id++-- | Removes repeated elements+--+-- @+--  yieldMany [0, 0, 1, 1, 1, 2, 2, 0] .| removeRepeatsC .| consume+-- @+--+-- is equivalent to @[0, 1, 2, 0]@+--+-- See 'uniqueC' and 'uniqueOnC'+removeRepeatsC :: (Eq a, Monad m) => C.Conduit a m a+removeRepeatsC = awaitJust removeRepeatsC'+    where+        removeRepeatsC' prev = C.await >>= \case+                                        Nothing -> C.yield prev+                                        Just next+                                            | next == prev -> removeRepeatsC' prev+                                            | otherwise -> do+                                                        C.yield prev+                                                        removeRepeatsC' next+++-- | Merge a list of sorted sources to produce a single (sorted) source+--+-- This takes a list of sorted sources and produces a 'C.Source' which outputs+-- all elements in sorted order.+--+-- See 'mergeC2'+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+        go [] = rest ()+        go allc@(CI.HaveOutput c_next _ v:larger) =+            CI.HaveOutput (norm1 c_next >>= go . insert1 larger) (finalizeAll allc) v+        go _ = error "This situation should have been impossible (mergeC/go)"+        insert1 larger CI.Done{} = larger+        insert1 larger c = insertBy compareHO c larger+        norm1 :: Monad m => CI.Pipe () i o () m () -> CI.Pipe () i o () m (CI.Pipe () i o () m ())+        norm1 c@CI.HaveOutput{} = return c+        norm1 c@CI.Done{} = return c+        norm1 (CI.PipeM p) = lift p >>= norm1+        norm1 (CI.NeedInput _ next) = norm1 (next ())+        norm1 (CI.Leftover next ()) = norm1 next+        isHO CI.HaveOutput{} = True+        isHO _ = False+        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+        st' <- mapM norm1 st+        go . sortBy compareHO . filter isHO $ st'+++-- | Take two sorted sources and merge them.+--+-- 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+        go CI.Done{} CI.Done{} = rest ()+        go (CI.PipeM p) left = do+            next <- lift p+            go next left+        go right (CI.PipeM p) = do+            next <- lift p+            go right next+        go (CI.NeedInput _ next) left = go (next ()) left+        go right (CI.NeedInput _ next) = go right (next ())+        go (CI.Leftover next ()) left = go next left+        go right (CI.Leftover next ()) = go right next+    in go (s1 CI.Done) (s2 CI.Done)

@@ -0,0 +1,246 @@+{-|+Module      : Data.Conduit.Algorithms.Async+Copyright   : 2013-2017 Luis Pedro Coelho+License     : MIT+Maintainer  : luis@luispedro.org++Higher level async processing interfaces.+-}+{-# LANGUAGE ScopedTypeVariables, FlexibleContexts, CPP, TupleSections #-}++module Data.Conduit.Algorithms.Async+    ( conduitPossiblyCompressedFile+    , asyncMapC+    , asyncMapEitherC+    , asyncGzipTo+    , asyncGzipToFile+    , asyncGzipFrom+    , asyncGzipFromFile+    , unorderedAsyncMapC+    ) where+++import qualified Data.ByteString as B+import qualified Control.Concurrent.Async as A+import qualified Control.Concurrent.STM.TBQueue as TQ+import           Control.Concurrent.STM (atomically)++import qualified Data.Conduit.Combinators as C+import qualified Data.Conduit.Async as CA+import qualified Data.Conduit.TQueue as CA+import qualified Data.Conduit.List as CL+import qualified Data.Conduit.Zlib as CZ+import qualified Data.Conduit.Lzma as CX+#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 ((.|))++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.Exception (evaluate)+import           Control.DeepSeq+import           System.IO+import           Data.List (isSuffixOf)+import           Data.Conduit.Algorithms.Utils (awaitJust)++++-- | This is like 'Data.Conduit.List.map', except that each element is processed+-- in a separate thread (up to 'maxThreads' can be queued up at any one time).+-- Results are evaluated to normal form (not weak-head normal form!, i.e., the+-- structure is deeply evaluated) to ensure that the computation is fully+-- evaluated in the worker thread.+--+-- Note that there is some overhead in threading. It is often a good idea to+-- build larger chunks of input before passing it to 'asyncMapC' to amortize+-- the costs. That is, when @f@ is not a lot of work, instead of @asyncMapC f@,+-- it is sometimes better to do+--+-- @+--    CC.conduitVector 4096 .| asyncMapC (V.map f) .| CC.concat+-- @+--+-- where @CC@ refers to 'Data.Conduit.Combinators'+--+-- See 'unorderedAsyncMapC'+asyncMapC :: forall a m b . (MonadIO m, NFData b) =>+                    Int -- ^ Maximum number of worker threads+                    -> (a -> b) -- ^ Function to execute+                    -> C.Conduit a m b+asyncMapC = asyncMapCHelper True++-- | A version of 'asyncMapC' which can reorder results in the stream+--+-- If the order of the results is not important, this function can lead to a+-- better use of resources if some of the chunks take longer to complete.+--+-- See 'asyncMapC'+unorderedAsyncMapC :: forall a m b . (MonadIO m, NFData b) =>+                    Int -- ^ Maximum number of worker threads+                    -> (a -> b) -- ^ Function to execute+                    -> C.Conduit a m b+unorderedAsyncMapC = asyncMapCHelper False++asyncMapCHelper  :: forall a m b . (MonadIO m, NFData b) =>+                    Bool+                    -> Int -- ^ Maximum number of worker threads+                    -> (a -> b) -- ^ Function to execute+                    -> C.Conduit a m b+asyncMapCHelper isSynchronous maxThreads f = initLoop (0 :: Int) (Seq.empty :: Seq.Seq (A.Async b))+    where+        initLoop :: Int -> Seq.Seq (A.Async b) -> C.Conduit a m b+        initLoop size q+            | size == maxThreads = loop q+            | otherwise = C.await >>= \case+                Nothing -> yAll q+                Just v -> do+                        v' <- sched v+                        initLoop (size + 1) (q |> v')+        sched :: a -> C.ConduitM a b m (A.Async b)+        sched v = liftIO . A.async . evaluate . force $ f v++        -- | yield all+        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++        loop :: Seq.Seq (A.Async b) -> C.Conduit a m b+        loop q = C.await >>= \case+                Nothing -> yAll q+                Just v -> do+                    v' <- sched v+                    (r, q') <- liftIO $ retrieveResult q+                    yieldOrCleanup q' 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+            | isSynchronous = case Seq.viewl q of+                        (r :< rest) -> (, rest) <$> A.wait r+                        _ -> error "Impossible situation"+            | otherwise = do+                (k, r) <- liftIO (A.waitAny (toList q))+                return (r, Seq.filter (/= k) q)+++-- | 'asyncMapC' with error handling. The inner function can now return an+-- error (as a 'Left'). When the first error is seen, it 'throwError's in the+-- main monad. Note that 'f' may be evaluated for arguments beyond the first+-- error (as some threads may be running in the background and already+-- processing elements after the first error).+--+-- See 'asyncMapC'+asyncMapEitherC :: forall a m b e . (MonadIO m, NFData b, NFData e, MonadError e m) => Int -> (a -> Either e b) -> C.Conduit a m b+asyncMapEitherC maxThreads f = asyncMapC maxThreads f .| (C.awaitForever $ \case+                                Right v -> C.yield v+                                Left err -> throwError err)+++-- | concatenates input into larger chunks and yields it. Its indended use is+-- to build up larger blocks from smaller ones so that they can be sent across+-- thread barriers with little overhead.+--+-- the chunkSize parameter is a hint, not an exact element. In particular,+-- larger chunks are not split up and smaller chunks can be yielded too.+bsConcatTo :: MonadIO m => Int -- ^ chunk hint+                            -> C.Conduit B.ByteString m [B.ByteString]+bsConcatTo chunkSize = awaitJust start+    where+        start v+            | B.length v >= chunkSize = C.yield [v] >> bsConcatTo chunkSize+            | otherwise = continue [v] (B.length v)+        continue chunks s = C.await >>= \case+            Nothing -> C.yield chunks+            Just v+                | B.length v + s > chunkSize -> C.yield chunks >> start v+                | otherwise -> continue (v:chunks) (s + B.length v)++untilNothing :: forall m i. (Monad m) => C.Conduit (Maybe i) m i+untilNothing = C.await >>= \case+    Just (Just val) -> do+        C.yield val+        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, MonadBaseControl IO m) => Handle -> C.Sink B.ByteString m ()+asyncGzipTo h = do+    let drain q = liftIO . C.runConduit $+                CA.sourceTBQueue q+                    .| untilNothing+                    .| CL.map (B.concat . reverse)+                    .| CZ.gzip+                    .| 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'+asyncGzipToFile :: forall m. (MonadResource m, MonadBaseControl IO m) => FilePath -> C.Sink B.ByteString m ()+asyncGzipToFile fname = C.bracketP+    (openFile fname WriteMode)+    hClose+    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+-- 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 h = do+    let prod q = liftIO $ do+                    C.runConduit $+                        C.sourceHandle h+                            .| CZ.multiple CZ.ungzip+                            .| CL.map Just+                            .| CA.sinkTBQueue q+                    atomically (TQ.writeTBQueue q Nothing)+    CA.gatherFrom 8 prod+        .| untilNothing++-- | Open and read a gzip file with the uncompression being performed in a+-- separate thread.+--+-- See also 'asyncGzipFrom'+asyncGzipFromFile :: forall m. (MonadResource m, MonadBaseControl IO m) => FilePath -> C.Source m B.ByteString+asyncGzipFromFile fname = C.bracketP+    (openFile fname ReadMode)+    hClose+    asyncGzipFrom++-- | If the filename indicates a gzipped file (or, on Unix, also a bz2 file),+-- then it reads it and uncompresses it.+--+-- 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 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+    | otherwise = C.sourceFile fname+        where oneGBmembuffer = Just $ 1024 * 1024 * 1024+

@@ -0,0 +1,70 @@+{-|+Module      : Data.Conduit.Algorithms.Async.ByteString+Copyright   : 2018 Luis Pedro Coelho+License     : MIT+Maintainer  : luis@luispedro.org++Higher level async processing interfaces.+-}+{-# LANGUAGE ScopedTypeVariables, FlexibleContexts, TupleSections #-}++module Data.Conduit.Algorithms.Async.ByteString+    ( asyncMapLineGroupsC+    , asyncFilterLinesC+    ) where+++import qualified Data.ByteString as B+import qualified Data.ByteString.Lazy as BL+import qualified Data.Conduit.Algorithms.Async as CAlg+import qualified Data.Conduit.List as CL+import qualified Data.Conduit as C+import           Data.Conduit ((.|))++import           Control.Monad (unless)+import           Control.Monad.IO.Class (MonadIO)+import           Control.DeepSeq+++-- | Apply a function to groups of lines+--+-- Note that this is much more efficient than the (more or less equivalent,+-- except that the intermediate lists can be of varying sizes):+--+-- @+--      CB.lines .| CC.conduitVector N .| CAlg.asyncMapC nthreads (f . V.toList)+-- @+--+-- The reason being that splitting into lines then becomes the bottleneck and+-- processing a single line is typically a tiny chunk of work so that the+-- threading overhead overwhelms the advantage of using multiple cores.+-- Instead, 'asyncMapLineGroupsC' will pass big chunks to the worker thread and+-- perform most of the line splitting _in the worker thread_.+--+-- Only Unix-style ASCII lines are supported (splitting at Bytes with value+-- 10, i.e., \n). When Windows lines (\r\n) are passed to this function, this+-- results in each element having an extra \r at the end.+asyncMapLineGroupsC :: (MonadIO m, NFData a) => Int -> ([B.ByteString] -> a) -> C.Conduit B.ByteString m a+asyncMapLineGroupsC nthreads f = breakAtLineBoundary .| CAlg.asyncMapC nthreads (f . asLines)+    where+        asLines :: BL.ByteString -> [B.ByteString]+        asLines = fmap BL.toStrict . BL.split 10++        -- The purpose is to break input blocks at a line boundary+        breakAtLineBoundary :: Monad m => C.Conduit B.ByteString m BL.ByteString+        breakAtLineBoundary = continue BL.empty+        continue prev = C.await >>= \case+                    Nothing -> unless (BL.null prev) $+                                C.yield prev+                    Just n -> case B.elemIndexEnd 10 n of+                        Nothing -> continue (BL.append prev (BL.fromStrict n))+                        Just p -> do+                            let (first, rest) = B.splitAt p n+                            C.yield (BL.append prev (BL.fromStrict first))+                            continue (BL.fromStrict $ B.drop 1 rest) -- skip \n char++-- | Filter lines using multiple threads+asyncFilterLinesC :: MonadIO m => Int -> (B.ByteString -> Bool) -> C.Conduit B.ByteString m B.ByteString+asyncFilterLinesC n f = asyncMapLineGroupsC n (filter f) .| CL.concat+{-# INLINE asyncFilterLinesC #-}+

@@ -0,0 +1,68 @@+{-|+Module      : Data.Conduit.Algorithms.Storable+Copyright   : 2018 Luis Pedro Coelho+License     : MIT+Maintainer  : luis@luispedro.org++Higher level async processing interfaces.+-}+{-# LANGUAGE FlexibleContexts, ScopedTypeVariables #-}+module Data.Conduit.Algorithms.Storable+    ( writeStorableV+    , readStorableV+    ) where++import qualified Data.ByteString as B+import qualified Data.ByteString.Unsafe as BU+import qualified Data.Vector.Storable as VS+import qualified Data.Vector.Storable.Mutable as VSM+import Control.Monad.IO.Class++import Foreign.Ptr+import Foreign.Marshal.Utils+import Foreign.Storable+import Control.Monad (when)++import qualified Data.Conduit.List as CL+import qualified Data.Conduit.Combinators as CC+import qualified Data.Conduit as C+import           Data.Conduit ((.|))++-- | write a Storable vector+--+-- This uses the same format as in-memory+--+-- See |readStorableV|+writeStorableV :: forall m a. (MonadIO m, Monad m, Storable a) => C.Conduit (VS.Vector a) m B.ByteString+writeStorableV = CL.mapM (liftIO. encodeStorable')+    where+        encodeStorable' :: Storable a => VS.Vector a -> IO B.ByteString+        encodeStorable' v' = VS.unsafeWith v' $ \p ->+                                    B.packCStringLen (castPtr p, VS.length v' * (sizeOf (undefined :: a)))+++-- | read a Storable vector+--+-- This expects the same format as the in-memory vector+--+-- See |writeStorableV|+readStorableV :: forall m a. (MonadIO m, Storable a) => Int -> C.ConduitM B.ByteString (VS.Vector a) m ()+readStorableV nelems = CC.chunksOfE blockBytes .| parseBlocks+    where+        blockBytes = nelems * (sizeOf a')+        a' :: a+        a' = undefined+++        parseBlocks :: MonadIO m => C.Conduit B.ByteString m (VS.Vector a)+        parseBlocks = C.awaitForever $ \bs -> do+            let (n,rest) = B.length bs `divMod` sizeOf a'+            r <- liftIO $ do+                v <- VSM.new n+                BU.unsafeUseAsCStringLen bs $ \(p, _) ->+                    VSM.unsafeWith v $ \vp ->+                        moveBytes (castPtr vp) p (n * sizeOf a')+                VS.unsafeFreeze v+            C.yield r+            when (rest > 0) $ do+                C.leftover (B.drop (n * sizeOf a') bs)

@@ -1,149 +0,0 @@-{- Copyright 2017-2018 Luis Pedro Coelho- - License: MIT- -}-{-# LANGUAGE TemplateHaskell, QuasiQuotes, FlexibleContexts, OverloadedStrings #-}-module Main where--import Test.Framework.TH-import Test.HUnit-import Test.Framework.Providers.HUnit--import qualified Data.ByteString as B-import qualified Data.ByteString.Char8 as B8-import qualified Data.Conduit as C-import qualified Data.Conduit.Combinators as CC-import qualified Data.Conduit.Binary as CB-import qualified Data.Conduit.List as CL-import qualified Data.Vector.Storable as VS-import           Data.Conduit ((.|))-import           Data.List (sort)-import           System.Directory (removeFile)-import           Control.Monad (forM_)--import qualified Data.Conduit.Algorithms as CAlg-import qualified Data.Conduit.Algorithms.Storable as CAlg-import qualified Data.Conduit.Algorithms.Utils as CAlg-import qualified Data.Conduit.Algorithms.Async as CAlg-import qualified Data.Conduit.Algorithms.Async.ByteString as CAlg--main :: IO ()-main = $(defaultMainGenerator)--testingFileNameGZ :: FilePath-testingFileNameGZ = "file_just_for_testing_delete_me_please.gz"-testingFileNameGZ2 :: FilePath-testingFileNameGZ2 = "file_just_for_testing_delete_me_please_2.gz"--extract c = C.runConduitPure (c .| CC.sinkList)--extractIO c = C.runConduitRes (c .| CC.sinkList)--shouldProduce values cond = extract cond @?= values-shouldProduceIO values cond = do-    p <- extractIO cond-    p @?= values--case_uniqueC = extract (CC.yieldMany [1,2,3,1,1,2,3] .| CAlg.uniqueC) @=? [1,2,3 :: Int]-case_mergeC = shouldProduce expected $-                            CAlg.mergeC-                                [ CC.yieldMany i1-                                , CC.yieldMany i2-                                , CC.yieldMany i3-                                , CC.yieldMany i3-                                ]-    where-        expected = sort (concat [i1, i2, i3, i3])-        i1 = [ 1, 2, 4 :: Int]-        i2 = [ 1, 4, 4, 5]-        i3 = [-1, 0, 7]--case_mergeCmonad = shouldProduce expected $-                            CAlg.mergeC-                                [ mYield i1-                                , mYield i2-                                , mYield i3-                                ]-    where-        expected = sort (concat [i1, i2, i3])-        mYield lst = do-            let lst' = map return lst-            forM_ lst' $ \elem -> do-                elem' <- elem-                C.yield elem'-        i1 = [ 0, 2, 4 :: Int]-        i2 = [ 1, 3, 4, 5]-        i3 = [-1, 0, 7]---case_mergeC2 = shouldProduce [0, 1, 1, 2, 3, 5 :: Int] $-                            CAlg.mergeC2-                                (CC.yieldMany [0, 1, 2])-                                (CC.yieldMany [1, 3, 5])--case_mergeC2same = shouldProduce [0, 0, 1, 1, 2, 2 :: Int] $-                            CAlg.mergeC2-                                (CC.yieldMany [0, 1, 2])-                                (CC.yieldMany [0, 1, 2])--case_mergeC2monad = shouldProduce [0, 1, 2, 2, 3, 4 :: Int] $ do-                            CAlg.mergeC2-                                (CC.yieldMany [0, 2])-                                (CC.yieldMany [1, 2])-                            CC.yieldMany [3]-                            CC.yieldMany [4]--case_groupC = shouldProduce [[0,1,2], [3,4,5], [6,7,8], [9, 10 :: Int]] $-                            CC.yieldMany [0..10] .| CAlg.groupC 3--case_enumerateC = shouldProduce [(0,'z'), (1,'o'), (2,'t')] $-                            CC.yieldMany ("zot" :: [Char]) .| CAlg.enumerateC--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])--case_unorderedAsyncMapC :: IO ()-case_unorderedAsyncMapC = do-    vals <- extractIO (CC.yieldMany [0..10] .| CAlg.unorderedAsyncMapC 3 (+ (1:: Int)))-    (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 @?= "Hello World"-    removeFile testingFileNameGZ---case_async_gzip_to_from = do-    let testdata = [0 :: Int .. 12]-    C.runConduitRes $-        CC.yieldMany testdata-            .| CL.map (B8.pack . (\n -> show n ++ "\n"))-            .| CAlg.asyncGzipToFile testingFileNameGZ-    C.runConduitRes $-        CAlg.asyncGzipFromFile testingFileNameGZ-        .| CAlg.asyncGzipToFile testingFileNameGZ2-    shouldProduceIO testdata $-        CAlg.asyncGzipFromFile testingFileNameGZ2-            .| CB.lines-            .| CL.map (read . B8.unpack)-    removeFile testingFileNameGZ-    removeFile testingFileNameGZ2--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."])--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."])--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])

@@ -0,0 +1,63 @@+{-|+Module      : Data.Conduit.Algorithms.Utils+Copyright   : 2013-2017 Luis Pedro Coelho+License     : MIT+Maintainer  : luis@luispedro.org++A few miscellaneous conduit utils+-}+module Data.Conduit.Algorithms.Utils+    ( awaitJust+    , enumerateC+    , groupC+    ) where++import qualified Data.Conduit as C+import           Data.Maybe (maybe)+import           Control.Monad (unless)++-- | Act on the next input (do nothing if no input). @awaitJust f@ is equivalent to+--+--+-- @ do+--      next <- C.await+--      case next of+--          Just val -> f val+--          Nothing -> return ()+-- @+--+-- This is a simple utility adapted from+-- http://neilmitchell.blogspot.de/2015/07/thoughts-on-conduits.html+awaitJust :: Monad m => (a -> C.Conduit a m b) -> C.Conduit a m b+awaitJust f = C.await >>= maybe (return ()) f+{-# INLINE awaitJust #-}++-- | Conduit analogue to Python's enumerate function+enumerateC :: Monad m => C.Conduit a m (Int, a)+enumerateC = enumerateC' 0+    where+        enumerateC' !i = awaitJust $ \v -> do+                                        C.yield (i, v)+                                        enumerateC' (i + 1)+{-# INLINE enumerateC #-}++-- | groupC yields the input as groups of 'n' elements. If the input is not a+-- multiple of 'n', the last element will be incomplete+--+-- Example:+--+-- @+--      CC.yieldMany [0..10] .| groupC 3 .| CC.consumeList+-- @+--+-- results in @[ [0,1,2], [3,4,5], [6,7,8], [9, 10] ]@+--+-- This function is deprecated; use 'Data.Conduit.List.chunksOf'+groupC :: (Monad m) => Int -> C.Conduit a m [a]+groupC n = loop n []+    where+        loop 0 ps = C.yield (reverse ps) >> loop n []+        loop c ps = C.await >>= \case+            Nothing -> unless (null ps) $ C.yield (reverse ps)+            Just p -> loop (c-1) (p:ps)+

@@ -1,9 +1,11 @@--- This file has been generated from package.yaml by hpack version 0.18.1.+-- This file has been generated from package.yaml by hpack version 0.20.0. -- -- see: https://github.com/sol/hpack+--+-- hash: af773895c6365b46822705a9fd57b521ba0e8692c5053f7a6eba56b6cc57b07b  name:           conduit-algorithms-version:        0.0.7.0+version:        0.0.7.1 synopsis:       Conduit-based algorithms description:    Algorithms on Conduits, including higher level asynchronous processing and some other utilities. category:       Conduit@@ -28,12 +30,12 @@   default-extensions: BangPatterns OverloadedStrings LambdaCase TupleSections   ghc-options: -Wall   build-depends:-      base > 4.8 && < 5-    , async+      async+    , base >4.8 && <5     , bytestring     , bzlib-conduit-    , conduit >= 1.0-    , conduit-combinators >= 1.1.2+    , conduit >=1.0+    , conduit-combinators >=1.1.2     , conduit-extra     , containers     , deepseq@@ -41,38 +43,47 @@     , mtl     , resourcet     , stm-    , stm-conduit >= 2.7-    , vector+    , stm-conduit >=2.7     , transformers-  other-modules:-      Paths_conduit_algorithms+    , vector+  exposed-modules:+      Data.Conduit.Algorithms+      Data.Conduit.Algorithms.Utils+      Data.Conduit.Algorithms.Async+      Data.Conduit.Algorithms.Async.ByteString+      Data.Conduit.Algorithms.Storable   default-language: Haskell2010  test-suite conduit-algorithms-test   type: exitcode-stdio-1.0-  main-is: Data/Conduit/Algorithms/Tests.hs+  main-is: Tests.hs+  hs-source-dirs:+      ./tests   default-extensions: BangPatterns OverloadedStrings LambdaCase TupleSections   ghc-options: -Wall   build-depends:-      base > 4.8 && < 5+      HUnit     , async+    , base >4.8 && <5     , bytestring     , bzlib-conduit-    , conduit >= 1.0-    , conduit-combinators >= 1.1.2+    , conduit >=1.0+    , conduit-algorithms+    , conduit-combinators >=1.1.2     , conduit-extra     , containers     , deepseq+    , directory     , lzma-conduit     , mtl     , resourcet     , stm-    , stm-conduit >= 2.7-    , vector-    , transformers-    , directory-    , HUnit+    , stm-conduit >=2.7     , test-framework     , test-framework-hunit     , test-framework-th+    , transformers+    , vector+  other-modules:+      Paths_conduit_algorithms   default-language: Haskell2010

@@ -0,0 +1,149 @@+{- Copyright 2017-2018 Luis Pedro Coelho+ - License: MIT+ -}+{-# LANGUAGE TemplateHaskell, QuasiQuotes, FlexibleContexts, OverloadedStrings #-}+module Main where++import Test.Framework.TH+import Test.HUnit+import Test.Framework.Providers.HUnit++import qualified Data.ByteString as B+import qualified Data.ByteString.Char8 as B8+import qualified Data.Conduit as C+import qualified Data.Conduit.Combinators as CC+import qualified Data.Conduit.Binary as CB+import qualified Data.Conduit.List as CL+import qualified Data.Vector.Storable as VS+import           Data.Conduit ((.|))+import           Data.List (sort)+import           System.Directory (removeFile)+import           Control.Monad (forM_)++import qualified Data.Conduit.Algorithms as CAlg+import qualified Data.Conduit.Algorithms.Storable as CAlg+import qualified Data.Conduit.Algorithms.Utils as CAlg+import qualified Data.Conduit.Algorithms.Async as CAlg+import qualified Data.Conduit.Algorithms.Async.ByteString as CAlg++main :: IO ()+main = $(defaultMainGenerator)++testingFileNameGZ :: FilePath+testingFileNameGZ = "file_just_for_testing_delete_me_please.gz"+testingFileNameGZ2 :: FilePath+testingFileNameGZ2 = "file_just_for_testing_delete_me_please_2.gz"++extract c = C.runConduitPure (c .| CC.sinkList)++extractIO c = C.runConduitRes (c .| CC.sinkList)++shouldProduce values cond = extract cond @?= values+shouldProduceIO values cond = do+    p <- extractIO cond+    p @?= values++case_uniqueC = extract (CC.yieldMany [1,2,3,1,1,2,3] .| CAlg.uniqueC) @=? [1,2,3 :: Int]+case_mergeC = shouldProduce expected $+                            CAlg.mergeC+                                [ CC.yieldMany i1+                                , CC.yieldMany i2+                                , CC.yieldMany i3+                                , CC.yieldMany i3+                                ]+    where+        expected = sort (concat [i1, i2, i3, i3])+        i1 = [ 1, 2, 4 :: Int]+        i2 = [ 1, 4, 4, 5]+        i3 = [-1, 0, 7]++case_mergeCmonad = shouldProduce expected $+                            CAlg.mergeC+                                [ mYield i1+                                , mYield i2+                                , mYield i3+                                ]+    where+        expected = sort (concat [i1, i2, i3])+        mYield lst = do+            let lst' = map return lst+            forM_ lst' $ \elem -> do+                elem' <- elem+                C.yield elem'+        i1 = [ 0, 2, 4 :: Int]+        i2 = [ 1, 3, 4, 5]+        i3 = [-1, 0, 7]+++case_mergeC2 = shouldProduce [0, 1, 1, 2, 3, 5 :: Int] $+                            CAlg.mergeC2+                                (CC.yieldMany [0, 1, 2])+                                (CC.yieldMany [1, 3, 5])++case_mergeC2same = shouldProduce [0, 0, 1, 1, 2, 2 :: Int] $+                            CAlg.mergeC2+                                (CC.yieldMany [0, 1, 2])+                                (CC.yieldMany [0, 1, 2])++case_mergeC2monad = shouldProduce [0, 1, 2, 2, 3, 4 :: Int] $ do+                            CAlg.mergeC2+                                (CC.yieldMany [0, 2])+                                (CC.yieldMany [1, 2])+                            CC.yieldMany [3]+                            CC.yieldMany [4]++case_groupC = shouldProduce [[0,1,2], [3,4,5], [6,7,8], [9, 10 :: Int]] $+                            CC.yieldMany [0..10] .| CAlg.groupC 3++case_enumerateC = shouldProduce [(0,'z'), (1,'o'), (2,'t')] $+                            CC.yieldMany ("zot" :: [Char]) .| CAlg.enumerateC++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])++case_unorderedAsyncMapC :: IO ()+case_unorderedAsyncMapC = do+    vals <- extractIO (CC.yieldMany [0..10] .| CAlg.unorderedAsyncMapC 3 (+ (1:: Int)))+    (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 @?= "Hello World"+    removeFile testingFileNameGZ+++case_async_gzip_to_from = do+    let testdata = [0 :: Int .. 12]+    C.runConduitRes $+        CC.yieldMany testdata+            .| CL.map (B8.pack . (\n -> show n ++ "\n"))+            .| CAlg.asyncGzipToFile testingFileNameGZ+    C.runConduitRes $+        CAlg.asyncGzipFromFile testingFileNameGZ+        .| CAlg.asyncGzipToFile testingFileNameGZ2+    shouldProduceIO testdata $+        CAlg.asyncGzipFromFile testingFileNameGZ2+            .| CB.lines+            .| CL.map (read . B8.unpack)+    removeFile testingFileNameGZ+    removeFile testingFileNameGZ2++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."])++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."])++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])