diff --git a/ChangeLog b/ChangeLog --- a/ChangeLog +++ b/ChangeLog @@ -1,3 +1,7 @@ +Version 0.0.12.0 2022-03-20 by luispedro + * Add dispatchC_ conduit + * Switch to fingertree package (for GHC 9) + Version 0.0.11.0 2019-09-20 by luispedro * Add dispatchC conduit * Add deprecation warning for groupC diff --git a/Data/Conduit/Algorithms.hs b/Data/Conduit/Algorithms.hs --- a/Data/Conduit/Algorithms.hs +++ b/Data/Conduit/Algorithms.hs @@ -1,6 +1,6 @@ {-| Module : Data.Conduit.Algorithms -Copyright : 2013-2018 Luis Pedro Coelho +Copyright : 2013-2021 Luis Pedro Coelho License : MIT Maintainer : luis@luispedro.org @@ -18,7 +18,7 @@ import qualified Data.Conduit as C import qualified Data.Conduit.Internal as CI import qualified Data.Set as S -import qualified Data.PQueue.Prio.Min as PQ +import qualified Data.PriorityQueue.FingerTree as PQ import Control.Monad.Trans.Class (lift) import Control.Monad (foldM) @@ -86,16 +86,16 @@ _ -> error "This situation should have been impossible (mergeC/go)" -- norm1insert inserts the pipe in into the queue after ensuring that the pipe is CI.HaveOutput norm1insert :: (Monad m, Ord o) - => PQ.MinPQueue o (CI.Pipe () i o () m ()) + => PQ.PQueue o (CI.Pipe () i o () m ()) -> CI.Pipe () i o () m () - -> CI.Pipe () i o () m (PQ.MinPQueue o (CI.Pipe () i o () m ())) + -> CI.Pipe () i o () m (PQ.PQueue o (CI.Pipe () i o () m ())) norm1insert q c@(CI.HaveOutput _ v) = return (PQ.insert v c q) norm1insert q CI.Done{} = return q norm1insert q (CI.PipeM p) = lift p >>= norm1insert q norm1insert q (CI.NeedInput _ next) = norm1insert q (next ()) norm1insert q (CI.Leftover next ()) = norm1insert q next in do - let st = map (($ CI.Done) . CI.unConduitT) cs + let st = map (\c -> CI.unConduitT c $ CI.Done) cs go =<< foldM norm1insert PQ.empty st -- | Take two sorted sources and merge them. 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 @@ -1,12 +1,12 @@ {-| Module : Data.Conduit.Algorithms.Async -Copyright : 2013-2019 Luis Pedro Coelho +Copyright : 2013-2021 Luis Pedro Coelho License : MIT Maintainer : luis@luispedro.org Higher level async processing interfaces. -} -{-# LANGUAGE ScopedTypeVariables, FlexibleContexts, CPP, TupleSections #-} +{-# LANGUAGE ScopedTypeVariables, FlexibleContexts, CPP, TupleSections, LambdaCase , BangPatterns #-} module Data.Conduit.Algorithms.Async ( conduitPossiblyCompressedFile @@ -127,12 +127,11 @@ -- | yield all yAll :: Seq.Seq (A.Async b) -> C.ConduitT a b m () - yAll q - | Seq.null q = return () - | otherwise = do - (r, q') <- liftIO $ retrieveResult q - C.yield r - yAll q' + yAll Seq.Empty = return () + yAll q = do + (r, q') <- liftIO $ retrieveResult q + C.yield r + yAll q' loop :: Seq.Seq (A.Async b) -> C.ConduitT a b m () loop q = C.await >>= \case @@ -174,16 +173,13 @@ -- larger chunks are not split up and smaller chunks can be yielded too. bsConcatTo :: MonadIO m => Int -- ^ chunk hint -> C.ConduitT B.ByteString [B.ByteString] m () -bsConcatTo chunkSize = awaitJust start +bsConcatTo chunkSize = C.awaitForever (\v -> continue [v] (B.length v)) 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) + continue chunks !s + | s >= chunkSize = C.yield chunks + | otherwise = C.await >>= maybe + (C.yield chunks) + (\v -> continue (v:chunks) (s + B.length v)) untilNothing :: forall m i. (Monad m) => C.ConduitT (Maybe i) i m () untilNothing = awaitJust $ \case @@ -249,6 +245,9 @@ -- | 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. +-- +-- Note: unlike the ungzip conduit from 'Data.Conduit.Zlib', this function will +-- read *all* the compressed files in the stream (not just the first). -- -- See also 'asyncGzipFromFile' asyncGzipFrom :: forall m. (MonadIO m, MonadResource m, MonadUnliftIO m) => Handle -> C.ConduitT () B.ByteString m () diff --git a/Data/Conduit/Algorithms/Utils.hs b/Data/Conduit/Algorithms/Utils.hs --- a/Data/Conduit/Algorithms/Utils.hs +++ b/Data/Conduit/Algorithms/Utils.hs @@ -1,6 +1,6 @@ {-| Module : Data.Conduit.Algorithms.Utils -Copyright : 2013-2019 Luis Pedro Coelho +Copyright : 2013-2021 Luis Pedro Coelho License : MIT Maintainer : luis@luispedro.org @@ -11,13 +11,14 @@ , enumerateC , groupC , dispatchC + , dispatchC_ ) where import qualified Data.Conduit as C import qualified Data.Conduit.List as CL import Data.Conduit ((.|)) import Data.Maybe (maybe) -import Control.Monad (unless) +import Control.Monad (unless, void) -- | Act on the next input (do nothing if no input). @awaitJust f@ is equivalent to -- @@ -75,7 +76,7 @@ -- ,(1, "two") -- ,(0, "three") -- ] --- CC.yieldMany .| dispatches [sink1, sink2] +-- CC.yieldMany input .| dispatches [sink1, sink2] -- @ -- -- Then 'sink1' will receive "one" and "three", while 'sink2' will receive "two" @@ -89,3 +90,8 @@ if j == i || (i == n - 1 && j >= n) || (i == 0 && j < 0) then Just val else Nothing + +-- | Version of 'dispatchC' that returns () +dispatchC_ :: Monad m => [C.ConduitT a C.Void m ()] -> C.ConduitT (Int, a) C.Void m () +dispatchC_ sinks = void $ dispatchC sinks + diff --git a/conduit-algorithms.cabal b/conduit-algorithms.cabal --- a/conduit-algorithms.cabal +++ b/conduit-algorithms.cabal @@ -1,13 +1,13 @@ cabal-version: 1.12 --- This file has been generated from package.yaml by hpack version 0.31.2. +-- This file has been generated from package.yaml by hpack version 0.34.4. -- -- see: https://github.com/sol/hpack -- --- hash: 93525d4124a62b6b071240ceaaf942dfa47aaf1316df09321dc68cb9cdde507a +-- hash: e2659f2fc195dbd840de065369897ff04fd6484a0d867884a7d8ef845f40ba32 name: conduit-algorithms -version: 0.0.11.0 +version: 0.0.12.0 synopsis: Conduit-based algorithms description: Algorithms on Conduits, including higher level asynchronous processing and some other utilities. category: Conduit @@ -33,7 +33,11 @@ Data.Conduit.Algorithms.Async Data.Conduit.Algorithms.Async.ByteString Data.Conduit.Algorithms.Storable - default-extensions: BangPatterns OverloadedStrings LambdaCase TupleSections + default-extensions: + BangPatterns + OverloadedStrings + LambdaCase + TupleSections ghc-options: -Wall build-depends: async @@ -47,10 +51,10 @@ , containers , deepseq , exceptions + , fingertree , lzma-conduit , monad-control , mtl - , pqueue , resourcet , stm , stm-conduit >=4.0 @@ -69,10 +73,15 @@ Paths_conduit_algorithms hs-source-dirs: ./tests - default-extensions: BangPatterns OverloadedStrings LambdaCase TupleSections + default-extensions: + BangPatterns + OverloadedStrings + LambdaCase + TupleSections ghc-options: -Wall build-depends: - HUnit + HUnit >=1.3 + , QuickCheck >=2.8 , async , base >4.8 && <5 , bytestring @@ -86,17 +95,18 @@ , deepseq , directory , exceptions + , fingertree , lzma-conduit , monad-control , mtl - , pqueue , resourcet , stm , stm-conduit >=4.0 , streaming-commons - , test-framework - , test-framework-hunit - , test-framework-th + , tasty + , tasty-hunit + , tasty-quickcheck + , tasty-th , transformers , unliftio-core , vector @@ -109,7 +119,11 @@ main-is: Bench.hs hs-source-dirs: ./bench - default-extensions: BangPatterns OverloadedStrings LambdaCase TupleSections + default-extensions: + BangPatterns + OverloadedStrings + LambdaCase + TupleSections ghc-options: -Wall -Wall -fwarn-tabs -fno-warn-missing-signatures -threaded -rtsopts "-with-rtsopts=-A64m -n4m -H" build-depends: async @@ -125,10 +139,10 @@ , criterion , deepseq , exceptions + , fingertree , lzma-conduit , monad-control , mtl - , pqueue , resourcet , stm , stm-conduit >=4.0 diff --git a/tests/Tests.hs b/tests/Tests.hs --- a/tests/Tests.hs +++ b/tests/Tests.hs @@ -1,12 +1,11 @@ -{- Copyright 2017-2019 Luis Pedro Coelho +{- Copyright 2017-2021 Luis Pedro Coelho - License: MIT -} {-# LANGUAGE TemplateHaskell, CPP, QuasiQuotes, FlexibleContexts, OverloadedStrings #-} module Main where -import Test.Framework.TH -import Test.HUnit -import Test.Framework.Providers.HUnit +import Test.Tasty.TH +import Test.Tasty.HUnit import qualified Data.ByteString as B import qualified Data.ByteString.Char8 as B8 @@ -20,7 +19,7 @@ import Data.List (sort) import System.Directory (removeFile) import Control.Exception (catch, ErrorCall) -import Control.Monad (forM_) +import Control.Monad (forM_, void) import Control.Monad.Trans.Resource.Internal (ResourceT) import qualified Control.Monad.Trans.Resource as R import Control.Monad.IO.Unlift (MonadUnliftIO) @@ -323,3 +322,4 @@ Nothing -> return n Just x -> acc (n + x) C.runConduitPure (CC.yieldMany v .| CAlg.dispatchC [acc i | i <- [0 :: Int ..2]]) @=? [6,2,-2 :: Int] + C.runConduitPure (CC.yieldMany v .| CAlg.dispatchC_ [void (acc i) | i <- [0 :: Int ..2]]) @=? ()