conduit 1.0.7.4 → 1.0.8
raw patch · 5 files changed
+256/−17 lines, 5 files
Files
- Data/Conduit/Binary.hs +25/−14
- Data/Conduit/List.hs +13/−1
- Data/Conduit/Text.hs +109/−1
- conduit.cabal +1/−1
- test/main.hs +108/−0
Data/Conduit/Binary.hs view
@@ -12,6 +12,7 @@ , sourceHandle , sourceIOHandle , sourceFileRange+ , sourceHandleRange -- ** Sinks , sinkFile , sinkHandle@@ -47,6 +48,7 @@ import Data.Word (Word8, Word64) import Control.Applicative ((<$>)) import System.Directory (getTemporaryDirectory, removeFile)+import Data.ByteString.Lazy.Internal (defaultChunkSize) #if CABAL_OS_WINDOWS import qualified System.Win32File as F #elif NO_HANDLES@@ -83,7 +85,7 @@ loop where loop = do- bs <- liftIO (S.hGetSome h 4096)+ bs <- liftIO (S.hGetSome h defaultChunkSize) if S.null bs then return () else yield bs >> loop@@ -131,25 +133,34 @@ sourceFileRange fp offset count = bracketP (IO.openBinaryFile fp IO.ReadMode) IO.hClose- start- where- start handle = do- case offset of- Nothing -> return ()- Just off -> liftIO $ IO.hSeek handle IO.AbsoluteSeek off- case count of- Nothing -> pullUnlimited handle- Just c -> pullLimited (fromInteger c) handle+ (\h -> sourceHandleRange h offset count) - pullUnlimited handle = do+-- | Stream the contents of a handle as binary data, starting from a certain+-- offset and only consuming up to a certain number of bytes.+--+-- Since 1.0.8+sourceHandleRange :: MonadIO m+ => IO.Handle+ -> Maybe Integer -- ^ Offset+ -> Maybe Integer -- ^ Maximum count+ -> Producer m S.ByteString+sourceHandleRange handle offset count = do+ case offset of+ Nothing -> return ()+ Just off -> liftIO $ IO.hSeek handle IO.AbsoluteSeek off+ case count of+ Nothing -> pullUnlimited+ Just c -> pullLimited (fromInteger c)+ where+ pullUnlimited = do bs <- liftIO $ S.hGetSome handle 4096 if S.null bs then return () else do yield bs- pullUnlimited handle+ pullUnlimited - pullLimited c handle = do+ pullLimited c = do bs <- liftIO $ S.hGetSome handle (min c 4096) let c' = c - S.length bs assert (c' >= 0) $@@ -157,7 +168,7 @@ then return () else do yield bs- pullLimited c' handle+ pullLimited c' -- | Stream all incoming data to the given file. --
Data/Conduit/List.hs view
@@ -26,6 +26,7 @@ , consume , sinkNull -- ** Monadic+ , foldMapM , foldM , mapM_ -- * Conduits@@ -69,7 +70,7 @@ import Data.Monoid (Monoid, mempty, mappend) import qualified Data.Foldable as F import Data.Conduit-import Control.Monad (when, (<=<))+import Control.Monad (when, (<=<), liftM) import Control.Monad.Trans.Class (lift) -- | Generate a source from a seed value.@@ -159,6 +160,17 @@ fold combiner mempty where combiner accum = mappend accum . f++-- | A monoidal strict left fold in a Monad.+--+-- Since 1.0.8+foldMapM :: (Monad m, Monoid b)+ => (a -> m b)+ -> Consumer a m b+foldMapM f =+ foldM combiner mempty+ where+ combiner accum = liftM (mappend accum) . f -- | Apply the action to all values in the stream. --
Data/Conduit/Text.hs view
@@ -23,10 +23,16 @@ , lines , linesBounded , TextException (..)+ , takeWhile+ , dropWhile+ , take+ , drop+ , foldLines+ , withLine ) where import qualified Prelude-import Prelude hiding (head, drop, takeWhile, lines, zip, zip3, zipWith, zipWith3)+import Prelude hiding (head, drop, takeWhile, lines, zip, zip3, zipWith, zipWith3, take, dropWhile) import Control.Arrow (first) import qualified Control.Exception as Exc@@ -368,3 +374,105 @@ maybeDecode (a, b) = case tryEvaluate a of Left _ -> Nothing Right _ -> Just (a, b)++-- |+--+-- Since 1.0.8+takeWhile :: Monad m+ => (Char -> Bool)+ -> Conduit T.Text m T.Text+takeWhile p =+ loop+ where+ loop = await >>= maybe (return ()) go+ go t =+ case T.span p t of+ (x, y)+ | T.null y -> yield x >> loop+ | otherwise -> yield x >> leftover y++-- |+--+-- Since 1.0.8+dropWhile :: Monad m+ => (Char -> Bool)+ -> Consumer T.Text m ()+dropWhile p =+ loop+ where+ loop = await >>= maybe (return ()) go+ go t+ | T.null x = loop+ | otherwise = leftover x+ where+ x = T.dropWhile p t++-- |+--+-- Since 1.0.8+take :: Monad m => Int -> Conduit T.Text m T.Text+take =+ loop+ where+ loop i = await >>= maybe (return ()) (go i)+ go i t+ | diff == 0 = yield t+ | diff < 0 =+ let (x, y) = T.splitAt i t+ in yield x >> leftover y+ | otherwise = yield t >> loop diff+ where+ diff = i - T.length t++-- |+--+-- Since 1.0.8+drop :: Monad m => Int -> Consumer T.Text m ()+drop =+ loop+ where+ loop i = await >>= maybe (return ()) (go i)+ go i t+ | diff == 0 = return ()+ | diff < 0 = leftover $ T.drop i t+ | otherwise = loop diff+ where+ diff = i - T.length t++-- |+--+-- Since 1.0.8+foldLines :: Monad m+ => (a -> ConduitM T.Text o m a)+ -> a+ -> ConduitM T.Text o m a+foldLines f =+ start+ where+ start a = CL.peek >>= maybe (return a) (const $ loop $ f a)++ loop consumer = do+ a <- takeWhile (/= '\n') =$= do+ a <- CL.map (T.filter (/= '\r')) =$= consumer+ CL.sinkNull+ return a+ drop 1+ start a++-- |+--+-- Since 1.0.8+withLine :: Monad m+ => Sink T.Text m a+ -> Consumer T.Text m (Maybe a)+withLine consumer = toConsumer $ do+ mx <- CL.peek+ case mx of+ Nothing -> return Nothing+ Just _ -> do+ x <- takeWhile (/= '\n') =$ do+ x <- CL.map (T.filter (/= '\r')) =$ consumer+ CL.sinkNull+ return x+ drop 1+ return $ Just x
conduit.cabal view
@@ -1,5 +1,5 @@ Name: conduit-Version: 1.0.7.4+Version: 1.0.8 Synopsis: Streaming data processing library. Description: @conduit@ is a solution to the streaming data problem, allowing for production, transformation, and consumption of streams of data in constant memory. It is an alternative to lazy I\/O which guarantees deterministic resource handling, and fits in the same general solution space as @enumerator@\/@iteratee@ and @pipes@. For a tutorial, please visit <https://haskell.fpcomplete.com/user/snoyberg/library-documentation/conduit-overview>.
test/main.hs view
@@ -132,6 +132,15 @@ x <- CL.sourceList [[4],[2],[3],[1]] C.$$ CL.foldMap (++[(9 :: Int)]) x `shouldBe` [4,9,2,9,3,9,1,9] + describe "foldMapM" $ do+ it "sums 1..10" $ do+ Sum x <- CL.sourceList [1..(10 :: Int)] C.$$ CL.foldMapM (return . Sum)+ x `shouldBe` sum [1..10]++ it "preserves order" $ do+ x <- CL.sourceList [[4],[2],[3],[1]] C.$$ CL.foldMapM (return . (++[(9 :: Int)]))+ x `shouldBe` [4,9,2,9,3,9,1,9]+ describe "unfold" $ do it "works" $ do let f 0 = Nothing@@ -940,6 +949,105 @@ C.yield 3 lift $ return () (src C.$$ CL.consume) `shouldBe` Right [1, 2, 4 :: Int]++ describe "finalizers" $ do+ it "promptness" $ do+ imsgs <- I.newIORef []+ let add x = liftIO $ do+ msgs <- I.readIORef imsgs+ I.writeIORef imsgs $ msgs ++ [x]+ src' = C.bracketP+ (add "acquire")+ (const $ add "release")+ (const $ C.addCleanup (const $ add "inside") (mapM_ C.yield [1..5]))+ src = do+ src' C.$= CL.isolate 4+ add "computation"+ sink = CL.mapM (\x -> add (show x) >> return x) C.=$ CL.consume++ res <- C.runResourceT $ src C.$$ sink++ msgs <- I.readIORef imsgs+ -- FIXME this would be better msgs `shouldBe` words "acquire 1 2 3 4 inside release computation"+ msgs `shouldBe` words "acquire 1 2 3 4 release inside computation"++ res `shouldBe` [1..4 :: Int]++ it "left associative" $ do+ imsgs <- I.newIORef []+ let add x = liftIO $ do+ msgs <- I.readIORef imsgs+ I.writeIORef imsgs $ msgs ++ [x]+ p1 = C.bracketP (add "start1") (const $ add "stop1") (const $ add "inside1" >> C.yield ())+ p2 = C.bracketP (add "start2") (const $ add "stop2") (const $ add "inside2" >> C.await >>= maybe (return ()) C.yield)+ p3 = C.bracketP (add "start3") (const $ add "stop3") (const $ add "inside3" >> C.await)++ res <- C.runResourceT $ (p1 C.$= p2) C.$$ p3+ res `shouldBe` Just ()++ msgs <- I.readIORef imsgs+ msgs `shouldBe` words "start3 inside3 start2 inside2 start1 inside1 stop3 stop2 stop1"++ it "right associative" $ do+ imsgs <- I.newIORef []+ let add x = liftIO $ do+ msgs <- I.readIORef imsgs+ I.writeIORef imsgs $ msgs ++ [x]+ p1 = C.bracketP (add "start1") (const $ add "stop1") (const $ add "inside1" >> C.yield ())+ p2 = C.bracketP (add "start2") (const $ add "stop2") (const $ add "inside2" >> C.await >>= maybe (return ()) C.yield)+ p3 = C.bracketP (add "start3") (const $ add "stop3") (const $ add "inside3" >> C.await)++ res <- C.runResourceT $ p1 C.$$ (p2 C.=$ p3)+ res `shouldBe` Just ()++ msgs <- I.readIORef imsgs+ msgs `shouldBe` words "start3 inside3 start2 inside2 start1 inside1 stop3 stop2 stop1"++ describe "dan burton's associative tests" $ do+ let tellLn = tell . (++ "\n")+ finallyP fin = CI.addCleanup (const fin)+ printer = CI.awaitForever $ lift . tellLn . show+ idMsg msg = finallyP (tellLn msg) CI.idP+ takeP 0 = return ()+ takeP n = CI.awaitE >>= \ex -> case ex of+ Left _u -> return ()+ Right i -> CI.yield i >> takeP (pred n)++ testPipe p = execWriter $ runPipe $ printer <+< p <+< CI.sourceList ([1..] :: [Int])++ p1 = takeP (1 :: Int)+ p2 = idMsg "foo"+ p3 = idMsg "bar"++ (<+<) = (CI.<+<)+ runPipe = CI.runPipe++ test1L = testPipe $ (p1 <+< p2) <+< p3+ test1R = testPipe $ p1 <+< (p2 <+< p3)++ test2L = testPipe $ (p2 <+< p1) <+< p3+ test2R = testPipe $ p2 <+< (p1 <+< p3)++ test3L = testPipe $ (p2 <+< p3) <+< p1+ test3R = testPipe $ p2 <+< (p3 <+< p1)++ verify testL testR p1' p2' p3'+ | testL == testR = return () :: IO ()+ | otherwise = error $ unlines+ [ "FAILURE"+ , ""+ , "(" ++ p1' ++ " <+< " ++ p2' ++ ") <+< " ++ p3'+ , "------------------"+ , testL+ , ""+ , p1' ++ " <+< (" ++ p2' ++ " <+< " ++ p3' ++ ")"+ , "------------------"+ , testR+ ]++ it "test1" $ verify test1L test1R "p1" "p2" "p3"+ -- FIXME this is broken it "test2" $ verify test2L test2R "p2" "p1" "p3"+ it "test3" $ verify test3L test3R "p2" "p3" "p1" it' :: String -> IO () -> Spec it' = it