diff --git a/avro.cabal b/avro.cabal --- a/avro.cabal +++ b/avro.cabal @@ -2,10 +2,10 @@ -- -- see: https://github.com/sol/hpack -- --- hash: 0af56f5e0a9d8bce12730d5834259529fad0458a9efdf3d8127c1f3ac4e921cb +-- hash: 307ace222c16e9660fc8fd8f1d3b126a4254daea39bb616b93782a86b81d73b0 name: avro -version: 0.3.4.1 +version: 0.3.4.2 synopsis: Avro serialization support for Haskell description: Avro serialization and deserialization support for Haskell category: Data @@ -120,6 +120,7 @@ Avro.Codec.NestedSpec Avro.Codec.TextSpec Avro.Codec.ZigZagSpec + Avro.Decode.Lazy.ContainerSpec Avro.Decode.Lazy.ValuesSpec Avro.Deconflict.Reader Avro.Deconflict.Writer diff --git a/src/Data/Avro/Decode.hs b/src/Data/Avro/Decode.hs --- a/src/Data/Avro/Decode.hs +++ b/src/Data/Avro/Decode.hs @@ -69,19 +69,20 @@ (containedSchema,) <$> getBlocks (schemaToGet containedSchema) syncBytes decompress where getBlocks :: Get a -> BL.ByteString -> (BL.ByteString -> Get BL.ByteString) -> Get [[a]] - getBlocks getValue sync decompress = - do nrObj <- sFromIntegral =<< getLong - nrBytes <- getLong - bytes <- decompress =<< G.getLazyByteString nrBytes - r <- case runGetOrFail (replicateM nrObj getValue) bytes of - Right (_,_,x) -> return x - Left (_,_,s) -> fail s - marker <- G.getLazyByteString nrSyncBytes - when (marker /= sync) (fail "Invalid marker, does not match sync bytes.") - e <- G.isEmpty - if e - then return [r] - else (r :) <$> getBlocks getValue sync decompress + getBlocks getValue sync decompress = do + isEmpty <- G.isEmpty + if isEmpty + then return [] + else do + nrObj <- sFromIntegral =<< getLong + nrBytes <- getLong + bytes <- decompress =<< G.getLazyByteString nrBytes + r <- case runGetOrFail (replicateM nrObj getValue) bytes of + Right (_,_,x) -> return x + Left (_,_,s) -> fail s + marker <- G.getLazyByteString nrSyncBytes + when (marker /= sync) (fail "Invalid marker, does not match sync bytes.") + (r :) <$> getBlocks getValue sync decompress getCodec :: Monad m => Maybe BL.ByteString -> m (BL.ByteString -> m BL.ByteString) getCodec code | Just "null" <- code = diff --git a/src/Data/Avro/Decode/Lazy.hs b/src/Data/Avro/Decode/Lazy.hs --- a/src/Data/Avro/Decode/Lazy.hs +++ b/src/Data/Avro/Decode/Lazy.hs @@ -141,7 +141,8 @@ -> Either String (Schema, [[T.LazyValue Type]]) getContainerValuesWith schemaToGet bs = case runGetOrFail getAvro bs of - Left (bs', _, err) -> Left err + Left (bs', _, err) -> + Left err Right (bs', _, ContainerHeader {..}) -> Right (containedSchema, snd $ getBlocks (schemaToGet containedSchema) syncBytes bs' decompress) where @@ -164,16 +165,18 @@ -> (BL.ByteString -> Get BL.ByteString) -- ^ how to decompress -> (BL.ByteString, [[T.LazyValue Type]]) getBlocks getValue sync bs decompress = - case runGetOrFail (getRawBlock decompress) bs of - Left (bs', _, err) -> (bs', [[T.Error err]]) - Right (bs', _, (nrObj, bytes)) -> - let (_, vs) = consumeN (fromIntegral nrObj) getValue bytes - in case checkMarker sync bs' of - Left err -> (bs', [[T.Error err]]) - Right bs'' | BL.null bs'' -> (bs'', [vs]) - Right bs'' -> - let (rest, vs') = getBlocks getValue sync bs'' decompress - in (rest, vs : vs') + if BL.null bs + then (bs, []) + else case runGetOrFail (getRawBlock decompress) bs of + Left (bs', _, err) -> (bs', [[T.Error err]]) + Right (bs', _, (nrObj, bytes)) -> + let (_, vs) = consumeN (fromIntegral nrObj) getValue bytes + in case checkMarker sync bs' of + Left err -> (bs', [[T.Error err]]) + Right bs'' | BL.null bs'' -> (bs'', [vs]) + Right bs'' -> + let (rest, vs') = getBlocks getValue sync bs'' decompress + in (rest, vs : vs') decodeGet :: GetAvro a => (a -> T.LazyValue Type) -> BL.ByteString -> (BL.ByteString, T.LazyValue Type) decodeGet f bs = diff --git a/test/Avro/Decode/Lazy/ContainerSpec.hs b/test/Avro/Decode/Lazy/ContainerSpec.hs new file mode 100644 --- /dev/null +++ b/test/Avro/Decode/Lazy/ContainerSpec.hs @@ -0,0 +1,65 @@ +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TemplateHaskell #-} +module Avro.Decode.Lazy.ContainerSpec +where + +import Data.Avro as A +import Data.Avro.Decode.Lazy as DL +import Data.Avro.Decode.Lazy.Convert as TC +import Data.Avro.Deriving +import Data.Either (isLeft) +import Data.List (unfoldr) +import Data.Semigroup ((<>)) +import Data.Text (pack) + +import Test.Hspec + +{-# ANN module ("HLint: ignore Redundant do" :: String) #-} + +deriveAvro "test/data/small.avsc" + +spec :: Spec +spec = describe "Avro.Decode.Lazy.ContainerSpec" $ do + + it "should decode empty container" $ + encodeThenDecode ([] :: [[Endpoint]]) >>= (`shouldBe` []) + + it "should decode container with one block" $ do + let msg = mkEndpoint 1 + res <- encodeThenDecode [[msg]] + sequence res `shouldBe` Right [msg] + + it "should decode container with empty blocks" $ do + let msg = mkEndpoint 1 + res <- encodeThenDecode [[msg], [], []] + sequence res `shouldBe` Right [msg] + + it "should decode container with empty blocks in between" $ do + let (msg1, msg2) = (mkEndpoint 1, mkEndpoint 2) + res <- encodeThenDecode [[msg1], [], [], [msg2]] + sequence res `shouldBe` Right [msg1, msg2] + + it "should decode container with multiple blocks" $ do + let msgs = mkEndpoint <$> [1..10] + let chunks = chunksOf 4 msgs + res <- encodeThenDecode chunks + sequence res `shouldBe` Right msgs + +encodeThenDecode :: (FromAvro a, ToAvro a) => [[a]] -> IO [Either String a] +encodeThenDecode as = + DL.decodeContainer <$> A.encodeContainer as + +mkEndpoint :: Int -> Endpoint +mkEndpoint i = + Endpoint + { endpointIps = ["192.168.1." <> pack (show i), "127.0.0." <> pack (show i)] + , endpointPorts = [PortRange 1 10, PortRange 11 20] + , endpointOpaque = Opaque "16-b-long-string" + , endpointCorrelation = Opaque "opaq-correlation" + , endpointTag = Left (fromIntegral i) + } + +chunksOf :: Int -> [a] -> [[a]] +chunksOf n = takeWhile (not.null) . unfoldr (Just . splitAt n)