diff --git a/Data/Conduit/Binary.hs b/Data/Conduit/Binary.hs --- a/Data/Conduit/Binary.hs +++ b/Data/Conduit/Binary.hs @@ -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. -- diff --git a/Data/Conduit/List.hs b/Data/Conduit/List.hs --- a/Data/Conduit/List.hs +++ b/Data/Conduit/List.hs @@ -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. -- diff --git a/Data/Conduit/Text.hs b/Data/Conduit/Text.hs --- a/Data/Conduit/Text.hs +++ b/Data/Conduit/Text.hs @@ -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 diff --git a/conduit.cabal b/conduit.cabal --- a/conduit.cabal +++ b/conduit.cabal @@ -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 . diff --git a/test/main.hs b/test/main.hs --- a/test/main.hs +++ b/test/main.hs @@ -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