arbtt 0.10 → 0.10.0.1
raw patch · 6 files changed
+999/−11 lines, 6 filesdep +resourcetdep +textdep +unliftio-coredep −conduit-extradep ~aesondep ~conduitdep ~tasty
Dependencies added: resourcet, text, unliftio-core
Dependencies removed: conduit-extra
Dependency ranges changed: aeson, conduit, tasty, terminal-progress-bar
Files
- arbtt.cabal +25/−10
- src/Data/Conduit/Attoparsec.hs +246/−0
- src/Data/Conduit/Binary.hs +688/−0
- src/Data/Streaming/FileRead.hs +37/−0
- src/LockFile.hs +1/−0
- src/stats-main.hs +2/−1
arbtt.cabal view
@@ -1,6 +1,6 @@ name: arbtt-version: 0.10-license: GPL+version: 0.10.0.1+license: GPL-2 license-file: LICENSE category: Desktop cabal-version: >= 1.10@@ -42,7 +42,7 @@ build-depends: base >= 4.7 && < 4.11, filepath, directory, transformers, utf8-string, strict, containers,- aeson >= 0.6 && < 1.3,+ aeson >= 0.10 && < 1.3, binary >= 0.5, bytestring, deepseq @@ -61,6 +61,7 @@ UpgradeLog1 LeftFold LockFile+ DumpFormat ghc-options: -rtsopts @@ -104,9 +105,9 @@ binary >= 0.5, deepseq, bytestring, utf8-string, strict, transformers, directory, filepath,- aeson >= 0.6 && < 1.3,+ aeson >= 0.10 && < 1.3, array == 0.4.* || == 0.5.*,- terminal-progress-bar,+ terminal-progress-bar >= 0.2 && < 0.3, bytestring-progress, mtl @@ -147,7 +148,7 @@ base >= 4.7 && < 4.11, parsec == 3.*, containers == 0.5.*,- aeson >= 0.6 && < 1.3,+ aeson >= 0.10 && < 1.3, array == 0.4.* || == 0.5.*, binary >= 0.5, deepseq, bytestring, utf8-string, strict,@@ -185,9 +186,8 @@ parsec == 3.*, containers == 0.5.*, binary >= 0.5,- aeson >= 0.6 && < 1.3,- conduit == 1.2.*,- conduit-extra == 1.2.*,+ aeson >= 0.10 && < 1.3,+ conduit >= 1.2 && < 1.4, exceptions == 0.8.*, attoparsec == 0.13.*, deepseq, bytestring, utf8-string, strict,@@ -208,14 +208,29 @@ CommonStartup TimeLog LockFile+ DumpFormat ghc-options: -rtsopts if os(windows) cpp-options: -DWIN32+ other-modules:+ System.Win32.Mutex else other-modules: System.Locale.SetLocale default-language: Haskell98 + -- We inline Data.Conduit.Attoparsec, copied+ -- from conduit-extra to avoid dependencies on network etc.+ -- The build-depends below are exclusively used by these modules+ other-modules:+ Data.Conduit.Attoparsec+ Data.Conduit.Binary+ Data.Streaming.FileRead+ build-depends:+ text,+ resourcet >= 1.2,+ unliftio-core+ executable arbtt-recover main-is: recover-main.hs hs-source-dirs: src@@ -266,7 +281,7 @@ TimeLog Build-depends: base >= 4.7 && < 4.11- , tasty >= 0.7 && < 0.13+ , tasty >= 0.7 && < 1.1 , tasty-golden >= 2.2.0.2 && < 2.4 , tasty-hunit >= 0.2 && < 0.11 , process-extras >= 0.2 && < 0.8
+ src/Data/Conduit/Attoparsec.hs view
@@ -0,0 +1,246 @@+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE RankNTypes #-}++-- |+-- Copyright: 2011 Michael Snoyman, 2010 John Millikin+-- License: MIT+--+-- Consume attoparsec parsers via conduit.+--+-- This code was taken from attoparsec-enumerator and adapted for conduits.+module Data.Conduit.Attoparsec+ ( -- * Sink+ sinkParser+ , sinkParserEither+ -- * Conduit+ , conduitParser+ , conduitParserEither++ -- * Types+ , ParseError (..)+ , Position (..)+ , PositionRange (..)+ -- * Classes+ , AttoparsecInput+ ) where++import Control.Exception (Exception)+import Control.Monad (unless)+import qualified Data.ByteString as B+import qualified Data.Text as T+import qualified Data.Text.Internal as TI+import Data.Typeable (Typeable)+import Prelude hiding (lines)++import qualified Data.Attoparsec.ByteString+import qualified Data.Attoparsec.Text+import qualified Data.Attoparsec.Types as A+import Data.Conduit+import Control.Monad.Trans.Resource (MonadThrow, throwM)++-- | The context and message from a 'A.Fail' value.+data ParseError = ParseError+ { errorContexts :: [String]+ , errorMessage :: String+ , errorPosition :: Position+ } | DivergentParser+ deriving (Show, Typeable)++instance Exception ParseError++data Position = Position+ { posLine :: {-# UNPACK #-} !Int+ , posCol :: {-# UNPACK #-} !Int+ , posOffset :: {-# UNPACK #-} !Int+ -- ^ @since 1.2.0+ }+ deriving (Eq, Ord)++instance Show Position where+ show (Position l c off) = show l ++ ':' : show c ++ " (" ++ show off ++ ")"++data PositionRange = PositionRange+ { posRangeStart :: {-# UNPACK #-} !Position+ , posRangeEnd :: {-# UNPACK #-} !Position+ }+ deriving (Eq, Ord)++instance Show PositionRange where+ show (PositionRange s e) = show s ++ '-' : show e++-- | A class of types which may be consumed by an Attoparsec parser.+class AttoparsecInput a where+ parseA :: A.Parser a b -> a -> A.IResult a b+ feedA :: A.IResult a b -> a -> A.IResult a b+ empty :: a+ isNull :: a -> Bool+ notEmpty :: [a] -> [a]+ getLinesCols :: a -> Position++ -- | Return the beginning of the first input with the length of+ -- the second input removed. Assumes the second string is shorter+ -- than the first.+ stripFromEnd :: a -> a -> a++instance AttoparsecInput B.ByteString where+ parseA = Data.Attoparsec.ByteString.parse+ feedA = Data.Attoparsec.ByteString.feed+ empty = B.empty+ isNull = B.null+ notEmpty = filter (not . B.null)+ getLinesCols = B.foldl' f (Position 0 0 0)+ where+ f (Position l c o) ch+ | ch == 10 = Position (l + 1) 0 (o + 1)+ | otherwise = Position l (c + 1) (o + 1)+ stripFromEnd b1 b2 = B.take (B.length b1 - B.length b2) b1++instance AttoparsecInput T.Text where+ parseA = Data.Attoparsec.Text.parse+ feedA = Data.Attoparsec.Text.feed+ empty = T.empty+ isNull = T.null+ notEmpty = filter (not . T.null)+ getLinesCols = T.foldl' f (Position 0 0 0)+ where+ f (Position l c o) ch+ | ch == '\n' = Position (l + 1) 0 (o + 1)+ | otherwise = Position l (c + 1) (o + 1)+ stripFromEnd (TI.Text arr1 off1 len1) (TI.Text _ _ len2) =+ TI.text arr1 off1 (len1 - len2)++-- | Convert an Attoparsec 'A.Parser' into a 'Sink'. The parser will+-- be streamed bytes until it returns 'A.Done' or 'A.Fail'.+--+-- If parsing fails, a 'ParseError' will be thrown with 'throwM'.+--+-- Since 0.5.0+sinkParser :: (AttoparsecInput a, MonadThrow m) => A.Parser a b -> Consumer a m b+sinkParser = fmap snd . sinkParserPosErr (Position 1 1 0)++-- | Same as 'sinkParser', but we return an 'Either' type instead+-- of raising an exception.+--+-- Since 1.1.5+sinkParserEither :: (AttoparsecInput a, Monad m) => A.Parser a b -> Consumer a m (Either ParseError b)+sinkParserEither = (fmap.fmap) snd . sinkParserPos (Position 1 1 0)+++-- | Consume a stream of parsed tokens, returning both the token and+-- the position it appears at. This function will raise a 'ParseError'+-- on bad input.+--+-- Since 0.5.0+conduitParser :: (AttoparsecInput a, MonadThrow m) => A.Parser a b -> Conduit a m (PositionRange, b)+conduitParser parser =+ conduit $ Position 1 1 0+ where+ conduit !pos = await >>= maybe (return ()) go+ where+ go x = do+ leftover x+ (!pos', !res) <- sinkParserPosErr pos parser+ yield (PositionRange pos pos', res)+ conduit pos'+{-# SPECIALIZE conduitParser+ :: MonadThrow m+ => A.Parser T.Text b+ -> Conduit T.Text m (PositionRange, b) #-}+{-# SPECIALIZE conduitParser+ :: MonadThrow m+ => A.Parser B.ByteString b+ -> Conduit B.ByteString m (PositionRange, b) #-}++++-- | Same as 'conduitParser', but we return an 'Either' type instead+-- of raising an exception.+conduitParserEither+ :: (Monad m, AttoparsecInput a)+ => A.Parser a b+ -> Conduit a m (Either ParseError (PositionRange, b))+conduitParserEither parser =+ conduit $ Position 1 1 0+ where+ conduit !pos = await >>= maybe (return ()) go+ where+ go x = do+ leftover x+ eres <- sinkParserPos pos parser+ case eres of+ Left e -> yield $ Left e+ Right (!pos', !res) -> do+ yield $! Right (PositionRange pos pos', res)+ conduit pos'+{-# SPECIALIZE conduitParserEither+ :: Monad m+ => A.Parser T.Text b+ -> Conduit T.Text m (Either ParseError (PositionRange, b)) #-}+{-# SPECIALIZE conduitParserEither+ :: Monad m+ => A.Parser B.ByteString b+ -> Conduit B.ByteString m (Either ParseError (PositionRange, b)) #-}+++++sinkParserPosErr+ :: (AttoparsecInput a, MonadThrow m)+ => Position+ -> A.Parser a b+ -> Consumer a m (Position, b)+sinkParserPosErr pos0 p = sinkParserPos pos0 p >>= f+ where+ f (Left e) = throwM e+ f (Right a) = return a+{-# INLINE sinkParserPosErr #-}+++sinkParserPos+ :: (AttoparsecInput a, Monad m)+ => Position+ -> A.Parser a b+ -> Consumer a m (Either ParseError (Position, b))+sinkParserPos pos0 p = sink empty pos0 (parseA p)+ where+ sink prev pos parser = await >>= maybe close push+ where+ push c+ | isNull c = sink prev pos parser+ | otherwise = go False c $ parser c++ close = go True prev (feedA (parser empty) empty)++ go end c (A.Done lo x) = do+ let pos'+ | end = pos+ | otherwise = addLinesCols prev pos+ y = stripFromEnd c lo+ pos'' = addLinesCols y pos'+ unless (isNull lo) $ leftover lo+ pos'' `seq` return $! Right (pos'', x)+ go end c (A.Fail rest contexts msg) =+ let x = stripFromEnd c rest+ pos'+ | end = pos+ | otherwise = addLinesCols prev pos+ pos'' = addLinesCols x pos'+ in pos'' `seq` return $! Left (ParseError contexts msg pos'')+ go end c (A.Partial parser')+ | end = return $! Left DivergentParser+ | otherwise =+ pos' `seq` sink c pos' parser'+ where+ pos' = addLinesCols prev pos++ addLinesCols :: AttoparsecInput a => a -> Position -> Position+ addLinesCols x (Position lines cols off) =+ lines' `seq` cols' `seq` off' `seq` Position lines' cols' off'+ where+ Position dlines dcols doff = getLinesCols x+ lines' = lines + dlines+ cols' = (if dlines > 0 then 1 else cols) + dcols+ off' = off + doff+{-# INLINE sinkParserPos #-}
+ src/Data/Conduit/Binary.hs view
@@ -0,0 +1,688 @@+{-# LANGUAGE CPP, RankNTypes #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE ScopedTypeVariables #-}+-- | Functions for interacting with bytes.+--+-- For many purposes, it's recommended to use the conduit-combinators library,+-- which provides a more complete set of functions.+module Data.Conduit.Binary+ ( -- * Files and @Handle@s++ -- | Note that most of these functions live in the @MonadResource@ monad+ -- to ensure resource finalization even in the presence of exceptions. In+ -- order to run such code, you will need to use @runResourceT@.++ -- ** Sources+ sourceFile+ , sourceHandle+ , sourceHandleUnsafe+ , sourceIOHandle+ , sourceFileRange+ , sourceHandleRange+ , sourceHandleRangeWithBuffer+ , withSourceFile+ -- ** Sinks+ , sinkFile+ , sinkFileCautious+ , sinkTempFile+ , sinkSystemTempFile+ , sinkHandle+ , sinkIOHandle+ , sinkHandleBuilder+ , sinkHandleFlush+ , withSinkFile+ , withSinkFileBuilder+ , withSinkFileCautious+ -- ** Conduits+ , conduitFile+ , conduitHandle+ -- * Utilities+ -- ** Sources+ , sourceLbs+ -- ** Sinks+ , head+ , dropWhile+ , take+ , drop+ , sinkCacheLength+ , sinkLbs+ , mapM_+ -- *** Storable+ , sinkStorable+ , sinkStorableEx+ -- ** Conduits+ , isolate+ , takeWhile+ , Data.Conduit.Binary.lines+ ) where++import qualified Data.ByteString.Builder as BB+import qualified Data.Streaming.FileRead as FR+import Prelude hiding (head, take, drop, takeWhile, dropWhile, mapM_)+import qualified Data.ByteString as S+import Data.ByteString.Unsafe (unsafeUseAsCString)+import qualified Data.ByteString.Lazy as L+import Data.Conduit+import Data.Conduit.List (sourceList, consume)+import qualified Data.Conduit.List as CL+import Control.Exception (assert, finally, bracket)+import Control.Monad (unless, when)+import Control.Monad.IO.Class (liftIO, MonadIO)+import Control.Monad.IO.Unlift+import Control.Monad.Trans.Resource (allocate, release)+import Control.Monad.Trans.Class (lift)+import qualified System.IO as IO+import Data.Word (Word8, Word64)+#if (__GLASGOW_HASKELL__ < 710)+import Control.Applicative ((<$>))+#endif+import System.Directory (getTemporaryDirectory, removeFile)+import Data.ByteString.Lazy.Internal (defaultChunkSize)+import Data.ByteString.Internal (ByteString (PS), inlinePerformIO)+import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)+import Foreign.ForeignPtr (touchForeignPtr)+import Foreign.Ptr (plusPtr, castPtr)+import Foreign.Storable (Storable, peek, sizeOf)+import GHC.ForeignPtr (mallocPlainForeignPtrBytes)+import Control.Monad.Trans.Resource (MonadResource)+import Control.Monad.Catch (MonadThrow (..))+import Control.Exception (Exception)+import Data.Typeable (Typeable)+import Foreign.Ptr (Ptr)+#ifndef ALLOW_UNALIGNED_ACCESS+import Foreign.Marshal (alloca, copyBytes)+#endif+import System.Directory (renameFile)+import System.FilePath (takeDirectory, takeFileName, (<.>))+import System.IO (hClose, openBinaryTempFile)+import Control.Exception (throwIO, catch)+import System.IO.Error (isDoesNotExistError)+import qualified Data.ByteString.Builder as BB++-- | Stream the contents of a file as binary data.+--+-- Since 0.3.0+sourceFile :: MonadResource m+ => FilePath+ -> Producer m S.ByteString+sourceFile fp =+ bracketP+ (FR.openFile fp)+ FR.closeFile+ loop+ where+ loop h = do+ bs <- liftIO $ FR.readChunk h+ unless (S.null bs) $ do+ yield bs+ loop h++-- | Stream the contents of a 'IO.Handle' as binary data. Note that this+-- function will /not/ automatically close the @Handle@ when processing+-- completes, since it did not acquire the @Handle@ in the first place.+--+-- Since 0.3.0+sourceHandle :: MonadIO m+ => IO.Handle+ -> Producer m S.ByteString+sourceHandle h =+ loop+ where+ loop = do+ bs <- liftIO (S.hGetSome h defaultChunkSize)+ if S.null bs+ then return ()+ else yield bs >> loop++-- | Same as @sourceHandle@, but instead of allocating a new buffer for each+-- incoming chunk of data, reuses the same buffer. Therefore, the @ByteString@s+-- yielded by this function are not referentially transparent between two+-- different @yield@s.+--+-- This function will be slightly more efficient than @sourceHandle@ by+-- avoiding allocations and reducing garbage collections, but should only be+-- used if you can guarantee that you do not reuse a @ByteString@ (or any slice+-- thereof) between two calls to @await@.+--+-- Since 1.0.12+sourceHandleUnsafe :: MonadIO m => IO.Handle -> Source m ByteString+sourceHandleUnsafe handle = do+ fptr <- liftIO $ mallocPlainForeignPtrBytes defaultChunkSize+ let ptr = unsafeForeignPtrToPtr fptr+ loop = do+ count <- liftIO $ IO.hGetBuf handle ptr defaultChunkSize+ when (count > 0) $ do+ yield (PS fptr 0 count)+ loop++ loop++ liftIO $ touchForeignPtr fptr++-- | An alternative to 'sourceHandle'.+-- Instead of taking a pre-opened 'IO.Handle', it takes an action that opens+-- a 'IO.Handle' (in read mode), so that it can open it only when needed+-- and close it as soon as possible.+--+-- Since 0.3.0+sourceIOHandle :: MonadResource m+ => IO IO.Handle+ -> Producer m S.ByteString+sourceIOHandle alloc = bracketP alloc IO.hClose sourceHandle++-- | Stream all incoming data to the given 'IO.Handle'. Note that this function+-- does /not/ flush and will /not/ close the @Handle@ when processing completes.+--+-- Since 0.3.0+sinkHandle :: MonadIO m+ => IO.Handle+ -> Consumer S.ByteString m ()+sinkHandle h = awaitForever (liftIO . S.hPut h)++-- | Stream incoming builders, executing them directly on the buffer of the+-- given 'IO.Handle'. Note that this function does /not/ automatically close the+-- @Handle@ when processing completes.+-- Pass 'Data.ByteString.Builder.Extra.flush' to flush the buffer.+--+-- @since 1.2.2+sinkHandleBuilder :: MonadIO m => IO.Handle -> ConduitM BB.Builder o m ()+sinkHandleBuilder h = awaitForever (liftIO . BB.hPutBuilder h)++-- | Stream incoming @Flush@es, executing them on @IO.Handle@+-- Note that this function does /not/ automatically close the @Handle@ when+-- processing completes+--+-- @since 1.2.2+sinkHandleFlush :: MonadIO m+ => IO.Handle+ -> ConduitM (Flush S.ByteString) o m ()+sinkHandleFlush h =+ awaitForever $ \mbs -> liftIO $+ case mbs of+ Chunk bs -> S.hPut h bs+ Flush -> IO.hFlush h++-- | An alternative to 'sinkHandle'.+-- Instead of taking a pre-opened 'IO.Handle', it takes an action that opens+-- a 'IO.Handle' (in write mode), so that it can open it only when needed+-- and close it as soon as possible.+--+-- Since 0.3.0+sinkIOHandle :: MonadResource m+ => IO IO.Handle+ -> Consumer S.ByteString m ()+sinkIOHandle alloc = bracketP alloc IO.hClose sinkHandle++-- | Stream the contents of a file as binary data, starting from a certain+-- offset and only consuming up to a certain number of bytes.+--+-- Since 0.3.0+sourceFileRange :: MonadResource m+ => FilePath+ -> Maybe Integer -- ^ Offset+ -> Maybe Integer -- ^ Maximum count+ -> Producer m S.ByteString+sourceFileRange fp offset count = bracketP+ (IO.openBinaryFile fp IO.ReadMode)+ IO.hClose+ (\h -> sourceHandleRange h offset count)++-- | 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 =+ sourceHandleRangeWithBuffer handle offset count defaultChunkSize++-- | Stream the contents of a handle as binary data, starting from a certain+-- offset and only consuming up to a certain number of bytes. This function+-- consumes chunks as specified by the buffer size.+--+-- Since 1.1.8+sourceHandleRangeWithBuffer :: MonadIO m+ => IO.Handle+ -> Maybe Integer -- ^ Offset+ -> Maybe Integer -- ^ Maximum count+ -> Int -- ^ Buffer size+ -> Producer m S.ByteString+sourceHandleRangeWithBuffer handle offset count buffer = 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 buffer+ if S.null bs+ then return ()+ else do+ yield bs+ pullUnlimited++ pullLimited c = do+ bs <- liftIO $ S.hGetSome handle (min c buffer)+ let c' = c - S.length bs+ assert (c' >= 0) $+ if S.null bs+ then return ()+ else do+ yield bs+ pullLimited c'++-- | Stream all incoming data to the given file.+--+-- Since 0.3.0+sinkFile :: MonadResource m+ => FilePath+ -> Consumer S.ByteString m ()+sinkFile fp = sinkIOHandle (IO.openBinaryFile fp IO.WriteMode)++-- | Cautious version of 'sinkFile'. The idea here is to stream the+-- values to a temporary file in the same directory of the destination+-- file, and only on successfully writing the entire file, moves it+-- atomically to the destination path.+--+-- In the event of an exception occurring, the temporary file will be+-- deleted and no move will be made. If the application shuts down+-- without running exception handling (such as machine failure or a+-- SIGKILL), the temporary file will remain and the destination file+-- will be untouched.+--+-- @since 1.1.14+sinkFileCautious+ :: MonadResource m+ => FilePath+ -> ConduitM S.ByteString o m ()+sinkFileCautious fp =+ bracketP (cautiousAcquire fp) cautiousCleanup inner+ where+ inner (tmpFP, h) = do+ sinkHandle h+ liftIO $ do+ hClose h+ renameFile tmpFP fp++-- | Stream data into a temporary file in the given directory with the+-- given filename pattern, and return the temporary filename. The+-- temporary file will be automatically deleted when exiting the+-- active 'ResourceT' block, if it still exists.+--+-- @since 1.1.15+sinkTempFile :: MonadResource m+ => FilePath -- ^ temp directory+ -> String -- ^ filename pattern+ -> ConduitM ByteString o m FilePath+sinkTempFile tmpdir pattern = do+ (_releaseKey, (fp, h)) <- allocate+ (IO.openBinaryTempFile tmpdir pattern)+ (\(fp, h) -> IO.hClose h `finally` (removeFile fp `Control.Exception.catch` \e ->+ if isDoesNotExistError e+ then return ()+ else throwIO e))+ sinkHandle h+ liftIO $ IO.hClose h+ return fp++-- | Same as 'sinkTempFile', but will use the default temp file+-- directory for the system as the first argument.+--+-- @since 1.1.15+sinkSystemTempFile+ :: MonadResource m+ => String -- ^ filename pattern+ -> ConduitM ByteString o m FilePath+sinkSystemTempFile pattern = do+ dir <- liftIO getTemporaryDirectory+ sinkTempFile dir pattern++-- | Stream the contents of the input to a file, and also send it along the+-- pipeline. Similar in concept to the Unix command @tee@.+--+-- Since 0.3.0+conduitFile :: MonadResource m+ => FilePath+ -> Conduit S.ByteString m S.ByteString+conduitFile fp = bracketP+ (IO.openBinaryFile fp IO.WriteMode)+ IO.hClose+ conduitHandle++-- | Stream the contents of the input to a @Handle@, and also send it along the+-- pipeline. Similar in concept to the Unix command @tee@. Like @sourceHandle@,+-- does not close the handle on completion. Related to: @conduitFile@.+--+-- Since 1.0.9+conduitHandle :: MonadIO m => IO.Handle -> Conduit S.ByteString m S.ByteString+conduitHandle h = awaitForever $ \bs -> liftIO (S.hPut h bs) >> yield bs++-- | Ensure that only up to the given number of bytes are consumed by the inner+-- sink. Note that this does /not/ ensure that all of those bytes are in fact+-- consumed.+--+-- Since 0.3.0+isolate :: Monad m+ => Int+ -> Conduit S.ByteString m S.ByteString+isolate =+ loop+ where+ loop 0 = return ()+ loop count = do+ mbs <- await+ case mbs of+ Nothing -> return ()+ Just bs -> do+ let (a, b) = S.splitAt count bs+ case count - S.length a of+ 0 -> do+ unless (S.null b) $ leftover b+ yield a+ count' -> assert (S.null b) $ yield a >> loop count'++-- | Return the next byte from the stream, if available.+--+-- Since 0.3.0+head :: Monad m => Consumer S.ByteString m (Maybe Word8)+head = do+ mbs <- await+ case mbs of+ Nothing -> return Nothing+ Just bs ->+ case S.uncons bs of+ Nothing -> head+ Just (w, bs') -> leftover bs' >> return (Just w)++-- | Return all bytes while the predicate returns @True@.+--+-- Since 0.3.0+takeWhile :: Monad m => (Word8 -> Bool) -> Conduit S.ByteString m S.ByteString+takeWhile p =+ loop+ where+ loop = await >>= maybe (return ()) go++ go bs+ | S.null x = next+ | otherwise = yield x >> next+ where+ next = if S.null y then loop else leftover y+ (x, y) = S.span p bs++-- | Ignore all bytes while the predicate returns @True@.+--+-- Since 0.3.0+dropWhile :: Monad m => (Word8 -> Bool) -> Consumer S.ByteString m ()+dropWhile p =+ loop+ where+ loop = do+ mbs <- await+ case S.dropWhile p <$> mbs of+ Nothing -> return ()+ Just bs+ | S.null bs -> loop+ | otherwise -> leftover bs++-- | Take the given number of bytes, if available.+--+-- Since 0.3.0+take :: Monad m => Int -> Consumer S.ByteString m L.ByteString+take 0 = return L.empty+take n0 = go n0 id+ where+ go n front =+ await >>= maybe (return $ L.fromChunks $ front []) go'+ where+ go' bs =+ case S.length bs `compare` n of+ LT -> go (n - S.length bs) (front . (bs:))+ EQ -> return $ L.fromChunks $ front [bs]+ GT ->+ let (x, y) = S.splitAt n bs+ in assert (not $ S.null y) $ leftover y >> return (L.fromChunks $ front [x])++-- | Drop up to the given number of bytes.+--+-- Since 0.5.0+drop :: Monad m => Int -> Consumer S.ByteString m ()+drop 0 = return ()+drop n0 = go n0+ where+ go n =+ await >>= maybe (return ()) go'+ where+ go' bs =+ case S.length bs `compare` n of+ LT -> go (n - S.length bs)+ EQ -> return ()+ GT ->+ let y = S.drop n bs+ in assert (not $ S.null y) $ leftover y >> return ()++-- | Split the input bytes into lines. In other words, split on the LF byte+-- (10), and strip it from the output.+--+-- Since 0.3.0+lines :: Monad m => Conduit S.ByteString m S.ByteString+lines =+ loop []+ where+ loop acc = await >>= maybe (finish acc) (go acc)++ finish acc =+ let final = S.concat $ reverse acc+ in unless (S.null final) (yield final)++ go acc more =+ case S.uncons second of+ Just (_, second') -> yield (S.concat $ reverse $ first:acc) >> go [] second'+ Nothing -> loop $ more:acc+ where+ (first, second) = S.break (== 10) more++-- | Stream the chunks from a lazy bytestring.+--+-- Since 0.5.0+sourceLbs :: Monad m => L.ByteString -> Producer m S.ByteString+sourceLbs = sourceList . L.toChunks++-- | Stream the input data into a temp file and count the number of bytes+-- present. When complete, return a new @Source@ reading from the temp file+-- together with the length of the input in bytes.+--+-- All resources will be cleaned up automatically.+--+-- Since 1.0.5+sinkCacheLength :: (MonadResource m1, MonadResource m2)+ => Sink S.ByteString m1 (Word64, Source m2 S.ByteString)+sinkCacheLength = do+ tmpdir <- liftIO getTemporaryDirectory+ (releaseKey, (fp, h)) <- allocate+ (IO.openBinaryTempFile tmpdir "conduit.cache")+ (\(fp, h) -> IO.hClose h `finally` removeFile fp)+ len <- sinkHandleLen h+ liftIO $ IO.hClose h+ return (len, sourceFile fp >> release releaseKey)+ where+ sinkHandleLen :: MonadResource m => IO.Handle -> Sink S.ByteString m Word64+ sinkHandleLen h =+ loop 0+ where+ loop x =+ await >>= maybe (return x) go+ where+ go bs = do+ liftIO $ S.hPut h bs+ loop $ x + fromIntegral (S.length bs)++-- | Consume a stream of input into a lazy bytestring. Note that no lazy I\/O+-- is performed, but rather all content is read into memory strictly.+--+-- Since 1.0.5+sinkLbs :: Monad m => Sink S.ByteString m L.ByteString+sinkLbs = fmap L.fromChunks consume++mapM_BS :: Monad m => (Word8 -> m ()) -> S.ByteString -> m ()+mapM_BS f (PS fptr offset len) = do+ let start = unsafeForeignPtrToPtr fptr `plusPtr` offset+ end = start `plusPtr` len+ loop ptr+ | ptr >= end = inlinePerformIO (touchForeignPtr fptr) `seq` return ()+ | otherwise = do+ f (inlinePerformIO (peek ptr))+ loop (ptr `plusPtr` 1)+ loop start+{-# INLINE mapM_BS #-}++-- | Perform a computation on each @Word8@ in a stream.+--+-- Since 1.0.10+mapM_ :: Monad m => (Word8 -> m ()) -> Consumer S.ByteString m ()+mapM_ f = awaitForever (lift . mapM_BS f)+{-# INLINE mapM_ #-}++-- | Consume some instance of @Storable@ from the incoming byte stream. In the+-- event of insufficient bytes in the stream, returns a @Nothing@ and returns+-- all unused input as leftovers.+--+-- @since 1.1.13+sinkStorable :: (Monad m, Storable a) => Consumer S.ByteString m (Maybe a)+sinkStorable = sinkStorableHelper Just (return Nothing)++-- | Same as 'sinkStorable', but throws a 'SinkStorableInsufficientBytes'+-- exception (via 'throwM') in the event of insufficient bytes. This can be+-- more efficient to use than 'sinkStorable' as it avoids the need to+-- construct/deconstruct a @Maybe@ wrapper in the success case.+--+-- @since 1.1.13+sinkStorableEx :: (MonadThrow m, Storable a) => Consumer S.ByteString m a+sinkStorableEx = sinkStorableHelper id (throwM SinkStorableInsufficientBytes)++sinkStorableHelper :: forall m a b. (Monad m, Storable a)+ => (a -> b)+ -> (Consumer S.ByteString m b)+ -> Consumer S.ByteString m b+sinkStorableHelper wrap failure = do+ start+ where+ size = sizeOf (undefined :: a)++ -- try the optimal case: next chunk has all the data we need+ start = do+ mbs <- await+ case mbs of+ Nothing -> failure+ Just bs+ | S.null bs -> start+ | otherwise ->+ case compare (S.length bs) size of+ LT -> do+ -- looks like we're stuck concating+ leftover bs+ lbs <- take size+ let bs = S.concat $ L.toChunks lbs+ case compare (S.length bs) size of+ LT -> do+ leftover bs+ failure+ EQ -> process bs+ GT -> assert False (process bs)+ EQ -> process bs+ GT -> do+ let (x, y) = S.splitAt size bs+ leftover y+ process x++ -- Given a bytestring of exactly the correct size, grab the value+ process bs = return $! wrap $! inlinePerformIO $!+ unsafeUseAsCString bs (safePeek undefined . castPtr)++ safePeek :: a -> Ptr a -> IO a+#ifdef ALLOW_UNALIGNED_ACCESS+ safePeek _ = peek+#else+ safePeek val ptr = alloca (\t -> copyBytes t ptr (sizeOf val) >> peek t)+#endif+{-# INLINE sinkStorableHelper #-}++data SinkStorableException = SinkStorableInsufficientBytes+ deriving (Show, Typeable)+instance Exception SinkStorableException++-- | Like 'IO.withBinaryFile', but provides a source to read bytes from.+--+-- @since 1.2.1+withSourceFile+ :: (MonadUnliftIO m, MonadIO n)+ => FilePath+ -> (ConduitM i ByteString n () -> m a)+ -> m a+withSourceFile fp inner =+ withRunInIO $ \run ->+ IO.withBinaryFile fp IO.ReadMode $+ run . inner . sourceHandle++-- | Like 'IO.withBinaryFile', but provides a sink to write bytes to.+--+-- @since 1.2.1+withSinkFile+ :: (MonadUnliftIO m, MonadIO n)+ => FilePath+ -> (ConduitM ByteString o n () -> m a)+ -> m a+withSinkFile fp inner =+ withRunInIO $ \run ->+ IO.withBinaryFile fp IO.ReadMode $+ run . inner . sinkHandle++-- | Same as 'withSinkFile', but lets you use a 'BB.Builder'.+--+-- @since 1.2.1+withSinkFileBuilder+ :: (MonadUnliftIO m, MonadIO n)+ => FilePath+ -> (ConduitM BB.Builder o n () -> m a)+ -> m a+withSinkFileBuilder fp inner =+ withRunInIO $ \run ->+ IO.withBinaryFile fp IO.ReadMode $ \h ->+ run $ inner $ CL.mapM_ (liftIO . BB.hPutBuilder h)++-- | Like 'sinkFileCautious', but uses the @with@ pattern instead of+-- @MonadResource@.+--+-- @since 1.2.2+withSinkFileCautious+ :: (MonadUnliftIO m, MonadIO n)+ => FilePath+ -> (ConduitM S.ByteString o n () -> m a)+ -> m a+withSinkFileCautious fp inner =+ withRunInIO $ \run -> bracket+ (cautiousAcquire fp)+ cautiousCleanup+ (\(tmpFP, h) -> do+ a <- run $ inner $ sinkHandle h+ hClose h+ renameFile tmpFP fp+ return a)++-- | Helper function for Cautious functions above, do not export!+cautiousAcquire :: FilePath -> IO (FilePath, IO.Handle)+cautiousAcquire fp = openBinaryTempFile (takeDirectory fp) (takeFileName fp <.> "tmp")++-- | Helper function for Cautious functions above, do not export!+cautiousCleanup :: (FilePath, IO.Handle) -> IO ()+cautiousCleanup (tmpFP, h) = do+ hClose h+ removeFile tmpFP `Control.Exception.catch` \e ->+ if isDoesNotExistError e+ then return ()+ else throwIO e
+ src/Data/Streaming/FileRead.hs view
@@ -0,0 +1,37 @@+{-# LANGUAGE CPP #-}+-- | The standard @openFile@ call on Windows causing problematic file locking+-- in some cases. This module provides a cross-platform file reading API+-- without the file locking problems on Windows.+--+-- This module /always/ opens files in binary mode.+--+-- @readChunk@ will return an empty @ByteString@ on EOF.+module Data.Streaming.FileRead+ ( ReadHandle+ , openFile+ , closeFile+ , readChunk+ ) where++#if WINDOWS++import System.Win32File++#else++import qualified System.IO as IO+import qualified Data.ByteString as S+import Data.ByteString.Lazy.Internal (defaultChunkSize)++newtype ReadHandle = ReadHandle IO.Handle++openFile :: FilePath -> IO ReadHandle+openFile fp = ReadHandle `fmap` IO.openBinaryFile fp IO.ReadMode++closeFile :: ReadHandle -> IO ()+closeFile (ReadHandle h) = IO.hClose h++readChunk :: ReadHandle -> IO S.ByteString+readChunk (ReadHandle h) = S.hGetSome h defaultChunkSize++#endif
src/LockFile.hs view
@@ -7,6 +7,7 @@ #ifdef WIN32 import System.Win32.Mutex+import Control.Monad (unless) #else import System.Posix.IO #endif
src/stats-main.hs view
@@ -189,7 +189,8 @@ (_height, width) <- getTermSize hPutChar stderr '\r' hPutStr stderr $- mkProgressBar (msg "Processing data") percentage (fromIntegral width) (fromIntegral b) (fromIntegral size)+ mkProgressBar (msg "Processing data") percentage (fromIntegral width) $+ Progress (fromIntegral b) (fromIntegral size) when (fromIntegral b >= fromIntegral size) $ do hPutChar stderr '\r' hPutStr stderr (replicate width ' ')