diff --git a/hw-kafka-avro.cabal b/hw-kafka-avro.cabal --- a/hw-kafka-avro.cabal +++ b/hw-kafka-avro.cabal @@ -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 diff --git a/src/Kafka/Avro/Encode.hs b/src/Kafka/Avro/Encode.hs --- a/src/Kafka/Avro/Encode.hs +++ b/src/Kafka/Avro/Encode.hs @@ -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 diff --git a/src/Kafka/Avro/SchemaRegistry.hs b/src/Kafka/Avro/SchemaRegistry.hs --- a/src/Kafka/Avro/SchemaRegistry.hs +++ b/src/Kafka/Avro/SchemaRegistry.hs @@ -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 + + + + + + + + + + + + +