hw-kafka-avro 6.0.0 → 6.0.1

raw patch · 2 files changed

+61/−31 lines, 2 filesdep ~textPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependency ranges changed: text

API changes (from Hackage documentation)

Files

• hw-kafka-avro.cabal +3/−3
• src/Kafka/Avro/SchemaRegistry.hs +58/−28

@@ -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                   }

@@ -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