diff --git a/hw-kafka-avro.cabal b/hw-kafka-avro.cabal --- a/hw-kafka-avro.cabal +++ b/hw-kafka-avro.cabal @@ -1,110 +1,124 @@ -name: hw-kafka-avro -version: 1.4.0 -synopsis: Avro support for Kafka infrastructure -description: Please see README.md -homepage: https://github.com/haskell-works/hw-kafka-avro#readme -license: BSD3 -license-file: LICENSE -author: Alexey Raga -maintainer: alexey.raga@gmail.com -copyright: Alexey Raga -category: Services -build-type: Simple -extra-source-files: README.md -cabal-version: >=1.10 +-- This file has been generated from package.yaml by hpack version 0.28.2. +-- +-- see: https://github.com/sol/hpack +-- +-- hash: b0f4e354a38dab39408fe0e2d96c0fd0801511d2eeaa57b6757c21b884f7315f +name: hw-kafka-avro +version: 2.0.0 +synopsis: Avro support for Kafka infrastructure +description: Please see README.md +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 +cabal-version: >= 1.10 +extra-source-files: + README.md + +source-repository head + type: git + location: https://github.com/haskell-works/hw-kafka-avro + flag examples - default: False - manual: True description: Also compile examples + manual: True + default: False library - hs-source-dirs: src - exposed-modules: Kafka.Avro, - Kafka.Avro.Decode, - Kafka.Avro.Encode, - Kafka.Avro.SchemaRegistry - - build-depends: aeson, - avro >= 0.2, - base >= 4.7 && < 5, - binary, - bytestring, - cache, - containers, - errors, - hashable, - http-client, - http-types, - mtl, - pure-zlib >= 0.6, - semigroups, - servant, - servant-client, - tagged, - text, - transformers, - unordered-containers - default-language: Haskell2010 + exposed-modules: + Kafka.Avro + Kafka.Avro.Decode + Kafka.Avro.Encode + Kafka.Avro.SchemaRegistry + other-modules: + Paths_hw_kafka_avro + hs-source-dirs: + src + build-depends: + aeson + , avro >=0.2 + , 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 - hs-source-dirs: example - main-is: Main.hs - other-modules: Message - ghc-options: -threaded -rtsopts -with-rtsopts=-N - build-depends: aeson - , avro - , base - , binary - , bytestring - , cache - , containers - , errors - , hashable - , hw-kafka-avro - , http-client - , mtl - , pure-zlib >= 0.6 - , semigroups - , servant - , servant-client - , text - , transformers - , unordered-containers - default-language: Haskell2010 - if flag(examples) - buildable: True - else + main-is: Main.hs + other-modules: + Message + Paths_hw_kafka_avro + 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 + if !(flag(examples)) buildable: False + default-language: Haskell2010 test-suite kafka-avro-test - type: exitcode-stdio-1.0 - hs-source-dirs: test - main-is: Spec.hs - build-depends: QuickCheck - , aeson - , avro - , base - , binary - , bytestring - , cache - , containers - , errors - , hashable - , hw-kafka-avro - , hspec - , http-client - , mtl - , pure-zlib >= 0.6 - , semigroups - , servant - , servant-client - , text - , transformers - , unordered-containers - ghc-options: -threaded -rtsopts -with-rtsopts=-N - default-language: Haskell2010 - -source-repository head - type: git - location: https://github.com/haskell-works/hw-kafka-avro + type: exitcode-stdio-1.0 + main-is: Spec.hs + other-modules: + Paths_hw_kafka_avro + 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 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,59 +1,51 @@ -{-# LANGUAGE CPP #-} -{-# LANGUAGE DataKinds #-} {-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE LambdaCase #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-} -{-# LANGUAGE TypeOperators #-} module Kafka.Avro.SchemaRegistry ( schemaRegistry, loadSchema, sendSchema , SchemaId(..), Subject(..) , SchemaRegistry, SchemaRegistryError(..) -, RegisteredSchema(..) +, Schema(..) ) where -import Control.Monad.IO.Class -import Control.Monad.Trans.Except (ExceptT (..), runExceptT, withExceptT) +import Control.Arrow (first) +import Control.Lens (view, (&), (.~), (^.)) +import Control.Monad (void) +import Control.Monad.IO.Class (MonadIO, liftIO) import Data.Aeson -import Data.Avro.Schema (Schema, Type (..), typeName) -import Data.Cache as C -import Data.Hashable -import Data.Int -import Data.Proxy -import Data.Text (Text, append, cons) -import qualified Data.Text.Encoding as Text -import qualified Data.Text.Lazy.Encoding as LText -import GHC.Exception -import GHC.Generics (Generic) -import Network.HTTP.Client (Manager, defaultManagerSettings, newManager) -import Network.HTTP.Types -import Servant.API -import Servant.Client +import Data.Avro.Schema (Schema, Type (..), typeName) +import Data.Bifunctor (bimap) +import Data.Cache as C +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 GHC.Exception (SomeException, displayException, fromException) +import GHC.Generics (Generic) +import Network.HTTP.Client (HttpException (..), HttpExceptionContent (..), Manager, defaultManagerSettings, newManager) +import qualified Network.Wreq as Wreq -newtype SchemaId = SchemaId Int32 deriving (Eq, Ord, Show, Hashable) -newtype SchemaName = SchemaName Text deriving (Eq, Ord, Show, Hashable) +newtype SchemaId = SchemaId { unSchemaId :: Int32} deriving (Eq, Ord, Show, Hashable) +newtype SchemaName = SchemaName Text deriving (Eq, Ord, IsString, Show, Hashable) -newtype Subject = Subject Text deriving (Eq, Show, Generic, Hashable) +newtype Subject = Subject { unSubject :: Text} deriving (Eq, Show, IsString, Ord, Generic, Hashable) -newtype RegisteredSchema = RegisteredSchema Schema deriving (Generic, Show) +newtype RegisteredSchema = RegisteredSchema { unRegisteredSchema :: Schema} deriving (Generic, Show) data SchemaRegistry = SchemaRegistry { srCache :: Cache SchemaId Schema , srReverseCache :: Cache (Subject, SchemaName) SchemaId -#if MIN_VERSION_servant(0,9,1) - , srClientEnv :: ClientEnv -#else - , srManager :: Manager - , srBaseUrl :: BaseUrl -#endif + , srBaseUrl :: String } data SchemaRegistryError = SchemaRegistryConnectError String - | SchemaDecodeError SchemaId String | SchemaRegistryLoadError SchemaId | SchemaRegistrySchemaNotFound SchemaId - | SchemaRegistryError SchemaId | SchemaRegistrySendError String deriving (Show, Eq) @@ -62,26 +54,16 @@ SchemaRegistry <$> newCache Nothing <*> newCache Nothing -#if MIN_VERSION_servant(0,9,1) - <*> (ClientEnv <$> newManager defaultManagerSettings <*> parseBaseUrl url) -#else - <*> newManager defaultManagerSettings - <*> parseBaseUrl url -#endif + <*> pure url 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 -> do -#if MIN_VERSION_servant(0,9,1) - res <- loadSchemaFromSR (srClientEnv sr) sid -#else - res <- loadSchemaFromSR (srManager sr) (srBaseUrl sr) sid -#endif - _ <- traverse (cacheSchema sr sid) res - return res + Nothing -> liftIO $ do + res <- getSchemaById (srBaseUrl sr) sid + pure (unRegisteredSchema <$> res) sendSchema :: MonadIO m => SchemaRegistry -> Subject -> Schema -> m (Either SchemaRegistryError SchemaId) sendSchema sr subj sc = do @@ -89,67 +71,56 @@ case sid of Just sid' -> return (Right sid') Nothing -> do -#if MIN_VERSION_servant(0,9,1) - res <- sendSchemaToSR (srClientEnv sr) subj sc -#else - res <- sendSchemaToSR (srManager sr) (srBaseUrl sr) subj sc -#endif - _ <- traverse (cacheId sr subj schemaName) res - _ <- traverse (\sid' -> cacheSchema sr sid' sc) res - return res + 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 ------------------ PRIVATE: HELPERS -------------------------------------------- -type API = "schemas" :> "ids" :> Capture "id" Int32 :> Get '[JSON] RegisteredSchema - :<|> "subjects" :> Capture "subject" Subject :> "versions" :> ReqBody '[JSON] RegisteredSchema :> Post '[JSON] SchemaId -api :: Proxy API -api = Proxy - -#if MIN_VERSION_servant(0,9,1) ---ExceptT ServantError IO SchemaId -getSchemaById :: Int32 -> ClientM RegisteredSchema -putSchema :: Subject -> RegisteredSchema -> ClientM SchemaId -#else -getSchemaById :: Int32 -> Manager -> BaseUrl -> ClientM RegisteredSchema -putSchema :: Subject -> RegisteredSchema -> Manager -> BaseUrl -> ClientM SchemaId -#endif -getSchemaById :<|> putSchema = client api +wreqOpts :: Wreq.Options +wreqOpts = + let accept = ["application/vnd.schemaregistry.v1+json", "application/vnd.schemaregistry+json", "application/json"] + in Wreq.defaults & Wreq.header "Accept" .~ accept -#if MIN_VERSION_servant(0,9,1) -sendSchemaToSR :: MonadIO m => ClientEnv -> Subject -> Schema -> m (Either SchemaRegistryError SchemaId) -sendSchemaToSR env subj s = - runExceptT $ withExceptT toSRError $ runServant env $ putSchema subj (RegisteredSchema s) -#else -sendSchemaToSR :: MonadIO m => Manager -> BaseUrl -> Subject -> Schema -> m (Either SchemaRegistryError SchemaId) -sendSchemaToSR m u subj s = - liftExceptT . withExceptT toSRError $ putSchema subj (RegisteredSchema s) m u -#endif +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) where - toSRError msg = case msg of - ConnectionError ex -> SchemaRegistryConnectError (show ex) - DecodeFailure de _ _ -> SchemaRegistrySendError de - err -> SchemaRegistrySendError (show err) + wrapError :: SomeException -> SchemaRegistryError + wrapError someErr = case fromException someErr of + Nothing -> SchemaRegistryLoadError sid + Just httpErr -> fromHttpError httpErr $ \case + StatusCodeException r _ | r ^. Wreq.responseStatus . Wreq.statusCode == 404 -> SchemaRegistrySchemaNotFound sid + _ -> SchemaRegistryLoadError sid -#if MIN_VERSION_servant(0,9,1) -loadSchemaFromSR :: MonadIO m => ClientEnv -> SchemaId -> m (Either SchemaRegistryError Schema) -loadSchemaFromSR env sid@(SchemaId i) = - runExceptT (withExceptT toSRError $ runServant env $ unwrapResponse <$> getSchemaById i) -#else -loadSchemaFromSR :: MonadIO m => Manager -> BaseUrl -> SchemaId -> m (Either SchemaRegistryError Schema) -loadSchemaFromSR m u sid@(SchemaId i) = - liftExceptT (withExceptT toSRError $ unwrapResponse <$> getSchemaById i m u) -#endif +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) where - unwrapResponse (RegisteredSchema s) = s - toSRError msg = case msg of - ConnectionError ex -> SchemaRegistryConnectError (show ex) - FailureResponse {responseStatus=Status {statusCode=404}} -> SchemaRegistrySchemaNotFound sid - FailureResponse{} -> SchemaRegistryLoadError sid - DecodeFailure de _ _ -> SchemaDecodeError sid de - _ -> SchemaRegistryLoadError sid + 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 + InvalidUrlException fld err' -> SchemaRegistryConnectError (fld ++ ": " ++ err') + HttpExceptionRequest _ (ConnectionFailure err) -> SchemaRegistryConnectError (displayException err) + 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) + HttpExceptionRequest _ err' -> f err' + --------------------------------------------------------------------- fullTypeName :: Schema -> SchemaName fullTypeName r = SchemaName $ case r of @@ -190,13 +161,3 @@ parseJSON (Object v) = SchemaId <$> v .: "id" parseJSON _ = mempty -instance ToHttpApiData Subject where - toUrlPiece (Subject s) = toUrlPiece s - -#if MIN_VERSION_servant(0,9,1) -runServant :: MonadIO m => ClientEnv -> ClientM a -> ExceptT ServantError m a -runServant env cli = ExceptT $ liftIO (runClientM cli env) -#else -liftExceptT :: MonadIO m => ExceptT l IO r -> m (Either l r) -liftExceptT = liftIO . runExceptT -#endif