hw-kafka-avro 5.1.0 → 6.0.0
raw patch · 3 files changed
+146/−106 lines, 3 filesdep +hedgehogdep +hw-hspec-hedgehogdep +safe-exceptionsdep −QuickCheckdep −errorsdep −pure-zlibdep ~aesondep ~avrodep ~basePVP ok
version bump matches the API change (PVP)
Dependencies added: hedgehog, hw-hspec-hedgehog, safe-exceptions
Dependencies removed: QuickCheck, errors, pure-zlib
Dependency ranges changed: aeson, avro, base, text, transformers
API changes (from Hackage documentation)
- Kafka.Avro.SchemaRegistry: pattern Int' :: () => () => Schema
+ Kafka.Avro.SchemaRegistry: pattern Bytes' :: Schema
+ Kafka.Avro.SchemaRegistry: pattern Int' :: Schema
+ Kafka.Avro.SchemaRegistry: pattern Long' :: Schema
+ Kafka.Avro.SchemaRegistry: pattern String' :: Schema
+ Kafka.Avro.SchemaRegistry: schemaRegistry_ :: MonadIO m => Maybe Auth -> String -> m SchemaRegistry
Files
- hw-kafka-avro.cabal +70/−64
- src/Kafka/Avro/Encode.hs +2/−0
- src/Kafka/Avro/SchemaRegistry.hs +74/−42
hw-kafka-avro.cabal view
@@ -1,7 +1,7 @@ cabal-version: 2.4 name: hw-kafka-avro-version: 5.1.0+version: 6.0.0 synopsis: Avro support for Kafka infrastructure description: Avro support for Kafka infrastructure. category: Services@@ -13,7 +13,7 @@ license: BSD-3-Clause license-file: LICENSE build-type: Simple-tested-with: GHC == 8.10.1, GHC == 8.8.3, GHC == 8.6.5, GHC == 8.4.4+tested-with: GHC == 9.2.2, GHC == 9.0.2, GHC == 8.10.7, GHC == 8.8.4, GHC == 8.6.5 extra-source-files: README.md source-repository head@@ -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.1 }+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.5 && <0.6- , 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 >=0.5 && <0.6- , 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 >=0.5 && <0.6- , 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/Encode.hs view
@@ -1,5 +1,7 @@ {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-}+{-# OPTIONS_GHC -Wno-unrecognised-pragmas #-}+{-# HLINT ignore "Use newtype instead of data" #-} module Kafka.Avro.Encode ( EncodeError(..) , encodeKey
src/Kafka/Avro/SchemaRegistry.hs view
@@ -1,12 +1,12 @@ {-# LANGUAGE CPP #-} {-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-}-{-# LANGUAGE LambdaCase #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-} module Kafka.Avro.SchemaRegistry ( schemaRegistry, loadSchema, sendSchema+, schemaRegistry_ , loadSubjectSchema , getGlobalConfig, getSubjectConfig , getVersions, isCompatible@@ -19,12 +19,18 @@ import Control.Arrow (first) import Control.Exception (throwIO)+import Control.Exception.Safe (try) import Control.Lens (view, (&), (.~), (^.)) import Control.Monad (void) 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 qualified Data.Aeson.Key as A+import qualified Data.Aeson.KeyMap as KM import Data.Avro.Schema.Schema (Schema (..), typeName)+import Data.Foldable (traverse_)+import Data.Functor (($>)) import Data.Bifunctor (bimap) import Data.Cache as C import Data.Hashable (Hashable)@@ -59,6 +65,7 @@ { srCache :: Cache SchemaId Schema , srReverseCache :: Cache (Subject, SchemaName) SchemaId , srBaseUrl :: String+ , srAuth :: Maybe Wreq.Auth } data SchemaRegistryError = SchemaRegistryConnectError String@@ -68,33 +75,38 @@ deriving (Show, Eq) 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 url = liftIO $+ SchemaRegistry+ <$> newCache Nothing+ <*> newCache Nothing+ <*> pure url+ <*> pure auth++ 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+ resp <- liftIO $ Wreq.getWith (wreqOpts sr) url+ 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)+ 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)@@ -115,31 +127,31 @@ 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+ res <- liftIO $ putSchema sr subj (RegisteredSchema sc)+ traverse_ (cacheId sr subj schemaName) res+ traverse_ (\sid' -> cacheSchema sr sid' sc) res pure res where schemaName = fullTypeName sc 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+ let url = srBaseUrl sr ++ "/subjects/" ++ unpack sbj ++ "/versions"+ resp <- liftIO $ Wreq.getWith (wreqOpts sr) url pure $ 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)+ let url = srBaseUrl sr ++ "/compatibility/subjects/" ++ unpack sbj ++ "/versions/" ++ show version+ resp <- liftIO $ Wreq.postWith (wreqOpts sr) url (toJSON $ RegisteredSchema schema)+ 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 +160,47 @@ getGlobalConfig :: MonadIO m => SchemaRegistry -> m (Either SchemaRegistryError Compatibility) getGlobalConfig sr = do- let url = (srBaseUrl sr) ++ "/config"- resp <- liftIO $ Wreq.getWith wreqOpts url+ let url = srBaseUrl sr ++ "/config"+ resp <- liftIO $ Wreq.getWith (wreqOpts sr) url pure $ 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+ let url = srBaseUrl sr ++ "/config/" ++ unpack sbj+ resp <- liftIO $ Wreq.getWith (wreqOpts sr) url pure $ 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+ let url = srBaseUrl sr ++ "/subjects"+ resp <- liftIO $ Wreq.getWith (wreqOpts sr) url pure $ 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 .~ srAuth sr+ in Wreq.defaults & acceptHeader & authHeader -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 <- withExceptT wrapError . ExceptT . try $ 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 (Subject sbj) schema = runExceptT $ do+ let+ baseUrl = srBaseUrl sr+ schemaUrl = baseUrl ++ "/subjects/" ++ unpack sbj ++ "/versions"+ resp <- withExceptT wrapError . ExceptT . try $ 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 +211,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@@ -247,3 +266,16 @@ "FORWARD" -> return $ ForwardCompatibility "BACKWARD" -> return $ BackwardCompatibility _ -> typeMismatch "Compatibility" compatibility+++++++++++++