hw-kafka-avro 4.0.1 → 6.1.2
raw patch · 7 files changed
Files
- example/Main.hs +4/−7
- example/Message.hs +19/−35
- hw-kafka-avro.cabal +84/−78
- src/Kafka/Avro.hs +1/−1
- src/Kafka/Avro/Decode.hs +32/−34
- src/Kafka/Avro/Encode.hs +70/−33
- src/Kafka/Avro/SchemaRegistry.hs +215/−79
example/Main.hs view
@@ -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
example/Message.hs view
@@ -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- ]
hw-kafka-avro.cabal view
@@ -1,20 +1,20 @@-cabal-version: 1.12+cabal-version: 2.4 -name: hw-kafka-avro-version: 4.0.1-synopsis: Avro support for Kafka infrastructure-description: Avro support for Kafka infrastructure.-category: Services-homepage: https://github.com/haskell-works/hw-kafka-avro#readme-bug-reports: https://github.com/haskell-works/hw-kafka-avro/issues-author: Alexey Raga-maintainer: alexey.raga@gmail.com-copyright: Alexey Raga-license: BSD3-license-file: LICENSE-build-type: Simple-extra-source-files:- README.md+name: hw-kafka-avro+version: 6.1.2+synopsis: Avro support for Kafka infrastructure+description: Avro support for Kafka infrastructure.+category: Services+homepage: https://github.com/haskell-works/hw-kafka-avro#readme+bug-reports: https://github.com/haskell-works/hw-kafka-avro/issues+author: Alexey Raga+maintainer: alexey.raga@gmail.com+copyright: Alexey Raga+license: BSD-3-Clause+license-file: LICENSE+build-type: Simple+tested-with: GHC == 9.10.1, GHC == 9.8.2, GHC == 9.6.6+extra-source-files: README.md source-repository head type: git@@ -25,88 +25,94 @@ manual: True default: False +common base { build-depends: base >= 4 && < 5 }+common avro { build-depends: avro >= 0.6.0.2 && < 0.7 }++common aeson { build-depends: aeson >= 2.0.1.0 }+common binary { build-depends: binary }+common bytestring { build-depends: bytestring }+common containers { build-depends: containers }+common cache { build-depends: cache }+common fail { build-depends: fail }+common hashable { build-depends: hashable }+common hedgehog { build-depends: hedgehog }+common hw-hspec-hedgehog { build-depends: hw-hspec-hedgehog }+common hspec { build-depends: hspec }+common lens { build-depends: lens }+common mtl { build-depends: mtl }+common semigroups { build-depends: semigroups }+common tagged { build-depends: tagged }+common http-client { build-depends: http-client }+common http-types { build-depends: http-types }+common text { build-depends: text >= 1.2.3 && < 1.3 || >= 2.0 && < 2.2 }+common transformers { build-depends: transformers >= 0.5.6.2 && < 0.7 }+common unordered-containers { build-depends: unordered-containers }+common safe-exceptions { build-depends: safe-exceptions >= 0.1.7.2 && < 0.2 }+common wreq { build-depends: wreq }++common config+ default-language: Haskell2010+ library+ import: base+ , aeson+ , avro+ , binary+ , bytestring+ , cache+ , containers+ , hashable+ , http-client+ , http-types+ , lens+ , mtl+ , safe-exceptions+ , semigroups+ , tagged+ , text+ , transformers+ , unordered-containers+ , wreq+ , config exposed-modules: Kafka.Avro Kafka.Avro.Decode Kafka.Avro.Encode Kafka.Avro.SchemaRegistry hs-source-dirs: src- build-depends:- aeson- , avro >=0.4- , base >=4.7 && <5- , binary- , bytestring- , cache- , containers- , errors- , hashable- , http-client- , http-types- , lens- , mtl- , pure-zlib >=0.6- , semigroups- , tagged- , text- , transformers- , unordered-containers- , wreq- default-language: Haskell2010 executable kafka-avro-example+ import: base+ , aeson+ , avro+ , bytestring+ , cache+ , containers+ , lens+ , mtl+ , semigroups+ , text+ , transformers+ , unordered-containers+ , config main-is: Main.hs other-modules: Message hs-source-dirs: example ghc-options: -threaded -rtsopts -with-rtsopts=-N- build-depends:- aeson- , avro- , base- , binary- , bytestring- , cache- , containers- , errors- , hashable- , http-client- , hw-kafka-avro- , mtl- , pure-zlib >=0.6- , semigroups- , text- , transformers- , unordered-containers+ build-depends: hw-kafka-avro if !(flag(examples)) buildable: False- default-language: Haskell2010 test-suite kafka-avro-test+ import: base+ , config+ , hedgehog+ , hspec+ , hw-hspec-hedgehog type: exitcode-stdio-1.0 main-is: Spec.hs other-modules: hs-source-dirs: test ghc-options: -threaded -rtsopts -with-rtsopts=-N- build-depends:- QuickCheck- , aeson- , avro- , base- , binary- , bytestring- , cache- , containers- , errors- , hashable- , hspec- , http-client- , hw-kafka-avro- , mtl- , pure-zlib >=0.6- , semigroups- , text- , transformers- , unordered-containers- default-language: Haskell2010+ build-depends: hw-kafka-avro
src/Kafka/Avro.hs view
@@ -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
src/Kafka/Avro/Decode.hs view
@@ -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-
src/Kafka/Avro/Encode.hs view
@@ -1,14 +1,23 @@-{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# OPTIONS_GHC -Wno-unrecognised-pragmas #-}+{-# HLINT ignore "Use newtype instead of data" #-} 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 +30,70 @@ 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)+-- | Encodes a provided value as a message key with "<subject>-key" subject.+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 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)+-- | Encodes a provided value as a message key 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 with "<subject>-value" subject.+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 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
src/Kafka/Avro/SchemaRegistry.hs view
@@ -1,44 +1,61 @@ {-# LANGUAGE CPP #-} {-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-}-{-# LANGUAGE LambdaCase #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TupleSections #-} module Kafka.Avro.SchemaRegistry ( schemaRegistry, loadSchema, sendSchema+, schemaRegistry_+, schemaRegistryWithHeaders , loadSubjectSchema , getGlobalConfig, getSubjectConfig , getVersions, isCompatible , getSubjects+, defaultSchemaRegistryConfig+, cfgAuth+, cfgHeaders+, cfgAutoRegisterSchemas , SchemaId(..), Subject(..)+, SchemaRegistryConfig , SchemaRegistry, SchemaRegistryError(..) , Schema(..) , Compatibility(..), Version(..) ) where -import Control.Arrow (first)-import Control.Exception (throwIO)-import Control.Lens (view, (&), (.~), (^.))-import Control.Monad (void)-import Control.Monad.IO.Class (MonadIO, liftIO)+import Control.Arrow (first)+import Control.Exception (SomeException (SomeException), throwIO)+import Control.Exception.Safe (MonadCatch, try)+import Control.Lens (view, (%~), (&), (.~), (^.))+import Control.Monad (void)+import Control.Monad.Except (liftEither)+import Control.Monad.IO.Class (MonadIO, liftIO)+import Control.Monad.Trans.Except (ExceptT (ExceptT), except, runExceptT, withExceptT) import Data.Aeson-import Data.Aeson.Types (typeMismatch)-import Data.Avro.Schema (Schema, Type (..), typeName)-import Data.Bifunctor (bimap)-import Data.Cache as C-import Data.Hashable (Hashable)-import qualified Data.HashMap.Lazy as HM-import Data.Int (Int32)-import Data.String (IsString)-import Data.Text (Text, append, cons, unpack)-import qualified Data.Text.Encoding as Text-import qualified Data.Text.Lazy.Encoding as LText-import Data.Word (Word32)-import GHC.Exception (SomeException, displayException, fromException)-import GHC.Generics (Generic)-import Network.HTTP.Client (HttpException (..), HttpExceptionContent (..), Manager, defaultManagerSettings, newManager)-import qualified Network.Wreq as Wreq+import qualified Data.Aeson.Key as A+import qualified Data.Aeson.KeyMap as KM+import Data.Aeson.Types (typeMismatch)+import Data.Avro.Schema.Schema (Schema (..), typeName)+import Data.Bifunctor (bimap)+import Data.Cache as C+import Data.Foldable (traverse_)+import Data.Functor (($>))+import Data.Hashable (Hashable)+import qualified Data.HashMap.Lazy as HM+import Data.Int (Int32)+import Data.List (find)+import Data.String (IsString)+import Data.Text (Text, append, cons, unpack)+import qualified Data.Text.Encoding as Text+import qualified Data.Text.Lazy.Encoding as LText+import Data.Word (Word32)+import GHC.Exception (SomeException, displayException, fromException)+import GHC.Generics (Generic)+import Network.HTTP.Client (HttpException (..), HttpExceptionContent (..), Manager, defaultManagerSettings, newManager, responseStatus)+import Network.HTTP.Types.Header (Header)+import Network.HTTP.Types.Status (notFound404)+import qualified Network.Wreq as Wreq newtype SchemaId = SchemaId { unSchemaId :: Int32} deriving (Eq, Ord, Show, Hashable) newtype SchemaName = SchemaName Text deriving (Eq, Ord, IsString, Show, Hashable)@@ -55,47 +72,95 @@ | BackwardCompatibility deriving (Eq, Show, Ord) +data SchemaRegistryConfig = SchemaRegistryConfig+ { cAuth :: Maybe Wreq.Auth+ , cExtraHeaders :: [Header]+ , cAutoRegisterSchemas :: Bool+ }+ data SchemaRegistry = SchemaRegistry { srCache :: Cache SchemaId Schema , srReverseCache :: Cache (Subject, SchemaName) SchemaId , srBaseUrl :: String+ , srConfig :: SchemaRegistryConfig } data SchemaRegistryError = SchemaRegistryConnectError String | SchemaRegistryLoadError SchemaId | SchemaRegistrySchemaNotFound SchemaId+ | SchemaRegistrySubjectNotFound Subject+ | SchemaRegistryNoCompatibleSchemaFound Schema+ | SchemaRegistryUrlNotFound String | SchemaRegistrySendError String+ | SchemaRegistryCacheError deriving (Show, Eq) +defaultSchemaRegistryConfig :: SchemaRegistryConfig+defaultSchemaRegistryConfig = SchemaRegistryConfig+ { cAuth = Nothing+ , cExtraHeaders = []+ , cAutoRegisterSchemas = True+ }+ schemaRegistry :: MonadIO m => String -> m SchemaRegistry-schemaRegistry url = liftIO $- SchemaRegistry- <$> newCache Nothing- <*> newCache Nothing- <*> pure url+schemaRegistry = schemaRegistry_ Nothing +schemaRegistry_ :: MonadIO m => Maybe Wreq.Auth -> String -> m SchemaRegistry+schemaRegistry_ auth = schemaRegistryWithHeaders auth []++schemaRegistryWithHeaders :: MonadIO m => Maybe Wreq.Auth -> [Header] -> String -> m SchemaRegistry+schemaRegistryWithHeaders auth headers url+ = schemaRegistryWithConfig url $ cfgAuth auth $ cfgHeaders headers defaultSchemaRegistryConfig++schemaRegistryWithConfig :: MonadIO m => String -> SchemaRegistryConfig -> m SchemaRegistry+schemaRegistryWithConfig url config = liftIO $+ SchemaRegistry+ <$> newCache Nothing+ <*> newCache Nothing+ <*> pure url+ <*> pure config++-- | Add authentication options+cfgAuth :: Maybe Wreq.Auth -> SchemaRegistryConfig -> SchemaRegistryConfig+cfgAuth auth config = config { cAuth = auth }++-- | Add extra headers+cfgHeaders :: [Header] -> SchemaRegistryConfig -> SchemaRegistryConfig+cfgHeaders headers config = config { cExtraHeaders = headers }++-- | Set whether to auto-publish schemas+-- If set to 'False', encoding will fail if there is no compatible schema+-- in the schema registy.+-- This is equivalent to the confluent 'auto.register.schemas' option.+cfgAutoRegisterSchemas :: Bool -> SchemaRegistryConfig -> SchemaRegistryConfig+cfgAutoRegisterSchemas autoRegisterSchemas config = config { cAutoRegisterSchemas = autoRegisterSchemas }+ loadSchema :: MonadIO m => SchemaRegistry -> SchemaId -> m (Either SchemaRegistryError Schema) loadSchema sr sid = do sc <- cachedSchema sr sid case sc of Just s -> return (Right s) Nothing -> liftIO $ do- res <- getSchemaById (srBaseUrl sr) sid+ res <- getSchemaById sr sid traverse ((\schema -> schema <$ cacheSchema sr sid schema) . unRegisteredSchema) res loadSubjectSchema :: MonadIO m => SchemaRegistry -> Subject -> Version -> m (Either SchemaRegistryError Schema) loadSubjectSchema sr (Subject sbj) (Version version) = do- let url = (srBaseUrl sr) ++ "/subjects/" ++ unpack sbj ++ "/versions/" ++ show version- resp <- liftIO $ Wreq.getWith wreqOpts url- wrapped <- pure $ bimap wrapError (view Wreq.responseBody) (Wreq.asValue resp)+ let url = srBaseUrl sr ++ "/subjects/" ++ unpack sbj ++ "/versions/" ++ show version+ respE <- liftIO . try $ Wreq.getWith (wreqOpts sr) url+ case respE of+ Left exc -> pure . Left $ wrapErrorWithUrl url exc+ Right resp -> do - schema <- getData "schema" wrapped- schemaId <- getData "id" wrapped+ let wrapped = bimap wrapError (view Wreq.responseBody) (Wreq.asValue resp)+ schema <- getData "schema" wrapped+ schemaId <- getData "id" wrapped - case (,) <$> schema <*> schemaId of- Left err -> return $ Left err- Right ((RegisteredSchema schema), schemaId) -> cacheSchema sr schemaId schema *> (return $ Right schema)+ case (,) <$> schema <*> schemaId of+ Left err -> return $ Left err+ Right (RegisteredSchema schema, schemaId) -> cacheSchema sr schemaId schema $> Right schema where+ getData :: (MonadIO m, FromJSON a) => String -> Either e Value -> m (Either e a) getData key = either (pure . Left) (viewData key) @@ -109,37 +174,68 @@ Success a -> Right a Error e -> Left e ++-- | Get the schema ID.+-- If the 'SchemaRegistry' is configured to auto-register schemas,+-- this posts the schema to the schema registry server.+-- Otherwise, this searches for a compatible schema and returns a 'SchemaRegistryNoCompatibleSchemaFound'+-- if none is found. sendSchema :: MonadIO m => SchemaRegistry -> Subject -> Schema -> m (Either SchemaRegistryError SchemaId) sendSchema sr subj sc = do+ let schemaName = fullTypeName sc sid <- cachedId sr subj schemaName case sid of Just sid' -> return (Right sid')- Nothing -> do- res <- liftIO $ putSchema (srBaseUrl sr) subj (RegisteredSchema sc)- void $ traverse (cacheId sr subj schemaName) res- void $ traverse (\sid' -> cacheSchema sr sid' sc) res- pure res- where- schemaName = fullTypeName sc+ Nothing -> if cAutoRegisterSchemas (srConfig sr)+ then registerSchema sr subj sc+ else getCompatibleSchema sr subj sc +registerSchema :: MonadIO m => SchemaRegistry -> Subject -> Schema -> m (Either SchemaRegistryError SchemaId)+registerSchema sr subj sc = do+ let schemaName = fullTypeName sc+ res <- liftIO $ putSchema sr subj (RegisteredSchema sc)+ traverse_ (cacheId sr subj schemaName) res+ traverse_ (\sid' -> cacheSchema sr sid' sc) res+ pure res++getCompatibleSchema :: MonadIO m => SchemaRegistry -> Subject -> Schema -> m (Either SchemaRegistryError SchemaId)+getCompatibleSchema sr subj sc = liftIO . runExceptT $ do+ let schemaName = fullTypeName sc+ versions <- liftEither =<< getVersions sr subj+ compatibilites <- liftEither . sequenceA + =<< traverse (\ver -> fmap (,ver) <$> isCompatible sr subj ver sc) versions+ let mCompatibleVersion = snd <$> find fst compatibilites+ compatibleVersion <- liftEither + $ case mCompatibleVersion of+ Just version -> Right version+ Nothing -> Left $ SchemaRegistryNoCompatibleSchemaFound sc+ _ <- liftEither =<< loadSubjectSchema sr subj compatibleVersion -- caches the schema ID+ mSid <- cachedId sr subj schemaName+ liftEither $ case mSid of+ Just sid' -> pure sid'+ Nothing -> Left SchemaRegistryCacheError+ getVersions :: MonadIO m => SchemaRegistry -> Subject -> m (Either SchemaRegistryError [Version])-getVersions sr (Subject sbj) = do- let url = (srBaseUrl sr) ++ "/subjects/" ++ unpack sbj ++ "/versions"- resp <- liftIO $ Wreq.getWith wreqOpts url- pure $ bimap wrapError (fmap Version . view Wreq.responseBody) (Wreq.asJSON resp)+getVersions sr subj@(Subject sbj) = liftIO . runExceptT $ do+ let url = srBaseUrl sr ++ "/subjects/" ++ unpack sbj ++ "/versions"+ resp <- tryWith (wrapErrorWithSubject subj) $ Wreq.getWith (wreqOpts sr) url+ except $ bimap wrapError (fmap Version . view Wreq.responseBody) (Wreq.asJSON resp) isCompatible :: MonadIO m => SchemaRegistry -> Subject -> Version -> Schema -> m (Either SchemaRegistryError Bool) isCompatible sr (Subject sbj) (Version version) schema = do- let url = (srBaseUrl sr) ++ "/compatibility/subjects/" ++ unpack sbj ++ "/versions/" ++ show version- resp <- liftIO $ Wreq.postWith wreqOpts url (toJSON $ RegisteredSchema schema)- wrapped <- pure $ bimap wrapError (view Wreq.responseBody) (Wreq.asValue resp)- either (return . Left) getCompatibility wrapped+ let url = srBaseUrl sr ++ "/compatibility/subjects/" ++ unpack sbj ++ "/versions/" ++ show version+ respE <- liftIO . try $ Wreq.postWith (wreqOpts sr) url (toJSON $ RegisteredSchema schema)+ case respE of+ Left exc -> pure . Left $ wrapErrorWithUrl url exc+ Right resp -> do+ let wrapped = bimap wrapError (view Wreq.responseBody) (Wreq.asValue resp)+ either (return . Left) getCompatibility wrapped where getCompatibility :: MonadIO m => Value -> m (Either e Bool) getCompatibility = liftIO . maybe (throwIO $ Wreq.JSONError "Missing key 'is_compatible' in Schema Registry response") (return . return) . viewCompatibility viewCompatibility :: Value -> Maybe Bool- viewCompatibility (Object obj) = HM.lookup "is_compatible" obj >>= toBool+ viewCompatibility (Object obj) = KM.lookup "is_compatible" obj >>= toBool viewCompatibility _ = Nothing toBool :: Value -> Maybe Bool@@ -148,40 +244,50 @@ getGlobalConfig :: MonadIO m => SchemaRegistry -> m (Either SchemaRegistryError Compatibility) getGlobalConfig sr = do- let url = (srBaseUrl sr) ++ "/config"- resp <- liftIO $ Wreq.getWith wreqOpts url- pure $ bimap wrapError (view Wreq.responseBody) (Wreq.asJSON resp)+ let url = srBaseUrl sr ++ "/config"+ respE <- liftIO . try $ Wreq.getWith (wreqOpts sr) url+ pure $ case respE of+ Left exc -> Left $ wrapError exc+ Right resp -> bimap wrapError (view Wreq.responseBody) (Wreq.asJSON resp) getSubjectConfig :: MonadIO m => SchemaRegistry -> Subject -> m (Either SchemaRegistryError Compatibility)-getSubjectConfig sr (Subject sbj) = do- let url = (srBaseUrl sr) ++ "/config/" ++ unpack sbj- resp <- liftIO $ Wreq.getWith wreqOpts url- pure $ bimap wrapError (view Wreq.responseBody) (Wreq.asJSON resp)+getSubjectConfig sr subj@(Subject sbj) = liftIO . runExceptT $ do+ let url = srBaseUrl sr ++ "/config/" ++ unpack sbj+ resp <- tryWith (wrapErrorWithSubject subj) $ Wreq.getWith (wreqOpts sr) url+ except $ bimap wrapError (view Wreq.responseBody) (Wreq.asJSON resp) getSubjects :: MonadIO m => SchemaRegistry -> m (Either SchemaRegistryError [Subject])-getSubjects sr = do- let url = (srBaseUrl sr) ++ "/subjects"- resp <- liftIO $ Wreq.getWith wreqOpts url- pure $ bimap wrapError (fmap Subject . view Wreq.responseBody) (Wreq.asJSON resp)+getSubjects sr = liftIO . runExceptT $ do+ let url = srBaseUrl sr ++ "/subjects"+ resp <- tryWith wrapError $ Wreq.getWith (wreqOpts sr) url+ except $ bimap wrapError (fmap Subject . view Wreq.responseBody) (Wreq.asJSON resp) ------------------ PRIVATE: HELPERS -------------------------------------------- -wreqOpts :: Wreq.Options-wreqOpts =- let accept = ["application/vnd.schemaregistry.v1+json", "application/vnd.schemaregistry+json", "application/json"]- in Wreq.defaults & Wreq.header "Accept" .~ accept+wreqOpts :: SchemaRegistry -> Wreq.Options+wreqOpts sr =+ let+ accept = ["application/vnd.schemaregistry.v1+json", "application/vnd.schemaregistry+json", "application/json"]+ acceptHeader = Wreq.header "Accept" .~ accept+ authHeader = Wreq.auth .~ cAuth (srConfig sr)+ extraHeaders = Wreq.headers %~ (++ cExtraHeaders (srConfig sr))+ in Wreq.defaults & acceptHeader & authHeader & extraHeaders -getSchemaById :: String -> SchemaId -> IO (Either SchemaRegistryError RegisteredSchema)-getSchemaById baseUrl sid@(SchemaId i) = do- let schemaUrl = baseUrl ++ "/schemas/ids/" ++ show i- resp <- Wreq.getWith wreqOpts schemaUrl- pure $ bimap (const (SchemaRegistryLoadError sid)) (view Wreq.responseBody) (Wreq.asJSON resp)+getSchemaById :: SchemaRegistry -> SchemaId -> IO (Either SchemaRegistryError RegisteredSchema)+getSchemaById sr sid@(SchemaId i) = runExceptT $ do+ let+ baseUrl = srBaseUrl sr+ schemaUrl = baseUrl ++ "/schemas/ids/" ++ show i+ resp <- tryWith (wrapErrorWithSchemaId sid) $ Wreq.getWith (wreqOpts sr) schemaUrl+ except $ bimap (const (SchemaRegistryLoadError sid)) (view Wreq.responseBody) (Wreq.asJSON resp) -putSchema :: String -> Subject -> RegisteredSchema -> IO (Either SchemaRegistryError SchemaId)-putSchema baseUrl (Subject sbj) schema = do- let schemaUrl = baseUrl ++ "/subjects/" ++ unpack sbj ++ "/versions"- resp <- Wreq.postWith wreqOpts schemaUrl (toJSON schema)- pure $ bimap wrapError (view Wreq.responseBody) (Wreq.asJSON resp)+putSchema :: SchemaRegistry -> Subject -> RegisteredSchema -> IO (Either SchemaRegistryError SchemaId)+putSchema sr subj@(Subject sbj) schema = runExceptT $ do+ let+ baseUrl = srBaseUrl sr+ schemaUrl = baseUrl ++ "/subjects/" ++ unpack sbj ++ "/versions"+ resp <- tryWith (wrapErrorWithSubject subj) $ Wreq.postWith (wreqOpts sr) schemaUrl (toJSON schema)+ except $ bimap wrapError (view Wreq.responseBody) (Wreq.asJSON resp) fromHttpError :: HttpException -> (HttpExceptionContent -> SchemaRegistryError) -> SchemaRegistryError fromHttpError err f = case err of@@ -192,9 +298,9 @@ HttpExceptionRequest _ ConnectionClosed -> SchemaRegistryConnectError (displayException err) HttpExceptionRequest _ (InvalidDestinationHost _) -> SchemaRegistryConnectError (displayException err) HttpExceptionRequest _ TlsNotSupported -> SchemaRegistryConnectError (displayException err)-#if MIN_VERSION_http_client(0,5,7)+ HttpExceptionRequest _ (InvalidProxySettings _) -> SchemaRegistryConnectError (displayException err)-#endif+ HttpExceptionRequest _ err' -> f err' wrapError :: SomeException -> SchemaRegistryError@@ -202,6 +308,23 @@ Nothing -> SchemaRegistrySendError (displayException someErr) Just httpErr -> fromHttpError httpErr (\_ -> SchemaRegistrySendError (displayException someErr)) +wrapErrorWithSchemaId :: SchemaId -> SomeException -> SchemaRegistryError+wrapErrorWithSchemaId = wrapErrorWith SchemaRegistrySchemaNotFound++wrapErrorWithSubject :: Subject -> SomeException -> SchemaRegistryError+wrapErrorWithSubject = wrapErrorWith SchemaRegistrySubjectNotFound++wrapErrorWithUrl :: String -> SomeException -> SchemaRegistryError+wrapErrorWithUrl = wrapErrorWith SchemaRegistryUrlNotFound++wrapErrorWith :: (a -> SchemaRegistryError) -> a -> SomeException -> SchemaRegistryError+wrapErrorWith mkError x exception = case fromException exception of+ Just (HttpExceptionRequest _ (StatusCodeException response _)) | responseStatus response == notFound404 -> mkError x+ _ -> wrapError exception++tryWith :: MonadCatch m => (SomeException -> e) -> m a -> ExceptT e m a+tryWith wrapException = withExceptT wrapException . ExceptT . try+ --------------------------------------------------------------------- fullTypeName :: Schema -> SchemaName fullTypeName r = SchemaName $ typeName r@@ -247,3 +370,16 @@ "FORWARD" -> return $ ForwardCompatibility "BACKWARD" -> return $ BackwardCompatibility _ -> typeMismatch "Compatibility" compatibility+++++++++++++