hw-kafka-avro 4.0.1 → 5.0.0

raw patch · 7 files changed

+133/−110 lines, 7 filesdep ~avronew-uploaderPVP ok

version bump matches the API change (PVP)

Dependency ranges changed: avro

API changes (from Hackage documentation)

Files

• example/Main.hs +4/−7
• example/Message.hs +19/−35
• hw-kafka-avro.cabal +6/−6
• src/Kafka/Avro.hs +1/−1
• src/Kafka/Avro/Decode.hs +32/−34
• src/Kafka/Avro/Encode.hs +70/−26
• src/Kafka/Avro/SchemaRegistry.hs +1/−1

@@ -4,9 +4,6 @@  import           Control.Monad.Trans.Except import qualified Data.Aeson                 as J-import qualified Data.Avro                  as A-import           Data.Avro.Schema           as S-import qualified Data.Avro.Types            as AT import           Data.Monoid  import Data.Int@@ -27,9 +24,9 @@  roundtrip :: SchemaRegistry -> ExceptT AppError IO TestMessage roundtrip sr = do-  enc <- withExceptT EncError (encode exampleMessage)-  dec <- withExceptT DecError (decode enc)+  enc <- withExceptT EncError (encode' exampleMessage)+  dec <- withExceptT DecError (decode' enc)   return dec   where-    encode msg = ExceptT $ encodeWithSchema sr (Subject "example-subject") exampleMessage-    decode msg = ExceptT $ decodeWithSchema sr msg+    encode' msg = ExceptT $ encode sr (Subject "example-subject") exampleMessage+    decode' msg = ExceptT $ decode sr msg

