conduit 1.0.2 → 1.0.3
raw patch · 5 files changed
+128/−73 lines, 5 filesPVP ok
version bump matches the API change (PVP)
API changes (from Hackage documentation)
Files
- Data/Conduit/Binary.hs +10/−0
- Data/Conduit/Internal.hs +64/−55
- System/PosixFile.hsc +32/−1
- System/Win32File.hsc +20/−15
- conduit.cabal +2/−2
Data/Conduit/Binary.hs view
@@ -161,7 +161,17 @@ sinkFile :: MonadResource m => FilePath -> Consumer S.ByteString m ()+#if NO_HANDLES+sinkFile fp =+ bracketP+ (F.openWrite fp)+ F.close+ loop+ where+ loop h = awaitForever $ liftIO . F.write h+#else sinkFile fp = sinkIOHandle (IO.openBinaryFile fp IO.WriteMode)+#endif -- | Stream the contents of the input to a file, and also send it along the -- pipeline. Similar in concept to the Unix command @tee@.
Data/Conduit/Internal.hs view
@@ -111,9 +111,9 @@ instance Monad m => Monad (Pipe l i o u m) where return = Done - Done x >>= fp = fp x HaveOutput p c o >>= fp = HaveOutput (p >>= fp) c o NeedInput p c >>= fp = NeedInput (p >=> fp) (c >=> fp)+ Done x >>= fp = fp x PipeM mp >>= fp = PipeM ((>>= fp) `liftM` mp) Leftover p i >>= fp = Leftover (p >>= fp) i @@ -312,26 +312,28 @@ -- Since 0.5.0 pipe :: Monad m => Pipe l a b r0 m r1 -> Pipe Void b c r1 m r2 -> Pipe l a c r0 m r2 pipe =- pipe' (return ())+ goRight (return ()) where- pipe' final left right =+ goRight final left right = case right of- Done r2 -> PipeM (final >> return (Done r2))- HaveOutput p c o -> HaveOutput (pipe' final left p) (c >> final) o- PipeM mp -> PipeM (liftM (pipe' final left) mp)- Leftover _ i -> absurd i- NeedInput rp rc -> upstream rp rc+ HaveOutput p c o -> HaveOutput (recurse p) (c >> final) o+ NeedInput rp rc -> goLeft rp rc final left+ Done r2 -> PipeM (final >> return (Done r2))+ PipeM mp -> PipeM (liftM recurse mp)+ Leftover _ i -> absurd i where- upstream rp rc =- case left of- Done r1 -> pipe (Done r1) (rc r1)- HaveOutput left' final' o -> pipe' final' left' (rp o)- PipeM mp -> PipeM (liftM (\left' -> pipe' final left' right) mp)- Leftover left' i -> Leftover (pipe' final left' right) i- NeedInput left' lc -> NeedInput- (\a -> pipe' final (left' a) right)- (\r0 -> pipe' final (lc r0) right)+ recurse = goRight final left + goLeft rp rc final left =+ case left of+ HaveOutput left' final' o -> goRight final' left' (rp o)+ NeedInput left' lc -> NeedInput (recurse . left') (recurse . lc)+ Done r1 -> goRight (return ()) (Done r1) (rc r1)+ PipeM mp -> PipeM (liftM recurse mp)+ Leftover left' i -> Leftover (recurse left') i+ where+ recurse = goLeft rp rc final+ -- | Same as 'pipe', but automatically applies 'injectLeftovers' to the right @Pipe@. -- -- Since 0.5.0@@ -342,25 +344,28 @@ -- -- However, this version tested as being significantly more efficient. pipeL =- pipe' (return ())+ goRight (return ()) where- pipe' :: Monad m => m () -> Pipe l a b r0 m r1 -> Pipe b b c r1 m r2 -> Pipe l a c r0 m r2- pipe' final left right =+ goRight final left right = case right of- Done r2 -> PipeM (final >> return (Done r2))- HaveOutput p c o -> HaveOutput (pipe' final left p) (c >> final) o- PipeM mp -> PipeM (liftM (pipe' final left) mp)- Leftover right' i -> pipe' final (HaveOutput left final i) right'- NeedInput rp rc ->- case left of- Done r1 -> pipe' (return ()) (Done r1) (rc r1)- HaveOutput left' final' o -> pipe' final' left' (rp o)- PipeM mp -> PipeM (liftM (\left' -> pipe' final left' right) mp)- NeedInput left' lc -> NeedInput- (\a -> pipe' final (left' a) right)- (\r0 -> pipe' final (lc r0) right)- Leftover left' i -> Leftover (pipe' final left' right) i+ HaveOutput p c o -> HaveOutput (recurse p) (c >> final) o+ NeedInput rp rc -> goLeft rp rc final left+ Done r2 -> PipeM (final >> return (Done r2))+ PipeM mp -> PipeM (liftM recurse mp)+ Leftover right' i -> goRight final (HaveOutput left final i) right'+ where+ recurse = goRight final left + goLeft rp rc final left =+ case left of+ HaveOutput left' final' o -> goRight final' left' (rp o)+ NeedInput left' lc -> NeedInput (recurse . left') (recurse . lc)+ Done r1 -> goRight (return ()) (Done r1) (rc r1)+ PipeM mp -> PipeM (liftM recurse mp)+ Leftover left' i -> Leftover (recurse left') i+ where+ recurse = goLeft rp rc final+ -- | Connect a @Source@ to a @Sink@ until the latter closes. Returns both the -- most recent state of the @Source@ and the result of the @Sink@. --@@ -372,23 +377,27 @@ => ResumableSource m o -> Sink o m r -> m (ResumableSource m o, r)-connectResume (ResumableSource (ConduitM left0) leftFinal0) =- go leftFinal0 left0 . unConduitM+connectResume (ResumableSource (ConduitM left0) leftFinal0) (ConduitM right0) =+ goRight leftFinal0 left0 right0 where- go leftFinal left right =+ goRight leftFinal left right = case right of- Done r2 -> return (ResumableSource (ConduitM left) leftFinal, r2)- PipeM mp -> mp >>= go leftFinal left HaveOutput _ _ o -> absurd o- Leftover p i -> go leftFinal (HaveOutput left leftFinal i) p- NeedInput rp rc ->- case left of- Leftover p () -> go leftFinal p right- HaveOutput left' leftFinal' o -> go leftFinal' left' (rp o)- NeedInput _ lc -> go leftFinal (lc ()) right- Done () -> go (return ()) (Done ()) (rc ())- PipeM mp -> mp >>= \left' -> go leftFinal left' right+ NeedInput rp rc -> goLeft rp rc leftFinal left+ Done r2 -> return (ResumableSource (ConduitM left) leftFinal, r2)+ PipeM mp -> mp >>= goRight leftFinal left+ Leftover p i -> goRight leftFinal (HaveOutput left leftFinal i) p + goLeft rp rc leftFinal left =+ case left of+ HaveOutput left' leftFinal' o -> goRight leftFinal' left' (rp o)+ NeedInput _ lc -> recurse (lc ())+ Done () -> goRight (return ()) (Done ()) (rc ())+ PipeM mp -> mp >>= recurse+ Leftover p () -> recurse p+ where+ recurse = goLeft rp rc leftFinal+ -- | Run a pipeline until processing completes. -- -- Since 0.5.0@@ -411,12 +420,12 @@ injectLeftovers = go [] where- go _ (Done r) = Done r go ls (HaveOutput p c o) = HaveOutput (go ls p) c o- go ls (PipeM mp) = PipeM (liftM (go ls) mp)- go ls (Leftover p l) = go (l:ls) p go (l:ls) (NeedInput p _) = go ls $ p l go [] (NeedInput p c) = NeedInput (go [] . p) (go [] . c)+ go _ (Done r) = Done r+ go ls (PipeM mp) = PipeM (liftM (go ls) mp)+ go ls (Leftover p l) = go (l:ls) p -- | Transform the monad that a @Pipe@ lives in. --@@ -509,30 +518,30 @@ sourceToPipe = go . unConduitM where+ go (HaveOutput p c o) = HaveOutput (go p) c o+ go (NeedInput _ c) = go $ c () go (Done ()) = Done () go (PipeM mp) = PipeM (liftM go mp)- go (NeedInput _ c) = go $ c ()- go (HaveOutput p c o) = HaveOutput (go p) c o go (Leftover p ()) = go p sinkToPipe :: Monad m => Sink i m r -> Pipe l i o u m r sinkToPipe = go . injectLeftovers . unConduitM where+ go (HaveOutput _ _ o) = absurd o+ go (NeedInput p c) = NeedInput (go . p) (const $ go $ c ()) go (Done r) = Done r go (PipeM mp) = PipeM (liftM go mp)- go (NeedInput p c) = NeedInput (go . p) (const $ go $ c ())- go (HaveOutput _ _ o) = absurd o go (Leftover _ l) = absurd l conduitToPipe :: Monad m => Conduit i m o -> Pipe l i o u m () conduitToPipe = go . injectLeftovers . unConduitM where+ go (HaveOutput p c o) = HaveOutput (go p) c o+ go (NeedInput p c) = NeedInput (go . p) (const $ go $ c ()) go (Done ()) = Done () go (PipeM mp) = PipeM (liftM go mp)- go (NeedInput p c) = NeedInput (go . p) (const $ go $ c ())- go (HaveOutput p c o) = HaveOutput (go p) c o go (Leftover _ l) = absurd l -- | Returns a tuple of the upstream and downstream results. Note that this
System/PosixFile.hsc view
@@ -2,11 +2,14 @@ {-# LANGUAGE GeneralizedNewtypeDeriving #-} module System.PosixFile ( openRead+ , openWrite , read+ , write , close ) where import Foreign.C.String (CString, withCString)+import Foreign.Ptr (castPtr) import Foreign.Marshal.Alloc (mallocBytes, free) #if __GLASGOW_HASKELL__ >= 704 import Foreign.C.Types (CInt (..))@@ -15,7 +18,7 @@ #endif import Foreign.C.Error (throwErrno) import Foreign.Ptr (Ptr)-import Data.Bits (Bits)+import Data.Bits (Bits, (.|.)) import Data.Word (Word8) import qualified Data.ByteString as S import qualified Data.ByteString.Unsafe as BU@@ -28,14 +31,22 @@ #{enum Flag, Flag , oRdonly = O_RDONLY+ , oWronly = O_WRONLY+ , oCreat = O_CREAT } foreign import ccall "open" c_open :: CString -> Flag -> IO CInt +foreign import ccall "open"+ c_open_mode :: CString -> Flag -> CInt -> IO CInt+ foreign import ccall "read" c_read :: FD -> Ptr Word8 -> CInt -> IO CInt +foreign import ccall "write"+ c_write :: FD -> Ptr Word8 -> CInt -> IO CInt+ foreign import ccall "close" close :: FD -> IO () @@ -48,6 +59,13 @@ then throwErrno $ "Could not open file: " ++ fp else return $ FD h +openWrite :: FilePath -> IO FD+openWrite fp = do+ h <- withCString fp $ \str -> c_open_mode str (oWronly .|. oCreat) 438 -- == octal 666+ if h < 0+ then throwErrno $ "Could not open file: " ++ fp+ else return $ FD h+ read :: FD -> IO (Maybe S.ByteString) read fd = do cstr <- mallocBytes 4096@@ -58,3 +76,16 @@ cstr (fromIntegral len) (free cstr)++write :: FD -> S.ByteString -> IO ()+write _ bs | S.null bs = return ()+write fd bs = do+ (written, len) <- BU.unsafeUseAsCStringLen bs $ \(cstr, len') -> do+ let len = fromIntegral len'+ written <- c_write fd (castPtr cstr) len+ return (written, len)+ case () of+ ()+ | written == len -> return ()+ | written <= 0 -> throwErrno $ "Error writing to file"+ | otherwise -> write fd $ BU.unsafeDrop (fromIntegral $ len - written) bs
System/Win32File.hsc view
@@ -7,22 +7,27 @@ ) where import Foreign.C.String (CString)+import Foreign.Ptr (castPtr) import Foreign.Marshal.Alloc (mallocBytes, free)+import Foreign.ForeignPtr (ForeignPtr, withForeignPtr) #if __GLASGOW_HASKELL__ >= 704 import Foreign.C.Types (CInt (..)) #else import Foreign.C.Types (CInt) #endif-import Foreign.C.Error (throwErrno)+import Foreign.C.Error (throwErrnoIfMinus1Retry) import Foreign.Ptr (Ptr) import Data.Bits (Bits, (.|.)) import qualified Data.ByteString as S import qualified Data.ByteString.Unsafe as BU+import qualified Data.ByteString.Internal as BI import Data.Text (pack) import Data.Text.Encoding (encodeUtf16LE) import Data.Word (Word8) import Prelude hiding (read)+import GHC.ForeignPtr (mallocPlainForeignPtrBytes) + #include <fcntl.h> #include <Share.h> #include <SYS/Stat.h>@@ -34,6 +39,8 @@ #{enum OFlag, OFlag , oBinary = _O_BINARY , oRdonly = _O_RDONLY+ , oWronly = _O_WRONLY+ , oCreat = _O_CREAT } newtype SHFlag = SHFlag CInt@@ -48,6 +55,7 @@ #{enum PMode, PMode , pIread = _S_IREAD+ , pIwrite = _S_IWRITE } foreign import ccall "_wsopen"@@ -56,6 +64,9 @@ foreign import ccall "_read" c_read :: FD -> Ptr Word8 -> CInt -> IO CInt +foreign import ccall "_write"+ c_write :: FD -> Ptr Word8 -> CInt -> IO CInt+ foreign import ccall "_close" close :: FD -> IO () @@ -68,25 +79,19 @@ -- null octets to account for UTF16 encoding let bs = encodeUtf16LE $ pack $ fp ++ "\0" h <- BU.unsafeUseAsCString bs $ \str ->+ throwErrnoIfMinus1Retry "System.Win32File.openRead" $ c_wsopen str (oBinary .|. oRdonly) shDenyno pIread- if h < 0- then throwErrno $ "Could not open file: " ++ fp- else return $ FD h+ return $ FD h read :: FD -> IO (Maybe S.ByteString) read fd = do- cstr <- mallocBytes 4096- len <- c_read fd cstr 4096- if len == 0- then do- free cstr- return Nothing- else do- fmap Just $ BU.unsafePackCStringFinalizer- cstr- (fromIntegral len)- (free cstr)+ fp <- mallocPlainForeignPtrBytes 4096+ withForeignPtr fp $ \p -> do+ len <- throwErrnoIfMinus1Retry "System.Win32File.read" $ c_read fd p 4096+ if len == 0+ then return $! Nothing+ else return $! Just $! BI.PS fp 0 (fromIntegral len)
conduit.cabal view
@@ -1,5 +1,5 @@ Name: conduit-Version: 1.0.2+Version: 1.0.3 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>.@@ -37,8 +37,8 @@ cpp-options: -DCABAL_OS_WINDOWS other-modules: System.Win32File else+ other-modules: System.PosixFile if flag(nohandles)- other-modules: System.PosixFile cpp-options: -DNO_HANDLES Exposed-modules: Data.Conduit Data.Conduit.Binary