hw-kafka-avro 2.0.0 → 2.1.0
raw patch · 2 files changed
+79/−9 lines, 2 filesdep ~avro
Dependency ranges changed: avro
Files
- hw-kafka-avro.cabal +3/−3
- src/Kafka/Avro/SchemaRegistry.hs +76/−6
hw-kafka-avro.cabal view
@@ -2,10 +2,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: b0f4e354a38dab39408fe0e2d96c0fd0801511d2eeaa57b6757c21b884f7315f+-- hash: f725a4e610fd945f6b308f38bb7741ff5db49869d28a7058343203118cc735a7 name: hw-kafka-avro-version: 2.0.0+version: 2.1.0 synopsis: Avro support for Kafka infrastructure description: Please see README.md category: Services@@ -42,7 +42,7 @@ src build-depends: aeson- , avro >=0.2+ , avro >=0.3 , base >=4.7 && <5 , binary , bytestring
src/Kafka/Avro/SchemaRegistry.hs view
@@ -1,3 +1,4 @@+{-# LANGUAGE CPP #-} {-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE LambdaCase #-}@@ -6,25 +7,33 @@ module Kafka.Avro.SchemaRegistry ( schemaRegistry, loadSchema, sendSchema+, getGlobalConfig, getSubjectConfig+, getVersions, isCompatible+, getSubjects , SchemaId(..), Subject(..) , 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 Data.Aeson+import Data.Aeson.Types (typeMismatch) import Data.Avro.Schema (Schema, Type (..), typeName) import Data.Bifunctor (bimap) import Data.Cache as C+import qualified Data.HashMap.Lazy as HM import Data.Hashable (Hashable) 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)@@ -37,6 +46,14 @@ newtype RegisteredSchema = RegisteredSchema { unRegisteredSchema :: Schema} deriving (Generic, Show) +newtype Version = Version { unVersion :: Word32 } deriving (Eq, Ord, Show, Hashable)++data Compatibility = NoCompatibility+ | FullCompatibility+ | ForwardCompatibility+ | BackwardCompatibility+ deriving (Eq, Show, Ord)+ data SchemaRegistry = SchemaRegistry { srCache :: Cache SchemaId Schema , srReverseCache :: Cache (Subject, SchemaName) SchemaId@@ -78,6 +95,48 @@ 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+ 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)+ 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 _ = Nothing++ toBool :: Value -> Maybe Bool+ toBool (Bool b) = Just b+ toBool _ = Nothing++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)++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)++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)+ ------------------ PRIVATE: HELPERS -------------------------------------------- wreqOpts :: Wreq.Options@@ -103,11 +162,6 @@ let schemaUrl = baseUrl ++ "/subjects/" ++ unpack sbj ++ "/versions" resp <- Wreq.postWith wreqOpts schemaUrl (toJSON schema) pure $ bimap wrapError (view Wreq.responseBody) (Wreq.asJSON resp)- where- wrapError :: SomeException -> SchemaRegistryError- wrapError someErr = case fromException someErr of- Nothing -> SchemaRegistrySendError (displayException someErr)- Just httpErr -> fromHttpError httpErr (\_ -> SchemaRegistrySendError (displayException someErr)) fromHttpError :: HttpException -> (HttpExceptionContent -> SchemaRegistryError) -> SchemaRegistryError fromHttpError err f = case err of@@ -116,11 +170,18 @@ HttpExceptionRequest _ ConnectionTimeout -> SchemaRegistryConnectError (displayException err) HttpExceptionRequest _ ProxyConnectException{} -> SchemaRegistryConnectError (displayException err) HttpExceptionRequest _ ConnectionClosed -> SchemaRegistryConnectError (displayException err)- HttpExceptionRequest _ (InvalidProxySettings _) -> 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+wrapError someErr = case fromException someErr of+ Nothing -> SchemaRegistrySendError (displayException someErr)+ Just httpErr -> fromHttpError httpErr (\_ -> SchemaRegistrySendError (displayException someErr))+ --------------------------------------------------------------------- fullTypeName :: Schema -> SchemaName fullTypeName r = SchemaName $ case r of@@ -161,3 +222,12 @@ parseJSON (Object v) = SchemaId <$> v .: "id" parseJSON _ = mempty +instance FromJSON Compatibility where+ parseJSON = withObject "Compatibility" $ \v -> do+ compatibility <- v .: "compatibilityLevel"+ case compatibility of+ "NONE" -> return $ NoCompatibility+ "FULL" -> return $ FullCompatibility+ "FORWARD" -> return $ ForwardCompatibility+ "BACKWARD" -> return $ BackwardCompatibility+ _ -> typeMismatch "Compatibility" compatibility