@@ -1,42 +1,26 @@+{-# LANGUAGE DeriveGeneric     #-} {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE QuasiQuotes       #-}+{-# LANGUAGE TemplateHaskell   #-} module Message ( TestMessage(..)+, schema'TestMessage ) where--- -import           Data.Avro-import           Data.Avro.Schema-import qualified Data.Avro.Types  as AT-import           Data.Int-import           Data.Text--data TestMessage = TestMessage Int64 Text Bool Int64 deriving (Show, Eq, Ord)--testMessageSchema =-  let fld nm = Field nm [] Nothing Nothing-   in Record (TN "TestMessage" ["hw", "kafka", "avro", "test"]) [] Nothing Nothing-         [ fld "id" Long Nothing-         , fld "name" String Nothing-         , fld "is_active" Boolean Nothing-         , fld "timestamp" Long Nothing-         ]--instance HasAvroSchema TestMessage where-  schema = pure testMessageSchema+import Data.Avro+import Data.Avro.Deriving -instance FromAvro TestMessage where-  fromAvro (AT.Record _ r) =-    TestMessage <$> r .: "id"-                <*> r .: "name"-                <*> r .: "is_active"-                <*> r .: "timestamp"-  fromAvro v = badValue v "TestMessage"+deriveAvroFromByteString [r|+{+  "type": "record",+  "name": "TestMessage",+  "namespace": "hw.kafka.avro.test",+  "fields": [+    { "name": "id", "type": "long" },+    { "name": "name", "type": "string" },+    { "name": "is_active", "type": "boolean" },+    { "name": "timestamp", "type": "long" }+  ]+}+|] -instance ToAvro TestMessage where-  toAvro (TestMessage i s d t) =-    record testMessageSchema-      [ "id"        .= i-      , "name"      .= s-      , "is_active" .= d-      , "timestamp" .= t-      ]

@@ -1,7 +1,7 @@-cabal-version: 1.12+cabal-version:  2.4  name:           hw-kafka-avro-version:        4.0.1+version:        5.0.0 synopsis:       Avro support for Kafka infrastructure description:    Avro support for Kafka infrastructure. category:       Services@@ -10,7 +10,7 @@ author:         Alexey Raga maintainer:     alexey.raga@gmail.com copyright:      Alexey Raga-license:        BSD3+license:        BSD-3-Clause license-file:   LICENSE build-type:     Simple extra-source-files:@@ -34,7 +34,7 @@   hs-source-dirs: src   build-depends:       aeson-    , avro >=0.4+    , avro >=0.5 && <0.6     , base >=4.7 && <5     , binary     , bytestring@@ -63,7 +63,7 @@   ghc-options: -threaded -rtsopts -with-rtsopts=-N   build-depends:       aeson-    , avro+    , avro  >=0.5 && <0.6     , base     , binary     , bytestring@@ -92,7 +92,7 @@   build-depends:       QuickCheck     , aeson-    , avro+    , avro  >=0.5 && <0.6     , base     , binary     , bytestring

@@ -26,7 +26,7 @@                 -> Subject                 -> ByteString                 -> m (Either SchemaRegistryError (Maybe SchemaId))-propagateSchema sr subj bs = do+propagateSchema sr subj bs =   case extractSchemaId bs of     Nothing -> return $ Right Nothing     Just (sid, _) -> do

@@ -1,53 +1,57 @@ {-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeApplications    #-} module Kafka.Avro.Decode (   DecodeError(..)-, decodeWithSchema, extractSchemaId+, decode+, decodeWithSchema+, extractSchemaId ) where -import           Control.Arrow             (left)-import           Control.Monad.IO.Class    (MonadIO)-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           Control.Arrow              (left)+import           Control.Monad.IO.Class     (MonadIO)+import           Control.Monad.Trans.Except+import           Data.Avro                  (FromAvro, HasAvroSchema (..), Schema, decodeValueWithSchema, deconflict)+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           Data.Tagged                (untag) import           Kafka.Avro.SchemaRegistry  data DecodeError = DecodeRegistryError SchemaRegistryError                  | BadPayloadNoSchemaId                  | DecodeError Schema String+                 | IncompatibleSchema Schema String                  deriving (Show, Eq)  -- | Decodes a provided Avro-encoded value. -- The serialised value is expected to be in a "confluent-compatible" format -- where the "real" value bytestring is prepended with extra 5 bytes: -- a "magic" byte and 4 bytes representing the schema ID.-decodeWithSchema :: (MonadIO m, FromAvro a)-                 => SchemaRegistry-                 -> ByteString-                 -> m (Either DecodeError a)-decodeWithSchema sr bs =+decode :: forall a m. (MonadIO m, HasAvroSchema a, FromAvro a)+  => SchemaRegistry+  -> ByteString+  -> m (Either DecodeError a)+decode sr = decodeWithSchema sr (untag @a schema)+{-# INLINE decode #-}++decodeWithSchema :: forall a m. (MonadIO m, FromAvro a)+  => SchemaRegistry+  -> Schema+  -> ByteString+  -> m (Either DecodeError a)+decodeWithSchema sr readerSchema bs =   case schemaData of     Left err -> return $ Left err-    Right (sid, payload) -> do-      res <- left DecodeRegistryError <$> loadSchema sr sid-      return $ res >>= flip decodeWithDeconflict payload+    Right (sid, payload) -> runExceptT $ do+      writerSchema  <- withError DecodeRegistryError (loadSchema sr sid)+      readSchema    <- withPureError (IncompatibleSchema writerSchema) $ deconflict writerSchema readerSchema+      withPureError (DecodeError writerSchema) (decodeValueWithSchema readSchema payload)   where     schemaData = maybe (Left BadPayloadNoSchemaId) Right (extractSchemaId bs)--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)+    withError f = withExceptT f . ExceptT+    withPureError f = withError f . pure  extractSchemaId :: ByteString -> Maybe (SchemaId, ByteString) extractSchemaId bs = do@@ -59,10 +63,4 @@   let ints =  fromIntegral <$> [w4, w3, w2, w1] :: [Int32]   let int  =  sum $ zipWith shiftL ints [0, 8, 16, 24]   return (SchemaId int, b4)--resultToEither :: Schema -> A.Result a -> Either String a-resultToEither sc res = case res of-  Success a -> Right a-  Error msg -> Left msg- 

@@ -1,14 +1,21 @@-{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE OverloadedStrings   #-}+{-# LANGUAGE ScopedTypeVariables #-} module Kafka.Avro.Encode-( encodeKey, encodeValue-, keySubject, valueSubject+( EncodeError(..)+, encodeKey+, encodeValue+, encode++, encodeKeyWithSchema+, encodeValueWithSchema , encodeWithSchema-, EncodeError(..)++, keySubject, valueSubject ) where  import           Control.Monad.IO.Class    (MonadIO)-import           Data.Avro                 as A (ToAvro, encode, schemaOf)-import           Data.Avro.Schema          (Schema)+import           Data.Avro                 (HasAvroSchema, Schema, ToAvro, schemaOf)+import qualified Data.Avro                 as A import qualified Data.Binary               as B import           Data.Bits                 (shiftL) import           Data.ByteString.Lazy      (ByteString)@@ -21,42 +28,79 @@  keySubject :: Subject -> Subject keySubject (Subject subj) = Subject (subj <> "-key")+{-# INLINE keySubject #-}  valueSubject :: Subject -> Subject valueSubject (Subject subj) = Subject (subj <> "-value")+{-# INLINE valueSubject #-}  -- | Encodes a provided value as a message key. -- -- Registers the schema in SchemaRegistry with "<subject>-key" subject.-encodeKey :: (MonadIO m, ToAvro a)-          => SchemaRegistry-          -> Subject-          -> a-          -> m (Either EncodeError ByteString)-encodeKey sr subj = encodeWithSchema sr (keySubject subj)+encodeKey :: (MonadIO m, HasAvroSchema a, ToAvro a)+  => SchemaRegistry+  -> Subject+  -> a+  -> m (Either EncodeError ByteString)+encodeKey sr subj = encode sr (keySubject subj)+{-# INLINE encodeKey #-} +-- | Encodes a provided value as a message key.+--+-- Registers the schema in SchemaRegistry with "<subject>-key" subject.+encodeKeyWithSchema :: (MonadIO m, ToAvro a)+  => SchemaRegistry+  -> Subject+  -> Schema+  -> a+  -> m (Either EncodeError ByteString)+encodeKeyWithSchema sr subj = encodeWithSchema sr (keySubject subj)+{-# INLINE encodeKeyWithSchema #-}+ -- | Encodes a provided value as a message value. -- -- Registers the schema in SchemaRegistry with "<subject>-value" subject.-encodeValue :: (MonadIO m, ToAvro a)-            => SchemaRegistry-            -> Subject-            -> a-            -> m (Either EncodeError ByteString)-encodeValue sr subj = encodeWithSchema sr (valueSubject subj)+encodeValue :: (MonadIO m, HasAvroSchema a, ToAvro a)+  => SchemaRegistry+  -> Subject+  -> a+  -> m (Either EncodeError ByteString)+encodeValue sr subj = encode sr (valueSubject subj)+{-# INLINE encodeValue #-} +-- | Encodes a provided value as a message value.+--+-- Registers the schema in SchemaRegistry with "<subject>-value" subject.+encodeValueWithSchema :: (MonadIO m, ToAvro a)+  => SchemaRegistry+  -> Subject+  -> Schema+  -> a+  -> m (Either EncodeError ByteString)+encodeValueWithSchema sr subj = encodeWithSchema sr (valueSubject subj)+{-# INLINE encodeValueWithSchema #-}++encode :: (MonadIO m, HasAvroSchema a, ToAvro a)+  => SchemaRegistry+  -> Subject+  -> a+  -> m (Either EncodeError ByteString)+encode sr subj a = encodeWithSchema sr subj (schemaOf a) a+{-# INLINE encode #-}+ -- | Encodes a provided value into Avro -- and registers value's schema in SchemaRegistry.-encodeWithSchema :: (MonadIO m, ToAvro a)-                 => SchemaRegistry-                 -> Subject-                 -> a-                 -> m (Either EncodeError ByteString)-encodeWithSchema sr subj a = do-  mbSid <- sendSchema sr subj (schemaOf a)+encodeWithSchema :: forall a m. (MonadIO m, ToAvro a)+  => SchemaRegistry+  -> Subject+  -> Schema+  -> a+  -> m (Either EncodeError ByteString)+encodeWithSchema sr subj sch a = do+  mbSid <- sendSchema sr subj sch   case mbSid of     Left err  -> return . Left . EncodeRegistryError $ err-    Right sid -> return . Right $ appendSchemaId sid (encode a)+    Right sid -> return . Right $ appendSchemaId sid (A.encodeValueWithSchema sch a)   appendSchemaId :: SchemaId -> ByteString -> ByteString

@@ -24,7 +24,7 @@ import           Control.Monad.IO.Class  (MonadIO, liftIO) import           Data.Aeson import           Data.Aeson.Types        (typeMismatch)-import           Data.Avro.Schema        (Schema, Type (..), typeName)+import           Data.Avro.Schema.Schema (Schema (..), typeName) import           Data.Bifunctor          (bimap) import           Data.Cache              as C import           Data.Hashable           (Hashable)