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: 6.0.0 +version: 6.0.1 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 == 9.2.2, GHC == 9.0.2, GHC == 8.10.7, GHC == 8.8.4, GHC == 8.6.5 +tested-with: GHC == 9.10.1, GHC == 9.8.2, GHC == 9.6.6 extra-source-files: README.md source-repository head @@ -44,7 +44,7 @@ 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 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 } 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 @@ -18,8 +18,8 @@ ) where import Control.Arrow (first) -import Control.Exception (throwIO) -import Control.Exception.Safe (try) +import Control.Exception (throwIO, SomeException (SomeException)) +import Control.Exception.Safe (try, MonadCatch) import Control.Lens (view, (&), (.~), (^.)) import Control.Monad (void) import Control.Monad.IO.Class (MonadIO, liftIO) @@ -43,7 +43,8 @@ import Data.Word (Word32) import GHC.Exception (SomeException, displayException, fromException) import GHC.Generics (Generic) -import Network.HTTP.Client (HttpException (..), HttpExceptionContent (..), Manager, defaultManagerSettings, newManager) +import Network.HTTP.Client (HttpException (..), HttpExceptionContent (..), Manager, defaultManagerSettings, newManager, responseStatus) +import Network.HTTP.Types.Status (notFound404) import qualified Network.Wreq as Wreq newtype SchemaId = SchemaId { unSchemaId :: Int32} deriving (Eq, Ord, Show, Hashable) @@ -71,6 +72,8 @@ data SchemaRegistryError = SchemaRegistryConnectError String | SchemaRegistryLoadError SchemaId | SchemaRegistrySchemaNotFound SchemaId + | SchemaRegistrySubjectNotFound Subject + | SchemaRegistryUrlNotFound String | SchemaRegistrySendError String deriving (Show, Eq) @@ -97,17 +100,21 @@ 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 sr) url - let wrapped = 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 $> 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) @@ -121,6 +128,7 @@ Success a -> Right a Error e -> Left e + sendSchema :: MonadIO m => SchemaRegistry -> Subject -> Schema -> m (Either SchemaRegistryError SchemaId) sendSchema sr subj sc = do sid <- cachedId sr subj schemaName @@ -135,17 +143,20 @@ schemaName = fullTypeName sc getVersions :: MonadIO m => SchemaRegistry -> Subject -> m (Either SchemaRegistryError [Version]) -getVersions sr (Subject sbj) = do +getVersions sr subj@(Subject sbj) = liftIO . runExceptT $ do 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) + 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 sr) url (toJSON $ RegisteredSchema schema) - let wrapped = bimap wrapError (view Wreq.responseBody) (Wreq.asValue resp) - either (return . Left) getCompatibility wrapped + 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 @@ -161,20 +172,22 @@ getGlobalConfig :: MonadIO m => SchemaRegistry -> m (Either SchemaRegistryError Compatibility) getGlobalConfig sr = do let url = srBaseUrl sr ++ "/config" - resp <- liftIO $ Wreq.getWith (wreqOpts sr) url - pure $ bimap wrapError (view Wreq.responseBody) (Wreq.asJSON resp) + 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 +getSubjectConfig sr subj@(Subject sbj) = liftIO . runExceptT $ do let url = srBaseUrl sr ++ "/config/" ++ unpack sbj - resp <- liftIO $ Wreq.getWith (wreqOpts sr) url - pure $ bimap wrapError (view Wreq.responseBody) (Wreq.asJSON resp) + 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 +getSubjects sr = liftIO . runExceptT $ do let url = srBaseUrl sr ++ "/subjects" - resp <- liftIO $ Wreq.getWith (wreqOpts sr) url - pure $ bimap wrapError (fmap Subject . view Wreq.responseBody) (Wreq.asJSON resp) + resp <- tryWith wrapError $ Wreq.getWith (wreqOpts sr) url + except $ bimap wrapError (fmap Subject . view Wreq.responseBody) (Wreq.asJSON resp) ------------------ PRIVATE: HELPERS -------------------------------------------- @@ -191,15 +204,15 @@ let baseUrl = srBaseUrl sr schemaUrl = baseUrl ++ "/schemas/ids/" ++ show i - resp <- withExceptT wrapError . ExceptT . try $ Wreq.getWith (wreqOpts sr) schemaUrl + resp <- tryWith (wrapErrorWithSchemaId sid) $ Wreq.getWith (wreqOpts sr) schemaUrl except $ bimap (const (SchemaRegistryLoadError sid)) (view Wreq.responseBody) (Wreq.asJSON resp) putSchema :: SchemaRegistry -> Subject -> RegisteredSchema -> IO (Either SchemaRegistryError SchemaId) -putSchema sr (Subject sbj) schema = runExceptT $ do +putSchema sr subj@(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) + 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 @@ -220,6 +233,23 @@ wrapError someErr = case fromException someErr of 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