conduit 1.0.13.1 → 1.0.14
raw patch · 5 files changed
+166/−160 lines, 5 filesdep +text-stream-decodedep ~base
Dependencies added: text-stream-decode
Dependency ranges changed: base
Files
- Data/Conduit.hs +1/−1
- Data/Conduit/Lift.hs +47/−0
- Data/Conduit/Text.hs +34/−149
- conduit.cabal +3/−2
- test/main.hs +81/−8
Data/Conduit.hs view
@@ -316,7 +316,7 @@ pure = ZipSource . forever . yield (ZipSource f) <*> (ZipSource x) = ZipSource $ zipSourcesApp f x --- | Coalesce all values yielding by all of the @Source@s.+-- | Coalesce all values yielded by all of the @Source@s. -- -- Implemented on top of @ZipSource@, see that data type for more details. --
Data/Conduit/Lift.hs view
@@ -13,6 +13,10 @@ catchErrorC, -- liftCatchError, + -- * ExceptionT+ runExceptionC,+ catchExceptionC,+ -- * MaybeT maybeC, runMaybeC,@@ -65,6 +69,7 @@ import Control.Monad.Morph (hoist, lift, MFunctor(..), ) import Control.Monad.Trans.Class (MonadTrans(..))+import Control.Exception (SomeException) import Data.Monoid (Monoid(..)) @@ -167,6 +172,48 @@ go (HaveOutput p f o) = HaveOutput (go p) f o go (NeedInput x y) = NeedInput (go . x) (go . y) {-# INLINABLE catchErrorC #-}++-- | Run 'ExceptionT' in the base monad+--+-- Since 1.0.14+runExceptionC+ :: Monad m =>+ ConduitM i o (ExceptionT m) r -> ConduitM i o m (Either SomeException r)+runExceptionC =+ ConduitM . go . unConduitM+ where+ go (Done r) = Done (Right r)+ go (PipeM mp) = PipeM $ do+ eres <- runExceptionT mp+ return $ case eres of+ Left e -> Done $ Left e+ Right p -> go p+ go (Leftover p i) = Leftover (go p) i+ go (HaveOutput p f o) = HaveOutput (go p) (runExceptionT f >> return ()) o+ go (NeedInput x y) = NeedInput (go . x) (go . y)+{-# INLINABLE runExceptionC #-}++-- | Catch an exception in the base monad+--+-- Since 1.0.14+catchExceptionC+ :: Monad m =>+ ConduitM i o (ExceptionT m) r+ -> (SomeException -> ConduitM i o (ExceptionT m) r)+ -> ConduitM i o (ExceptionT m) r+catchExceptionC c0 h =+ ConduitM $ go $ unConduitM c0+ where+ go (Done r) = Done r+ go (PipeM mp) = PipeM $ do+ eres <- lift $ runExceptionT mp+ return $ case eres of+ Left e -> unConduitM $ h e+ Right p -> go p+ go (Leftover p i) = Leftover (go p) i+ go (HaveOutput p f o) = HaveOutput (go p) f o+ go (NeedInput x y) = NeedInput (go . x) (go . y)+{-# INLINABLE catchExceptionC #-} -- | Wrap the base monad in 'M.MaybeT' --
Data/Conduit/Text.hs view
@@ -34,23 +34,20 @@ import qualified Prelude import Prelude hiding (head, drop, takeWhile, lines, zip, zip3, zipWith, zipWith3, take, dropWhile) -import Control.Arrow (first) import qualified Control.Exception as Exc-import Data.Bits ((.&.), (.|.), shiftL) import qualified Data.ByteString as B import qualified Data.ByteString.Char8 as B8 import Data.Char (ord)-import Data.Maybe (catMaybes) import qualified Data.Text as T import qualified Data.Text.Encoding as TE-import Data.Word (Word8, Word16)-import System.IO.Unsafe (unsafePerformIO)+import Data.Word (Word8) import Data.Typeable (Typeable) import Data.Conduit import qualified Data.Conduit.List as CL import Control.Monad.Trans.Class (lift) import Control.Monad (unless,when)+import Data.Text.StreamDecoding -- | A specific character encoding. --@@ -66,6 +63,7 @@ (TextException, B.ByteString) B.ByteString) }+ | NewCodec T.Text (T.Text -> B.ByteString) (B.ByteString -> DecodeResult) instance Show Codec where showsPrec d c = showParen (d > 10) $@@ -141,6 +139,7 @@ -- -- Since 0.3.0 encode :: MonadThrow m => Codec -> Conduit T.Text m B.ByteString+encode (NewCodec _ enc _) = CL.map enc encode codec = CL.mapM $ \t -> do let (bs, mexc) = codecEncode codec t maybe (return bs) (monadThrow . fst) mexc@@ -151,6 +150,30 @@ -- -- Since 0.3.0 decode :: MonadThrow m => Codec -> Conduit B.ByteString m T.Text+decode (NewCodec name _ start) =+ loop 0 start+ where+ loop consumed dec = await >>= maybe (finish consumed dec) (go consumed dec)++ finish consumed dec =+ case dec B.empty of+ DecodeResultSuccess _ _ -> return ()+ DecodeResultFailure t rest -> onFailure consumed B.empty t rest++ go consumed dec bs | B.null bs = loop consumed dec+ go consumed dec bs =+ case dec bs of+ DecodeResultSuccess t dec' -> do+ unless (T.null t) (yield t)+ let consumed' = consumed + B.length bs+ consumed' `seq` loop consumed' dec'+ DecodeResultFailure t rest -> onFailure consumed bs t rest++ onFailure consumed bs t rest = do+ unless (T.null t) (yield t)+ unless (B.null rest) (leftover rest)+ let consumed' = consumed + B.length bs - B.length rest+ monadThrow $ NewDecodeException name consumed' (B.take 4 rest) decode codec = loop id where@@ -175,164 +198,34 @@ | EncodeException Codec Char | LengthExceeded Int | TextException Exc.SomeException+ | NewDecodeException !T.Text !Int !B.ByteString deriving (Show, Typeable) instance Exc.Exception TextException -byteSplits :: B.ByteString- -> [(B.ByteString, B.ByteString)]-byteSplits bytes = loop (B.length bytes) where- loop 0 = [(B.empty, bytes)]- loop n = B.splitAt n bytes : loop (n - 1)--splitSlowly :: (B.ByteString -> T.Text)- -> B.ByteString- -> (T.Text, Either- (TextException, B.ByteString)- B.ByteString)-splitSlowly dec bytes = valid where- valid = firstValid (Prelude.map decFirst splits)- splits = byteSplits bytes- firstValid = Prelude.head . catMaybes- tryDec = tryEvaluate . dec-- decFirst (a, b) = case tryDec a of- Left _ -> Nothing- Right text -> Just (text, case tryDec b of- Left exc -> Left (TextException exc, b)-- -- this case shouldn't occur, since splitSlowly- -- is only called when parsing failed somewhere- Right _ -> Right B.empty)- -- | -- Since 0.3.0 utf8 :: Codec-utf8 = Codec name enc dec where- name = T.pack "UTF-8"- enc text = (TE.encodeUtf8 text, Nothing)- dec bytes = case splitQuickly bytes >>= maybeDecode of- Just (text, extra) -> (text, Right extra)- Nothing -> splitSlowly TE.decodeUtf8 bytes-- -- Whether the given byte is a continuation byte.- isContinuation byte = byte .&. 0xC0 == 0x80-- -- The number of continuation bytes needed by the given- -- non-continuation byte. Returns -1 for an illegal UTF-8- -- non-continuation byte and the whole split quickly must fail so- -- as the input is passed to TE.decodeUtf8, which will issue a- -- suitable error.- required x0- | x0 .&. 0x80 == 0x00 = 0- | x0 .&. 0xE0 == 0xC0 = 1- | x0 .&. 0xF0 == 0xE0 = 2- | x0 .&. 0xF8 == 0xF0 = 3- | otherwise = -1-- splitQuickly bytes- | B.null l || req == -1 = Nothing- | req == B.length r = Just (TE.decodeUtf8 bytes, B.empty)- | otherwise = Just (TE.decodeUtf8 l', r')- where- (l, r) = B.spanEnd isContinuation bytes- req = required (B.last l)- l' = B.init l- r' = B.cons (B.last l) r+utf8 = NewCodec (T.pack "UTF-8") TE.encodeUtf8 streamUtf8 -- | -- Since 0.3.0 utf16_le :: Codec-utf16_le = Codec name enc dec where- name = T.pack "UTF-16-LE"- enc text = (TE.encodeUtf16LE text, Nothing)- dec bytes = case splitQuickly bytes of- Just (text, extra) -> (text, Right extra)- Nothing -> splitSlowly TE.decodeUtf16LE bytes-- splitQuickly bytes = maybeDecode (loop 0) where- maxN = B.length bytes-- loop n | n == maxN = decodeAll- | (n + 1) == maxN = decodeTo n- loop n = let- req = utf16Required- (B.index bytes n)- (B.index bytes (n + 1))- decodeMore = loop $! n + req- in if n + req > maxN- then decodeTo n- else decodeMore-- decodeTo n = first TE.decodeUtf16LE (B.splitAt n bytes)- decodeAll = (TE.decodeUtf16LE bytes, B.empty)+utf16_le = NewCodec (T.pack "UTF-16-LE") TE.encodeUtf16LE streamUtf16LE -- | -- Since 0.3.0 utf16_be :: Codec-utf16_be = Codec name enc dec where- name = T.pack "UTF-16-BE"- enc text = (TE.encodeUtf16BE text, Nothing)- dec bytes = case splitQuickly bytes of- Just (text, extra) -> (text, Right extra)- Nothing -> splitSlowly TE.decodeUtf16BE bytes-- splitQuickly bytes = maybeDecode (loop 0) where- maxN = B.length bytes-- loop n | n == maxN = decodeAll- | (n + 1) == maxN = decodeTo n- loop n = let- req = utf16Required- (B.index bytes (n + 1))- (B.index bytes n)- decodeMore = loop $! n + req- in if n + req > maxN- then decodeTo n- else decodeMore-- decodeTo n = first TE.decodeUtf16BE (B.splitAt n bytes)- decodeAll = (TE.decodeUtf16BE bytes, B.empty)--utf16Required :: Word8 -> Word8 -> Int-utf16Required x0 x1 = required where- required = if x >= 0xD800 && x <= 0xDBFF- then 4- else 2- x :: Word16- x = (fromIntegral x1 `shiftL` 8) .|. fromIntegral x0+utf16_be = NewCodec (T.pack "UTF-16-BE") TE.encodeUtf16BE streamUtf16BE -- | -- Since 0.3.0 utf32_le :: Codec-utf32_le = Codec name enc dec where- name = T.pack "UTF-32-LE"- enc text = (TE.encodeUtf32LE text, Nothing)- dec bs = case utf32SplitBytes TE.decodeUtf32LE bs of- Just (text, extra) -> (text, Right extra)- Nothing -> splitSlowly TE.decodeUtf32LE bs+utf32_le = NewCodec (T.pack "UTF-32-LE") TE.encodeUtf32LE streamUtf32LE -- | -- Since 0.3.0 utf32_be :: Codec-utf32_be = Codec name enc dec where- name = T.pack "UTF-32-BE"- enc text = (TE.encodeUtf32BE text, Nothing)- dec bs = case utf32SplitBytes TE.decodeUtf32BE bs of- Just (text, extra) -> (text, Right extra)- Nothing -> splitSlowly TE.decodeUtf32BE bs--utf32SplitBytes :: (B.ByteString -> T.Text)- -> B.ByteString- -> Maybe (T.Text, B.ByteString)-utf32SplitBytes dec bytes = split where- split = maybeDecode (dec toDecode, extra)- len = B.length bytes- lenExtra = mod len 4-- lenToDecode = len - lenExtra- (toDecode, extra) = if lenExtra == 0- then (bytes, B.empty)- else B.splitAt lenToDecode bytes+utf32_be = NewCodec (T.pack "UTF-32-BE") TE.encodeUtf32BE streamUtf32BE -- | -- Since 0.3.0@@ -366,14 +259,6 @@ else Just (EncodeException iso8859_1 (T.head unsafe), unsafe) dec bytes = (T.pack (B8.unpack bytes), Right B.empty)--tryEvaluate :: a -> Either Exc.SomeException a-tryEvaluate = unsafePerformIO . Exc.try . Exc.evaluate--maybeDecode :: (a, b) -> Maybe (a, b)-maybeDecode (a, b) = case tryEvaluate a of- Left _ -> Nothing- Right _ -> Just (a, b) -- | --
conduit.cabal view
@@ -1,5 +1,5 @@ Name: conduit-Version: 1.0.13.1+Version: 1.0.14 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>.@@ -58,10 +58,11 @@ , transformers >= 0.2.2 && < 0.4 , mtl , bytestring >= 0.9- , text >= 0.11 , void >= 0.5.5 , mmorph , directory+ , text >= 0.11+ , text-stream-decode ghc-options: -Wall test-suite test
test/main.hs view
@@ -40,6 +40,7 @@ import qualified Control.Concurrent.MVar as M import Control.Monad.Error (catchError, throwError, Error) import qualified Data.Map as Map+import Control.Arrow (first) (@=?) :: (Eq a, Show a) => a -> a -> IO () (@=?) = flip shouldBe@@ -402,25 +403,25 @@ (a, b) `shouldBe` (Just 1, [1..10]) describe "text" $ do- let go enc tenc tdec cenc = do- prop (enc ++ " single chunk") $ \chars -> runST $ runExceptionT_ $ do+ let go enc tenc tdec cenc = describe enc $ do+ prop "single chunk" $ \chars -> runST $ runExceptionT_ $ do let tl = TL.pack chars lbs = tenc tl src = CL.sourceList $ L.toChunks lbs ts <- src C.$= CT.decode cenc C.$$ CL.consume return $ TL.fromChunks ts == tl- prop (enc ++ " many chunks") $ \chars -> runIdentity $ runExceptionT_ $ do+ prop "many chunks" $ \chars -> runIdentity $ runExceptionT_ $ do let tl = TL.pack chars lbs = tenc tl src = mconcat $ map (CL.sourceList . return . S.singleton) $ L.unpack lbs- + ts <- src C.$= CT.decode cenc C.$$ CL.consume return $ TL.fromChunks ts == tl -- Check whether raw bytes are decoded correctly, in -- particular that Text decoding produces an error if -- and only if Conduit does.- prop (enc ++ " raw bytes") $ \bytes ->+ prop "raw bytes" $ \bytes -> let lbs = L.pack bytes src = CL.sourceList $ L.toChunks lbs etl = C.runException $ src C.$= CT.decode cenc C.$$ CL.consume@@ -428,18 +429,90 @@ in case etl of (Left _) -> (return $! TL.toStrict tl') `shouldThrow` anyException (Right tl) -> TL.fromChunks tl `shouldBe` tl'- prop (enc ++ " encoding") $ \chars -> runIdentity $ runExceptionT_ $ do+ prop "encoding" $ \chars -> runIdentity $ runExceptionT_ $ do let tss = map T.pack chars lbs = tenc $ TL.fromChunks tss src = mconcat $ map (CL.sourceList . return) tss bss <- src C.$= CT.encode cenc C.$$ CL.consume return $ L.fromChunks bss == lbs+ prop "valid then invalid" $ \x y chars -> runIdentity $ runExceptionT_ $ do+ let tss = map T.pack ([x, y]:chars)+ ts = T.concat tss+ lbs = tenc (TL.fromChunks tss) `L.append` "\0\0\0\0\0\0\0"+ src = mapM_ C.yield $ L.toChunks lbs+ Just x' <- src C.$$ CT.decode cenc C.=$ C.await+ return $ x' `T.isPrefixOf` ts go "utf8" TLE.encodeUtf8 TLE.decodeUtf8 CT.utf8 go "utf16_le" TLE.encodeUtf16LE TLE.decodeUtf16LE CT.utf16_le go "utf16_be" TLE.encodeUtf16BE TLE.decodeUtf16BE CT.utf16_be go "utf32_le" TLE.encodeUtf32LE TLE.decodeUtf32LE CT.utf32_le go "utf32_be" TLE.encodeUtf32BE TLE.decodeUtf32BE CT.utf32_be+ it "mixed utf16 and utf8" $ do+ let bs = "8\NUL:\NULu\NUL\215\216\217\218"+ src = C.yield bs C.$= CT.decode CT.utf16_le+ text <- src C.$$ C.await+ text `shouldBe` Just "8:u"+ (src C.$$ CL.sinkNull) `shouldThrow` anyException+ it "invalid utf8" $ do+ let bs = S.pack [0..255]+ src = C.yield bs C.$= CT.decode CT.utf8+ text <- src C.$$ C.await+ text `shouldBe` Just (T.pack $ map toEnum [0..127])+ (src C.$$ CL.sinkNull) `shouldThrow` anyException+ it "catch UTF8 exceptions" $ do+ let badBS = "this is good\128\128\0that was bad" + grabExceptions inner = C.catchC+ (inner C.=$= CL.map Right)+ (\e -> C.yield (Left (e :: CT.TextException)))++ res <- C.yield badBS C.$$ (,)+ <$> (grabExceptions (CT.decode CT.utf8) C.=$ CL.consume)+ <*> CL.consume++ first (map (either (Left . show) Right)) res `shouldBe`+ ( [ Right "this is good"+ , Left $ show $ CT.NewDecodeException "UTF-8" 12 "\128\128\0t"+ ]+ , ["\128\128\0that was bad"]+ )+ it "catch UTF8 exceptions, pure" $ do+ let badBS = "this is good\128\128\0that was bad"++ grabExceptions inner = do+ res <- C.runExceptionC $ inner C.=$= CL.map Right+ case res of+ Left e -> C.yield $ Left e+ Right () -> return ()++ let res = runIdentity $ C.yield badBS C.$$ (,)+ <$> (grabExceptions (CT.decode CT.utf8) C.=$ CL.consume)+ <*> CL.consume++ first (map (either (Left . show) Right)) res `shouldBe`+ ( [ Right "this is good"+ , Left $ show $ CT.NewDecodeException "UTF-8" 12 "\128\128\0t"+ ]+ , ["\128\128\0that was bad"]+ )+ it "catch UTF8 exceptions, catchExceptionC" $ do+ let badBS = "this is good\128\128\0that was bad"++ grabExceptions inner = C.catchExceptionC+ (inner C.=$= CL.map Right)+ (\e -> C.yield $ Left e)++ let res = C.runException_ $ C.yield badBS C.$$ (,)+ <$> (grabExceptions (CT.decode CT.utf8) C.=$ CL.consume)+ <*> CL.consume++ first (map (either (Left . show) Right)) res `shouldBe`+ ( [ Right "this is good"+ , Left $ show $ CT.NewDecodeException "UTF-8" 12 "\128\128\0t"+ ]+ , ["\128\128\0that was bad"]+ )+ describe "text lines" $ do it "works across split lines" $ (CL.sourceList [T.pack "abc", T.pack "d\nef"] C.$= CT.lines C.$$ CL.consume) ==@@ -1056,8 +1129,8 @@ test1L = testPipe $ (p1 <+< p2) <+< p3 test1R = testPipe $ p1 <+< (p2 <+< p3) - test2L = testPipe $ (p2 <+< p1) <+< p3- test2R = testPipe $ p2 <+< (p1 <+< p3)+ _test2L = testPipe $ (p2 <+< p1) <+< p3+ _test2R = testPipe $ p2 <+< (p1 <+< p3) test3L = testPipe $ (p2 <+< p3) <+< p1 test3R = testPipe $ p2 <+< (p3 <+< p1)