diff --git a/Data/Conduit/Binary.hs b/Data/Conduit/Binary.hs --- a/Data/Conduit/Binary.hs +++ b/Data/Conduit/Binary.hs @@ -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@. diff --git a/Data/Conduit/Internal.hs b/Data/Conduit/Internal.hs --- a/Data/Conduit/Internal.hs +++ b/Data/Conduit/Internal.hs @@ -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 diff --git a/System/PosixFile.hsc b/System/PosixFile.hsc --- a/System/PosixFile.hsc +++ b/System/PosixFile.hsc @@ -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 diff --git a/System/Win32File.hsc b/System/Win32File.hsc --- a/System/Win32File.hsc +++ b/System/Win32File.hsc @@ -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 #include #include @@ -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) diff --git a/conduit.cabal b/conduit.cabal --- a/conduit.cabal +++ b/conduit.cabal @@ -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 . @@ -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