diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,3 +1,11 @@ +## 1.1.13.1 + +* Fix an incorrect comment + +## 1.1.13 + +* Add `sinkStorable` and `sinkStorableEx` + ## 1.1.12.1 * Fix build for GHC `<= 7.8` [#260](https://github.com/snoyberg/conduit/issues/260) diff --git a/Data/Conduit/Attoparsec.hs b/Data/Conduit/Attoparsec.hs --- a/Data/Conduit/Attoparsec.hs +++ b/Data/Conduit/Attoparsec.hs @@ -105,7 +105,7 @@ f (Position l c) ch | ch == '\n' = Position (l + 1) 0 | otherwise = Position l (c + 1) stripFromEnd (TI.Text arr1 off1 len1) (TI.Text _ _ len2) = - TI.textP arr1 off1 (len1 - 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'. diff --git a/Data/Conduit/Binary.hs b/Data/Conduit/Binary.hs --- a/Data/Conduit/Binary.hs +++ b/Data/Conduit/Binary.hs @@ -1,4 +1,6 @@ {-# LANGUAGE CPP, RankNTypes #-} +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE ScopedTypeVariables #-} -- | Functions for interacting with bytes. -- -- For many purposes, it's recommended to use the conduit-combinators library, @@ -36,6 +38,9 @@ , sinkCacheLength , sinkLbs , mapM_ + -- *** Storable + , sinkStorable + , sinkStorableEx -- ** Conduits , isolate , takeWhile @@ -45,6 +50,7 @@ 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) @@ -55,16 +61,21 @@ 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) -import Foreign.Storable (peek) +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) -- | Stream the contents of a file as binary data. -- @@ -373,7 +384,7 @@ Just (_, second') -> yield (S.concat $ reverse $ first:acc) >> go [] second' Nothing -> loop $ more:acc where - (first, second) = S.breakByte 10 more + (first, second) = S.break (== 10) more -- | Stream the chunks from a lazy bytestring. -- @@ -435,3 +446,64 @@ 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 (peek . castPtr) +{-# INLINE sinkStorableHelper #-} + +data SinkStorableException = SinkStorableInsufficientBytes + deriving (Show, Typeable) +instance Exception SinkStorableException diff --git a/Data/Conduit/Lazy.hs b/Data/Conduit/Lazy.hs --- a/Data/Conduit/Lazy.hs +++ b/Data/Conduit/Lazy.hs @@ -1,6 +1,8 @@ {-# LANGUAGE CPP #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE Trustworthy #-} + +{-# OPTIONS_GHC -fno-warn-deprecations #-} -- Suppress warnings around Control.Monad.Trans.Error -- | Use lazy I\/O for consuming the contents of a source. Warning: All normal -- warnings of lazy I\/O apply. In particular, if you are using this with a -- @ResourceT@ transformer, you must force the list to be evaluated before @@ -31,7 +33,9 @@ import qualified Control.Monad.Trans.State.Strict as Strict ( StateT ) import qualified Control.Monad.Trans.Writer.Strict as Strict ( WriterT ) +#if (__GLASGOW_HASKELL__ < 710) import Data.Monoid (Monoid) +#endif import Control.Monad.ST (ST) import qualified Control.Monad.ST.Lazy as Lazy import Data.Functor.Identity (Identity) diff --git a/Data/Conduit/Network.hs b/Data/Conduit/Network.hs --- a/Data/Conduit/Network.hs +++ b/Data/Conduit/Network.hs @@ -38,17 +38,14 @@ , SN.HostPreference ) where -import Prelude hiding (catch) +import Prelude import Data.Conduit -import qualified Network.Socket as NS import Network.Socket (Socket) -import Network.Socket.ByteString (sendAll, recv) +import Network.Socket.ByteString (sendAll) import Data.ByteString (ByteString) import qualified GHC.Conc as Conc (yield) import qualified Data.ByteString as S -import qualified Data.ByteString.Char8 as S8 import Control.Monad.IO.Class (MonadIO (liftIO)) -import Control.Exception (throwIO, SomeException, try, finally, bracket, IOException, catch) import Control.Monad (unless, void) import Control.Monad.Trans.Control (MonadBaseControl, control, liftBaseWith) import Control.Monad.Trans.Class (lift) @@ -81,7 +78,10 @@ where loop = await >>= maybe (return ()) (\bs -> lift (liftIO $ sendAll socket bs) >> loop) +serverSettings :: Int -> SN.HostPreference -> SN.ServerSettings serverSettings = SN.serverSettingsTCP + +clientSettings :: Int -> ByteString -> SN.ClientSettings clientSettings = SN.clientSettingsTCP appSource :: (SN.HasReadWrite ad, MonadIO m) => ad -> Producer m ByteString diff --git a/Data/Conduit/Network/UDP.hs b/Data/Conduit/Network/UDP.hs --- a/Data/Conduit/Network/UDP.hs +++ b/Data/Conduit/Network/UDP.hs @@ -13,8 +13,7 @@ ) where import Data.Conduit -import Network.Socket (AddrInfo, SockAddr, Socket) -import qualified Network.Socket as NS +import Network.Socket (Socket) import Network.Socket.ByteString (recvFrom, send, sendAll, sendTo, sendAllTo) import Data.ByteString (ByteString) import Control.Monad.IO.Class (MonadIO (liftIO)) diff --git a/Data/Conduit/Network/Unix.hs b/Data/Conduit/Network/Unix.hs --- a/Data/Conduit/Network/Unix.hs +++ b/Data/Conduit/Network/Unix.hs @@ -26,20 +26,11 @@ , SN.setAfterBind ) where -import Data.Conduit -import Network.Socket (Socket) -import qualified Network.Socket as NS import Data.Conduit.Network (appSource, appSink, sourceSocket, sinkSocket) import qualified Data.Streaming.Network as SN -import Control.Monad.IO.Class (MonadIO (liftIO)) -import Control.Exception (throwIO, SomeException, try, finally, bracket, - bracketOnError, catch) -import Control.Monad (forever, void) -import Control.Monad.Trans.Control (control) -import Control.Concurrent (forkIO) -import System.Directory (removeFile) -import System.IO.Error (isDoesNotExistError) -import Control.Monad.Trans.Resource (MonadBaseControl) +clientSettings :: FilePath -> SN.ClientSettingsUnix clientSettings = SN.clientSettingsUnix + +serverSettings :: FilePath -> SN.ServerSettingsUnix serverSettings = SN.serverSettingsUnix diff --git a/Data/Conduit/Process.hs b/Data/Conduit/Process.hs --- a/Data/Conduit/Process.hs +++ b/Data/Conduit/Process.hs @@ -1,6 +1,7 @@ {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE TypeFamilies #-} {-# LANGUAGE RankNTypes #-} +{-# LANGUAGE CPP #-} {-# OPTIONS_GHC -fno-warn-orphans #-} -- | A full tutorial for this module is available on FP School of Haskell: -- . @@ -31,7 +32,9 @@ import Data.ByteString (ByteString) import Control.Concurrent.Async (runConcurrently, Concurrently(..)) import Control.Monad.Catch (MonadMask, onException, throwM, finally) +#if (__GLASGOW_HASKELL__ < 710) import Control.Applicative ((<$>), (<*>)) +#endif instance (r ~ (), MonadIO m, i ~ ByteString) => InputSource (ConduitM i o m r) where isStdStream = (\(Just h) -> return $ sinkHandle h, Just CreatePipe) @@ -48,8 +51,9 @@ -- return a tuple of the @ExitCode@ from the process and the output collected -- from the @Consumer@. -- --- If an exception is raised by the consumer, --- the process is terminated. +-- Note that, if an exception is raised by the consumer, the process is /not/ +-- terminated. This behavior is different from 'sourceProcessWithStreams' due +-- to historical reasons. -- -- Since 1.1.2 sourceProcessWithConsumer :: MonadIO m diff --git a/Data/Conduit/Text.hs b/Data/Conduit/Text.hs --- a/Data/Conduit/Text.hs +++ b/Data/Conduit/Text.hs @@ -38,7 +38,6 @@ , detectUtf ) where -import qualified Prelude import Prelude hiding (head, drop, takeWhile, lines, zip, zip3, zipWith, zipWith3, take, dropWhile) import qualified Control.Exception as Exc @@ -61,7 +60,7 @@ -- -- Since 0.3.0 data Codec = Codec - { codecName :: T.Text + { _codecName :: T.Text , codecEncode :: T.Text -> (B.ByteString, Maybe (TextException, T.Text)) @@ -74,7 +73,7 @@ | NewCodec T.Text (T.Text -> B.ByteString) (B.ByteString -> DecodeResult) instance Show Codec where - showsPrec d c = + showsPrec d c = let (cnst, name) = case c of Codec t _ _ -> ("Codec ", t) NewCodec t _ _ -> ("NewCodec ", t) @@ -150,7 +149,7 @@ -> Int -> (B.ByteString -> DecodeResult) -> Conduit B.ByteString m T.Text -decodeNew onFailure name = +decodeNew onFailure _name = loop where loop consumed dec = diff --git a/conduit-extra.cabal b/conduit-extra.cabal --- a/conduit-extra.cabal +++ b/conduit-extra.cabal @@ -1,5 +1,5 @@ Name: conduit-extra -Version: 1.1.12.1 +Version: 1.1.13.1 Synopsis: Batteries included conduit: adapters for common libraries. Description: The conduit package itself maintains relative small dependencies. The purpose of this package is to collect commonly used utility functions wrapping other library dependencies, without depending on heavier-weight dependencies. The basic idea is that this package should only depend on haskell-platform packages and conduit. @@ -78,6 +78,7 @@ , exceptions , process , resourcet + , QuickCheck , stm , streaming-commons , text diff --git a/test/Data/Conduit/BinarySpec.hs b/test/Data/Conduit/BinarySpec.hs --- a/test/Data/Conduit/BinarySpec.hs +++ b/test/Data/Conduit/BinarySpec.hs @@ -1,3 +1,5 @@ +{-# LANGUAGE GADTs #-} +{-# LANGUAGE RankNTypes #-} {-# LANGUAGE OverloadedStrings #-} module Data.Conduit.BinarySpec (spec) where @@ -15,6 +17,13 @@ import qualified Data.ByteString as S import qualified Data.ByteString.Char8 as S8 import Data.Functor.Identity +import Test.QuickCheck.Arbitrary (Arbitrary, arbitrary) +import Test.QuickCheck.Gen (Gen, oneof) +import Data.Word (Word8) +import Foreign.Storable (Storable, sizeOf, pokeByteOff) +import Data.Typeable (Typeable) +import Data.ByteString.Internal (unsafeCreate) +import Control.Applicative ((<$>), (<*>)) spec :: Spec spec = describe "Data.Conduit.Binary" $ do @@ -188,6 +197,99 @@ let src = CL.sourceList ["foobarbazbin"] src C.$= CB.isolate 10 C.$$ CL.head x `shouldBe` Just "foobarbazb" + + describe "Storable" $ do + let test name func = describe name $ do + let test' size = + prop ("chunk size " ++ show size) $ \stores0 -> do + let src = + loop (someStorables stores0) + where + loop bs + | S.null bs = return () + | otherwise = do + let (x, y) = S.splitAt size bs + C.yield x + loop y + + sink :: [SomeStorable] + -> C.Sink S.ByteString IO () + sink [] = do + mw <- CB.head + case mw of + Nothing -> return () + Just _ -> error "trailing bytes" + sink (next:rest) = do + withSomeStorable next checkOne + sink rest + + checkOne :: (Storable a, Eq a, Show a) + => a + -> C.Sink S.ByteString IO () + checkOne expected = do + mactual <- + if func + then CB.sinkStorable + else fmap Just CB.sinkStorableEx + actual <- + case mactual of + Nothing -> error "got Nothing" + Just actual -> return actual + liftIO $ actual `shouldBe` expected + + src C.$$ sink stores0 :: IO () + mapM_ test' [1, 5, 10, 100] + + test "sink Maybe" True + test "sink exception" False + + it' "insufficient bytes are leftovers, one chunk" $ do + let src = C.yield $ S.singleton 1 + src C.$$ do + mactual <- CB.sinkStorable + liftIO $ mactual `shouldBe` (Nothing :: Maybe Int) + lbs <- CB.sinkLbs + liftIO $ lbs `shouldBe` L.singleton 1 + + it' "insufficient bytes are leftovers, multiple chunks" $ do + let src = do + C.yield $ S.singleton 1 + C.yield $ S.singleton 2 + src C.$$ do + mactual <- CB.sinkStorable + liftIO $ mactual `shouldBe` (Nothing :: Maybe Int) + lbs <- CB.sinkLbs + liftIO $ lbs `shouldBe` L.pack [1, 2] + +data SomeStorable where + SomeStorable :: (Storable a, Eq a, Show a, Typeable a) => a -> SomeStorable +instance Show SomeStorable where + show (SomeStorable x) = show x +instance Arbitrary SomeStorable where + arbitrary = oneof + [ SomeStorable <$> (arbitrary :: Gen Int) + , SomeStorable <$> (arbitrary :: Gen Word8) + , SomeStorable <$> (arbitrary :: Gen Double) + ] + +withSomeStorable :: SomeStorable + -> (forall a. (Storable a, Eq a, Show a) => a -> b) + -> b +withSomeStorable (SomeStorable x) f = f x + +someStorables :: [SomeStorable] -> S.ByteString +someStorables stores0 = + unsafeCreate size start + where + size = sum $ map (\x -> withSomeStorable x sizeOf) stores0 + + start ptr = + go stores0 0 + where + go [] _ = return () + go (x:rest) off = do + withSomeStorable x (pokeByteOff ptr off) + go rest (off + withSomeStorable x sizeOf) it' :: String -> IO () -> Spec it' = it