hw-kafka-avro 1.4.0 → 2.0.0
raw patch · 2 files changed
+182/−207 lines, 2 filesdep +lensdep +wreqdep −servantdep −servant-client
Dependencies added: lens, wreq
Dependencies removed: servant, servant-client
Files
- hw-kafka-avro.cabal +114/−100
- src/Kafka/Avro/SchemaRegistry.hs +68/−107
hw-kafka-avro.cabal view
@@ -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
src/Kafka/Avro/SchemaRegistry.hs view
@@ -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