hw-kafka-avro 2.1.0 → 4.0.0
raw patch · 3 files changed
+38/−12 lines, 3 filesdep ~avro
Dependency ranges changed: avro
Files
- example/Message.hs +1/−1
- hw-kafka-avro.cabal +3/−3
- src/Kafka/Avro/SchemaRegistry.hs +34/−8
example/Message.hs view
@@ -14,7 +14,7 @@ testMessageSchema = let fld nm = Field nm [] Nothing Nothing- in Record "TestMessage" (Just "hw.kafka.avro.test") [] 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
hw-kafka-avro.cabal view
@@ -2,10 +2,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: f725a4e610fd945f6b308f38bb7741ff5db49869d28a7058343203118cc735a7+-- hash: 30f52e81862851b4a683aac241a304455f46ec73f8d755304cd003bc7a6fb41b name: hw-kafka-avro-version: 2.1.0+version: 4.0.0 synopsis: Avro support for Kafka infrastructure description: Please see README.md category: Services@@ -42,7 +42,7 @@ src build-depends: aeson- , avro >=0.3+ , avro >=0.4 , base >=4.7 && <5 , binary , bytestring
src/Kafka/Avro/SchemaRegistry.hs view
@@ -7,6 +7,7 @@ module Kafka.Avro.SchemaRegistry ( schemaRegistry, loadSchema, sendSchema+, loadSubjectSchema , getGlobalConfig, getSubjectConfig , getVersions, isCompatible , getSubjects@@ -26,17 +27,20 @@ import Data.Avro.Schema (Schema, Type (..), typeName) import Data.Bifunctor (bimap) import Data.Cache as C-import qualified Data.HashMap.Lazy as HM 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.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) import qualified Network.Wreq as Wreq newtype SchemaId = SchemaId { unSchemaId :: Int32} deriving (Eq, Ord, Show, Hashable)@@ -82,6 +86,32 @@ res <- getSchemaById (srBaseUrl sr) sid pure (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)++ 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)+ where+ getData :: (MonadIO m, FromJSON a) => String -> Either e Value -> m (Either e a)+ getData key = either (pure . Left) (viewData key)++ viewData :: (MonadIO m, FromJSON a) => String -> Value -> m (Either e a)+ viewData key value = liftIO $ either (throwIO . Wreq.JSONError)+ (return . return)+ (toData value)++ toData :: FromJSON a => Value -> Either String a+ toData value = case fromJSON value of+ 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@@ -184,11 +214,7 @@ --------------------------------------------------------------------- fullTypeName :: Schema -> SchemaName-fullTypeName r = SchemaName $ case r of- Record{} -> maybe (typeName r)- (\ns -> ns `append` ('.' `cons` typeName r))- (namespace r)- _ -> typeName r+fullTypeName r = SchemaName $ typeName r cachedSchema :: MonadIO m => SchemaRegistry -> SchemaId -> m (Maybe Schema) cachedSchema sr k = liftIO $ C.lookup (srCache sr) k