hw-kafka-avro 1.2.0 → 1.3.0

raw patch · 5 files changed

+51/−16 lines, 5 filesdep +http-typesPVP ok

version bump matches the API change (PVP)

Dependencies added: http-types

API changes (from Hackage documentation)

Files

• hw-kafka-avro.cabal +2/−1
• src/Kafka/Avro.hs +32/−2
• src/Kafka/Avro/Decode.hs +1/−1
• src/Kafka/Avro/Encode.hs +9/−6
• src/Kafka/Avro/SchemaRegistry.hs +7/−6

@@ -1,5 +1,5 @@ name:                hw-kafka-avro-version:             1.2.0+version:             1.3.0 synopsis:            Avro support for Kafka infrastructure description:         Please see README.md homepage:            https://github.com/haskell-works/hw-kafka-avro#readme@@ -35,6 +35,7 @@                        errors,                        hashable,                        http-client,+                       http-types,                        mtl,                        pure-zlib >= 0.6,                        semigroups,

@@ -1,7 +1,37 @@ module Kafka.Avro ( module X+, propagateKeySchema+, propagateValueSchema ) where +import Control.Monad.IO.Class+import Data.ByteString.Lazy++import Kafka.Avro.Decode         as X+import Kafka.Avro.Encode         as X import Kafka.Avro.SchemaRegistry as X-import Kafka.Avro.Decode as X-import Kafka.Avro.Encode as X++-- | Registers schema that was used for a given payload against the specified subject as a key shema.+-- It is possible that a given payload doesn't have schema registered against it, in this case no prapagation happens.+propagateKeySchema :: MonadIO m => SchemaRegistry -> Subject -> ByteString -> m (Either SchemaRegistryError (Maybe SchemaId))+propagateKeySchema sr subj = propagateSchema sr (keySubject subj)++-- | Registers schema that was used for a given payload against the specified subject as a value schema.+-- It is possible that a given payload doesn't have schema registered against it, in this case no prapagation happens.+propagateValueSchema :: MonadIO m => SchemaRegistry -> Subject -> ByteString -> m (Either SchemaRegistryError (Maybe SchemaId))+propagateValueSchema sr subj = propagateSchema sr (valueSubject subj)++propagateSchema :: MonadIO m+                => SchemaRegistry+                -> Subject+                -> ByteString+                -> m (Either SchemaRegistryError (Maybe SchemaId))+propagateSchema sr subj bs = do+  case extractSchemaId bs of+    Nothing -> return $ Right Nothing+    Just (sid, _) -> do+      mSchema <- loadSchema sr sid+      case mSchema of+        Left (SchemaRegistrySchemaNotFound _) -> return $ Right Nothing+        Left err                              -> return $ Left err+        Right schema                          -> fmap Just <$> sendSchema sr subj schema

@@ -1,7 +1,7 @@ module Kafka.Avro.Decode (   DecodeError(..)-, decodeWithSchema+, decodeWithSchema, extractSchemaId ) where  import           Control.Monad.IO.Class    (MonadIO)

@@ -1,6 +1,7 @@ {-# LANGUAGE OverloadedStrings #-} module Kafka.Avro.Encode ( encodeKey, encodeValue+, keySubject, valueSubject , encodeWithSchema , EncodeError(..) ) where@@ -18,6 +19,12 @@ data EncodeError = EncodeRegistryError SchemaRegistryError   deriving (Show, Eq) +keySubject :: Subject -> Subject+keySubject (Subject subj) = Subject (subj <> "-key")++valueSubject :: Subject -> Subject+valueSubject (Subject subj) = Subject (subj <> "-value")+ -- | Encodes a provided value as a message key. -- -- Registers the schema in SchemaRegistry with "<subject>-key" subject.@@ -26,9 +33,7 @@           -> Subject           -> a           -> m (Either EncodeError ByteString)-encodeKey sr (Subject subj) a =-  let keySubj = Subject (subj <> "-key")-   in encodeWithSchema sr keySubj a+encodeKey sr subj = encodeWithSchema sr (keySubject subj)  -- | Encodes a provided value as a message value. --@@ -38,9 +43,7 @@             -> Subject             -> a             -> m (Either EncodeError ByteString)-encodeValue sr (Subject subj) a =-  let valSubj = Subject (subj <> "-value")-   in encodeWithSchema sr valSubj a+encodeValue sr subj = encodeWithSchema sr (valueSubject subj)  -- | Encodes a provided value into Avro -- and registers value's schema in SchemaRegistry.

@@ -27,6 +27,7 @@ import           GHC.Exception import           GHC.Generics               (Generic) import           Network.HTTP.Client        (Manager, defaultManagerSettings, newManager)+import           Network.HTTP.Types import           Servant.API import           Servant.Client @@ -51,6 +52,7 @@ data SchemaRegistryError = SchemaRegistryConnectError String                          | SchemaDecodeError SchemaId String                          | SchemaRegistryLoadError SchemaId+                         | SchemaRegistrySchemaNotFound SchemaId                          | SchemaRegistryError SchemaId                          | SchemaRegistrySendError String                          deriving (Show, Eq)@@ -115,8 +117,6 @@ #endif getSchemaById :<|> putSchema = client api -type P = ServantError- #if MIN_VERSION_servant(0,9,1) sendSchemaToSR :: MonadIO m => ClientEnv -> Subject -> Schema -> m (Either SchemaRegistryError SchemaId) sendSchemaToSR env subj s =@@ -144,10 +144,11 @@   where     unwrapResponse (RegisteredSchema s) = s     toSRError msg = case msg of-      ConnectionError ex   -> SchemaRegistryConnectError (show ex)-      FailureResponse{}    -> SchemaRegistryLoadError sid-      DecodeFailure de _ _ -> SchemaDecodeError sid de-      _                    -> SchemaRegistryLoadError sid+      ConnectionError ex                                       -> SchemaRegistryConnectError (show ex)+      FailureResponse {responseStatus=Status {statusCode=404}} -> SchemaRegistrySchemaNotFound sid+      FailureResponse{}                                        -> SchemaRegistryLoadError sid+      DecodeFailure de _ _                                     -> SchemaDecodeError sid de+      _                                                        -> SchemaRegistryLoadError sid  --------------------------------------------------------------------- fullTypeName :: Schema -> SchemaName