hw-kafka-avro 1.3.0 → 1.4.0

raw patch · 2 files changed

+22/−11 lines, 2 filesdep +tagged

Dependencies added: tagged

Files

• hw-kafka-avro.cabal +2/−1
• src/Kafka/Avro/Decode.hs +20/−10

@@ -1,5 +1,5 @@ name:                hw-kafka-avro-version:             1.3.0+version:             1.4.0 synopsis:            Avro support for Kafka infrastructure description:         Please see README.md homepage:            https://github.com/haskell-works/hw-kafka-avro#readme@@ -41,6 +41,7 @@                        semigroups,                        servant,                        servant-client,+                       tagged,                        text,                        transformers,                        unordered-containers

@@ -1,17 +1,22 @@+{-# LANGUAGE ScopedTypeVariables #-} module Kafka.Avro.Decode (   DecodeError(..) , decodeWithSchema, extractSchemaId ) where +import           Control.Arrow             (left) import           Control.Monad.IO.Class    (MonadIO)-import           Data.Avro                 as A (FromAvro, Result (..))+import           Data.Avro                 as A (FromAvro, HasAvroSchema (..), Result (..), fromAvro) import qualified Data.Avro                 as A (decodeWithSchema)+import qualified Data.Avro.Decode          as A (decodeAvro)+import qualified Data.Avro.Deconflict      as A (deconflict) import           Data.Avro.Schema          (Schema) import           Data.Bits                 (shiftL) import           Data.ByteString.Lazy      (ByteString) import qualified Data.ByteString.Lazy      as BL hiding (zipWith) import           Data.Int+import           Data.Tagged               (Tagged, untag) import           Kafka.Avro.SchemaRegistry  data DecodeError = DecodeRegistryError SchemaRegistryError@@ -31,12 +36,19 @@   case schemaData of     Left err -> return $ Left err     Right (sid, payload) -> do-      res <- leftMap DecodeRegistryError <$> loadSchema sr sid-      return $ res >>= decode payload+      res <- left DecodeRegistryError <$> loadSchema sr sid+      return $ res >>= flip decodeWithDeconflict payload   where     schemaData = maybe (Left BadPayloadNoSchemaId) Right (extractSchemaId bs)-    decode p s = resultToEither s (A.decodeWithSchema s p) +decodeWithDeconflict :: forall a. (FromAvro a) => Schema -> ByteString -> Either DecodeError a+decodeWithDeconflict writerSchema bs =+  let readerSchema = untag (schema :: Tagged a Schema)+  in left (DecodeError readerSchema) $ do+    raw <- A.decodeAvro writerSchema bs+    val <- A.deconflict writerSchema readerSchema raw+    resultToEither readerSchema (fromAvro val)+ extractSchemaId :: ByteString -> Maybe (SchemaId, ByteString) extractSchemaId bs = do   (_ , b0) <- BL.uncons bs@@ -48,11 +60,9 @@   let int  =  sum $ zipWith shiftL ints [0, 8, 16, 24]   return (SchemaId int, b4) -leftMap :: (e -> e') -> Either e r -> Either e' r-leftMap _ (Right r) = Right r-leftMap f (Left e)  = Left (f e)--resultToEither :: Schema -> A.Result a -> Either DecodeError a+resultToEither :: Schema -> A.Result a -> Either String a resultToEither sc res = case res of   Success a -> Right a-  Error msg -> Left $ DecodeError sc msg+  Error msg -> Left msg++