diff --git a/example/Main.hs b/example/Main.hs --- a/example/Main.hs +++ b/example/Main.hs @@ -4,9 +4,6 @@ import Control.Monad.Trans.Except import qualified Data.Aeson as J -import qualified Data.Avro as A -import Data.Avro.Schema as S -import qualified Data.Avro.Types as AT import Data.Monoid import Data.Int @@ -27,9 +24,9 @@ roundtrip :: SchemaRegistry -> ExceptT AppError IO TestMessage roundtrip sr = do - enc <- withExceptT EncError (encode exampleMessage) - dec <- withExceptT DecError (decode enc) + enc <- withExceptT EncError (encode' exampleMessage) + dec <- withExceptT DecError (decode' enc) return dec where - encode msg = ExceptT $ encodeWithSchema sr (Subject "example-subject") exampleMessage - decode msg = ExceptT $ decodeWithSchema sr msg + encode' msg = ExceptT $ encode sr (Subject "example-subject") exampleMessage + decode' msg = ExceptT $ decode sr msg diff --git a/example/Message.hs b/example/Message.hs --- a/example/Message.hs +++ b/example/Message.hs @@ -1,42 +1,26 @@ +{-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE QuasiQuotes #-} +{-# LANGUAGE TemplateHaskell #-} module Message ( TestMessage(..) +, schema'TestMessage ) where --- -import Data.Avro -import Data.Avro.Schema -import qualified Data.Avro.Types as AT -import Data.Int -import Data.Text - -data TestMessage = TestMessage Int64 Text Bool Int64 deriving (Show, Eq, Ord) - -testMessageSchema = - let fld nm = Field nm [] Nothing Nothing - in Record (TN "TestMessage" ["hw", "kafka", "avro", "test"]) [] Nothing Nothing - [ fld "id" Long Nothing - , fld "name" String Nothing - , fld "is_active" Boolean Nothing - , fld "timestamp" Long Nothing - ] - -instance HasAvroSchema TestMessage where - schema = pure testMessageSchema +import Data.Avro +import Data.Avro.Deriving -instance FromAvro TestMessage where - fromAvro (AT.Record _ r) = - TestMessage <$> r .: "id" - <*> r .: "name" - <*> r .: "is_active" - <*> r .: "timestamp" - fromAvro v = badValue v "TestMessage" +deriveAvroFromByteString [r| +{ + "type": "record", + "name": "TestMessage", + "namespace": "hw.kafka.avro.test", + "fields": [ + { "name": "id", "type": "long" }, + { "name": "name", "type": "string" }, + { "name": "is_active", "type": "boolean" }, + { "name": "timestamp", "type": "long" } + ] +} +|] -instance ToAvro TestMessage where - toAvro (TestMessage i s d t) = - record testMessageSchema - [ "id" .= i - , "name" .= s - , "is_active" .= d - , "timestamp" .= t - ] diff --git a/hw-kafka-avro.cabal b/hw-kafka-avro.cabal --- a/hw-kafka-avro.cabal +++ b/hw-kafka-avro.cabal @@ -1,20 +1,20 @@ -cabal-version: 1.12 +cabal-version: 2.4 -name: hw-kafka-avro -version: 4.0.1 -synopsis: Avro support for Kafka infrastructure -description: Avro support for Kafka infrastructure. -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 -extra-source-files: - README.md +name: hw-kafka-avro +version: 6.1.2 +synopsis: Avro support for Kafka infrastructure +description: Avro support for Kafka infrastructure. +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: BSD-3-Clause +license-file: LICENSE +build-type: Simple +tested-with: GHC == 9.10.1, GHC == 9.8.2, GHC == 9.6.6 +extra-source-files: README.md source-repository head type: git @@ -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.2 } +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.4 - , 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 - , 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 - , 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.hs b/src/Kafka/Avro.hs --- a/src/Kafka/Avro.hs +++ b/src/Kafka/Avro.hs @@ -26,7 +26,7 @@ -> Subject -> ByteString -> m (Either SchemaRegistryError (Maybe SchemaId)) -propagateSchema sr subj bs = do +propagateSchema sr subj bs = case extractSchemaId bs of Nothing -> return $ Right Nothing Just (sid, _) -> do diff --git a/src/Kafka/Avro/Decode.hs b/src/Kafka/Avro/Decode.hs --- a/src/Kafka/Avro/Decode.hs +++ b/src/Kafka/Avro/Decode.hs @@ -1,53 +1,57 @@ {-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TypeApplications #-} module Kafka.Avro.Decode ( DecodeError(..) -, decodeWithSchema, extractSchemaId +, decode +, decodeWithSchema +, extractSchemaId ) where -import Control.Arrow (left) -import Control.Monad.IO.Class (MonadIO) -import Data.Avro as A (FromAvro, HasAvroSchema (..), Result (..), fromAvro) -import qualified Data.Avro as A (decodeWithSchema) -import qualified Data.Avro.Decode as A (decodeAvro) -import qualified Data.Avro.Deconflict as A (deconflict) -import Data.Avro.Schema (Schema) -import Data.Bits (shiftL) -import Data.ByteString.Lazy (ByteString) -import qualified Data.ByteString.Lazy as BL hiding (zipWith) +import Control.Arrow (left) +import Control.Monad.IO.Class (MonadIO) +import Control.Monad.Trans.Except +import Data.Avro (FromAvro, HasAvroSchema (..), Schema, decodeValueWithSchema, deconflict) +import Data.Bits (shiftL) +import Data.ByteString.Lazy (ByteString) +import qualified Data.ByteString.Lazy as BL hiding (zipWith) import Data.Int -import Data.Tagged (Tagged, untag) +import Data.Tagged (untag) import Kafka.Avro.SchemaRegistry data DecodeError = DecodeRegistryError SchemaRegistryError | BadPayloadNoSchemaId | DecodeError Schema String + | IncompatibleSchema Schema String deriving (Show, Eq) -- | Decodes a provided Avro-encoded value. -- The serialised value is expected to be in a "confluent-compatible" format -- where the "real" value bytestring is prepended with extra 5 bytes: -- a "magic" byte and 4 bytes representing the schema ID. -decodeWithSchema :: (MonadIO m, FromAvro a) - => SchemaRegistry - -> ByteString - -> m (Either DecodeError a) -decodeWithSchema sr bs = +decode :: forall a m. (MonadIO m, HasAvroSchema a, FromAvro a) + => SchemaRegistry + -> ByteString + -> m (Either DecodeError a) +decode sr = decodeWithSchema sr (untag @a schema) +{-# INLINE decode #-} + +decodeWithSchema :: forall a m. (MonadIO m, FromAvro a) + => SchemaRegistry + -> Schema + -> ByteString + -> m (Either DecodeError a) +decodeWithSchema sr readerSchema bs = case schemaData of Left err -> return $ Left err - Right (sid, payload) -> do - res <- left DecodeRegistryError <$> loadSchema sr sid - return $ res >>= flip decodeWithDeconflict payload + Right (sid, payload) -> runExceptT $ do + writerSchema <- withError DecodeRegistryError (loadSchema sr sid) + readSchema <- withPureError (IncompatibleSchema writerSchema) $ deconflict writerSchema readerSchema + withPureError (DecodeError writerSchema) (decodeValueWithSchema readSchema payload) where schemaData = maybe (Left BadPayloadNoSchemaId) Right (extractSchemaId bs) - -decodeWithDeconflict :: forall a. (FromAvro a) => Schema -> ByteString -> Either DecodeError a -decodeWithDeconflict writerSchema bs = - let readerSchema = untag (schema :: Tagged a Schema) - in left (DecodeError readerSchema) $ do - raw <- A.decodeAvro writerSchema bs - val <- A.deconflict writerSchema readerSchema raw - resultToEither readerSchema (fromAvro val) + withError f = withExceptT f . ExceptT + withPureError f = withError f . pure extractSchemaId :: ByteString -> Maybe (SchemaId, ByteString) extractSchemaId bs = do @@ -59,10 +63,4 @@ let ints = fromIntegral <$> [w4, w3, w2, w1] :: [Int32] let int = sum $ zipWith shiftL ints [0, 8, 16, 24] return (SchemaId int, b4) - -resultToEither :: Schema -> A.Result a -> Either String a -resultToEither sc res = case res of - Success a -> Right a - Error msg -> Left msg - 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,14 +1,23 @@ -{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# OPTIONS_GHC -Wno-unrecognised-pragmas #-} +{-# HLINT ignore "Use newtype instead of data" #-} module Kafka.Avro.Encode -( encodeKey, encodeValue -, keySubject, valueSubject +( EncodeError(..) +, encodeKey +, encodeValue +, encode + +, encodeKeyWithSchema +, encodeValueWithSchema , encodeWithSchema -, EncodeError(..) + +, keySubject, valueSubject ) where import Control.Monad.IO.Class (MonadIO) -import Data.Avro as A (ToAvro, encode, schemaOf) -import Data.Avro.Schema (Schema) +import Data.Avro (HasAvroSchema, Schema, ToAvro, schemaOf) +import qualified Data.Avro as A import qualified Data.Binary as B import Data.Bits (shiftL) import Data.ByteString.Lazy (ByteString) @@ -21,42 +30,70 @@ keySubject :: Subject -> Subject keySubject (Subject subj) = Subject (subj <> "-key") +{-# INLINE keySubject #-} valueSubject :: Subject -> Subject valueSubject (Subject subj) = Subject (subj <> "-value") +{-# INLINE valueSubject #-} --- | Encodes a provided value as a message key. --- --- Registers the schema in SchemaRegistry with "-key" subject. -encodeKey :: (MonadIO m, ToAvro a) - => SchemaRegistry - -> Subject - -> a - -> m (Either EncodeError ByteString) -encodeKey sr subj = encodeWithSchema sr (keySubject subj) +-- | Encodes a provided value as a message key with "-key" subject. +encodeKey :: (MonadIO m, HasAvroSchema a, ToAvro a) + => SchemaRegistry + -> Subject + -> a + -> m (Either EncodeError ByteString) +encodeKey sr subj = encode sr (keySubject subj) +{-# INLINE encodeKey #-} --- | Encodes a provided value as a message value. --- --- Registers the schema in SchemaRegistry with "-value" subject. -encodeValue :: (MonadIO m, ToAvro a) - => SchemaRegistry - -> Subject - -> a - -> m (Either EncodeError ByteString) -encodeValue sr subj = encodeWithSchema sr (valueSubject subj) +-- | Encodes a provided value as a message key with "-key" subject. +encodeKeyWithSchema :: (MonadIO m, ToAvro a) + => SchemaRegistry + -> Subject + -> Schema + -> a + -> m (Either EncodeError ByteString) +encodeKeyWithSchema sr subj = encodeWithSchema sr (keySubject subj) +{-# INLINE encodeKeyWithSchema #-} +-- | Encodes a provided value as a message value with "-value" subject. +encodeValue :: (MonadIO m, HasAvroSchema a, ToAvro a) + => SchemaRegistry + -> Subject + -> a + -> m (Either EncodeError ByteString) +encodeValue sr subj = encode sr (valueSubject subj) +{-# INLINE encodeValue #-} + +-- | Encodes a provided value as a message value with "-value" subject. +encodeValueWithSchema :: (MonadIO m, ToAvro a) + => SchemaRegistry + -> Subject + -> Schema + -> a + -> m (Either EncodeError ByteString) +encodeValueWithSchema sr subj = encodeWithSchema sr (valueSubject subj) +{-# INLINE encodeValueWithSchema #-} + +encode :: (MonadIO m, HasAvroSchema a, ToAvro a) + => SchemaRegistry + -> Subject + -> a + -> m (Either EncodeError ByteString) +encode sr subj a = encodeWithSchema sr subj (schemaOf a) a +{-# INLINE encode #-} + -- | Encodes a provided value into Avro --- and registers value's schema in SchemaRegistry. -encodeWithSchema :: (MonadIO m, ToAvro a) - => SchemaRegistry - -> Subject - -> a - -> m (Either EncodeError ByteString) -encodeWithSchema sr subj a = do - mbSid <- sendSchema sr subj (schemaOf a) +encodeWithSchema :: forall a m. (MonadIO m, ToAvro a) + => SchemaRegistry + -> Subject + -> Schema + -> a + -> m (Either EncodeError ByteString) +encodeWithSchema sr subj sch a = do + mbSid <- sendSchema sr subj sch case mbSid of Left err -> return . Left . EncodeRegistryError $ err - Right sid -> return . Right $ appendSchemaId sid (encode a) + Right sid -> return . Right $ appendSchemaId sid (A.encodeValueWithSchema sch a) appendSchemaId :: SchemaId -> ByteString -> ByteString 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,44 +1,61 @@ {-# LANGUAGE CPP #-} {-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} -{-# LANGUAGE LambdaCase #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TupleSections #-} module Kafka.Avro.SchemaRegistry ( schemaRegistry, loadSchema, sendSchema +, schemaRegistry_ +, schemaRegistryWithHeaders , loadSubjectSchema , getGlobalConfig, getSubjectConfig , getVersions, isCompatible , getSubjects +, defaultSchemaRegistryConfig +, cfgAuth +, cfgHeaders +, cfgAutoRegisterSchemas , SchemaId(..), Subject(..) +, SchemaRegistryConfig , 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 Control.Arrow (first) +import Control.Exception (SomeException (SomeException), throwIO) +import Control.Exception.Safe (MonadCatch, try) +import Control.Lens (view, (%~), (&), (.~), (^.)) +import Control.Monad (void) +import Control.Monad.Except (liftEither) +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 Data.Avro.Schema (Schema, Type (..), typeName) -import Data.Bifunctor (bimap) -import Data.Cache as C -import Data.Hashable (Hashable) -import qualified Data.HashMap.Lazy as HM -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) -import qualified Network.Wreq as Wreq +import qualified Data.Aeson.Key as A +import qualified Data.Aeson.KeyMap as KM +import Data.Aeson.Types (typeMismatch) +import Data.Avro.Schema.Schema (Schema (..), typeName) +import Data.Bifunctor (bimap) +import Data.Cache as C +import Data.Foldable (traverse_) +import Data.Functor (($>)) +import Data.Hashable (Hashable) +import qualified Data.HashMap.Lazy as HM +import Data.Int (Int32) +import Data.List (find) +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, responseStatus) +import Network.HTTP.Types.Header (Header) +import Network.HTTP.Types.Status (notFound404) +import qualified Network.Wreq as Wreq newtype SchemaId = SchemaId { unSchemaId :: Int32} deriving (Eq, Ord, Show, Hashable) newtype SchemaName = SchemaName Text deriving (Eq, Ord, IsString, Show, Hashable) @@ -55,47 +72,95 @@ | BackwardCompatibility deriving (Eq, Show, Ord) +data SchemaRegistryConfig = SchemaRegistryConfig + { cAuth :: Maybe Wreq.Auth + , cExtraHeaders :: [Header] + , cAutoRegisterSchemas :: Bool + } + data SchemaRegistry = SchemaRegistry { srCache :: Cache SchemaId Schema , srReverseCache :: Cache (Subject, SchemaName) SchemaId , srBaseUrl :: String + , srConfig :: SchemaRegistryConfig } data SchemaRegistryError = SchemaRegistryConnectError String | SchemaRegistryLoadError SchemaId | SchemaRegistrySchemaNotFound SchemaId + | SchemaRegistrySubjectNotFound Subject + | SchemaRegistryNoCompatibleSchemaFound Schema + | SchemaRegistryUrlNotFound String | SchemaRegistrySendError String + | SchemaRegistryCacheError deriving (Show, Eq) +defaultSchemaRegistryConfig :: SchemaRegistryConfig +defaultSchemaRegistryConfig = SchemaRegistryConfig + { cAuth = Nothing + , cExtraHeaders = [] + , cAutoRegisterSchemas = True + } + 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 = schemaRegistryWithHeaders auth [] + +schemaRegistryWithHeaders :: MonadIO m => Maybe Wreq.Auth -> [Header] -> String -> m SchemaRegistry +schemaRegistryWithHeaders auth headers url + = schemaRegistryWithConfig url $ cfgAuth auth $ cfgHeaders headers defaultSchemaRegistryConfig + +schemaRegistryWithConfig :: MonadIO m => String -> SchemaRegistryConfig -> m SchemaRegistry +schemaRegistryWithConfig url config = liftIO $ + SchemaRegistry + <$> newCache Nothing + <*> newCache Nothing + <*> pure url + <*> pure config + +-- | Add authentication options +cfgAuth :: Maybe Wreq.Auth -> SchemaRegistryConfig -> SchemaRegistryConfig +cfgAuth auth config = config { cAuth = auth } + +-- | Add extra headers +cfgHeaders :: [Header] -> SchemaRegistryConfig -> SchemaRegistryConfig +cfgHeaders headers config = config { cExtraHeaders = headers } + +-- | Set whether to auto-publish schemas +-- If set to 'False', encoding will fail if there is no compatible schema +-- in the schema registy. +-- This is equivalent to the confluent 'auto.register.schemas' option. +cfgAutoRegisterSchemas :: Bool -> SchemaRegistryConfig -> SchemaRegistryConfig +cfgAutoRegisterSchemas autoRegisterSchemas config = config { cAutoRegisterSchemas = autoRegisterSchemas } + 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 + respE <- liftIO . try $ Wreq.getWith (wreqOpts sr) url + case respE of + Left exc -> pure . Left $ wrapErrorWithUrl url exc + Right resp -> do - schema <- getData "schema" wrapped - schemaId <- getData "id" wrapped + 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) + case (,) <$> schema <*> schemaId of + Left err -> return $ Left err + 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) @@ -109,37 +174,68 @@ Success a -> Right a Error e -> Left e + +-- | Get the schema ID. +-- If the 'SchemaRegistry' is configured to auto-register schemas, +-- this posts the schema to the schema registry server. +-- Otherwise, this searches for a compatible schema and returns a 'SchemaRegistryNoCompatibleSchemaFound' +-- if none is found. sendSchema :: MonadIO m => SchemaRegistry -> Subject -> Schema -> m (Either SchemaRegistryError SchemaId) sendSchema sr subj sc = do + let schemaName = fullTypeName sc sid <- cachedId sr subj schemaName 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 - pure res - where - schemaName = fullTypeName sc + Nothing -> if cAutoRegisterSchemas (srConfig sr) + then registerSchema sr subj sc + else getCompatibleSchema sr subj sc +registerSchema :: MonadIO m => SchemaRegistry -> Subject -> Schema -> m (Either SchemaRegistryError SchemaId) +registerSchema sr subj sc = do + let schemaName = fullTypeName sc + res <- liftIO $ putSchema sr subj (RegisteredSchema sc) + traverse_ (cacheId sr subj schemaName) res + traverse_ (\sid' -> cacheSchema sr sid' sc) res + pure res + +getCompatibleSchema :: MonadIO m => SchemaRegistry -> Subject -> Schema -> m (Either SchemaRegistryError SchemaId) +getCompatibleSchema sr subj sc = liftIO . runExceptT $ do + let schemaName = fullTypeName sc + versions <- liftEither =<< getVersions sr subj + compatibilites <- liftEither . sequenceA + =<< traverse (\ver -> fmap (,ver) <$> isCompatible sr subj ver sc) versions + let mCompatibleVersion = snd <$> find fst compatibilites + compatibleVersion <- liftEither + $ case mCompatibleVersion of + Just version -> Right version + Nothing -> Left $ SchemaRegistryNoCompatibleSchemaFound sc + _ <- liftEither =<< loadSubjectSchema sr subj compatibleVersion -- caches the schema ID + mSid <- cachedId sr subj schemaName + liftEither $ case mSid of + Just sid' -> pure sid' + Nothing -> Left SchemaRegistryCacheError + 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) +getVersions sr subj@(Subject sbj) = liftIO . runExceptT $ do + let url = srBaseUrl sr ++ "/subjects/" ++ unpack sbj ++ "/versions" + resp <- tryWith (wrapErrorWithSubject subj) $ Wreq.getWith (wreqOpts sr) url + except $ 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 + let url = srBaseUrl sr ++ "/compatibility/subjects/" ++ unpack sbj ++ "/versions/" ++ show version + respE <- liftIO . try $ Wreq.postWith (wreqOpts sr) url (toJSON $ RegisteredSchema schema) + case respE of + Left exc -> pure . Left $ wrapErrorWithUrl url exc + Right resp -> do + 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 +244,50 @@ 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) + let url = srBaseUrl sr ++ "/config" + respE <- liftIO . try $ Wreq.getWith (wreqOpts sr) url + pure $ case respE of + Left exc -> Left $ wrapError exc + Right resp -> 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) +getSubjectConfig sr subj@(Subject sbj) = liftIO . runExceptT $ do + let url = srBaseUrl sr ++ "/config/" ++ unpack sbj + resp <- tryWith (wrapErrorWithSubject subj) $ Wreq.getWith (wreqOpts sr) url + except $ 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) +getSubjects sr = liftIO . runExceptT $ do + let url = srBaseUrl sr ++ "/subjects" + resp <- tryWith wrapError $ Wreq.getWith (wreqOpts sr) url + except $ 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 .~ cAuth (srConfig sr) + extraHeaders = Wreq.headers %~ (++ cExtraHeaders (srConfig sr)) + in Wreq.defaults & acceptHeader & authHeader & extraHeaders -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 <- tryWith (wrapErrorWithSchemaId sid) $ 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 subj@(Subject sbj) schema = runExceptT $ do + let + baseUrl = srBaseUrl sr + schemaUrl = baseUrl ++ "/subjects/" ++ unpack sbj ++ "/versions" + resp <- tryWith (wrapErrorWithSubject subj) $ 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 +298,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 @@ -202,6 +308,23 @@ Nothing -> SchemaRegistrySendError (displayException someErr) Just httpErr -> fromHttpError httpErr (\_ -> SchemaRegistrySendError (displayException someErr)) +wrapErrorWithSchemaId :: SchemaId -> SomeException -> SchemaRegistryError +wrapErrorWithSchemaId = wrapErrorWith SchemaRegistrySchemaNotFound + +wrapErrorWithSubject :: Subject -> SomeException -> SchemaRegistryError +wrapErrorWithSubject = wrapErrorWith SchemaRegistrySubjectNotFound + +wrapErrorWithUrl :: String -> SomeException -> SchemaRegistryError +wrapErrorWithUrl = wrapErrorWith SchemaRegistryUrlNotFound + +wrapErrorWith :: (a -> SchemaRegistryError) -> a -> SomeException -> SchemaRegistryError +wrapErrorWith mkError x exception = case fromException exception of + Just (HttpExceptionRequest _ (StatusCodeException response _)) | responseStatus response == notFound404 -> mkError x + _ -> wrapError exception + +tryWith :: MonadCatch m => (SomeException -> e) -> m a -> ExceptT e m a +tryWith wrapException = withExceptT wrapException . ExceptT . try + --------------------------------------------------------------------- fullTypeName :: Schema -> SchemaName fullTypeName r = SchemaName $ typeName r @@ -247,3 +370,16 @@ "FORWARD" -> return $ ForwardCompatibility "BACKWARD" -> return $ BackwardCompatibility _ -> typeMismatch "Compatibility" compatibility + + + + + + + + + + + + +