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,7 +1,7 @@ -cabal-version: 1.12 +cabal-version: 2.4 name: hw-kafka-avro -version: 4.0.1 +version: 5.0.0 synopsis: Avro support for Kafka infrastructure description: Avro support for Kafka infrastructure. category: Services @@ -10,7 +10,7 @@ author: Alexey Raga maintainer: alexey.raga@gmail.com copyright: Alexey Raga -license: BSD3 +license: BSD-3-Clause license-file: LICENSE build-type: Simple extra-source-files: @@ -34,7 +34,7 @@ hs-source-dirs: src build-depends: aeson - , avro >=0.4 + , avro >=0.5 && <0.6 , base >=4.7 && <5 , binary , bytestring @@ -63,7 +63,7 @@ ghc-options: -threaded -rtsopts -with-rtsopts=-N build-depends: aeson - , avro + , avro >=0.5 && <0.6 , base , binary , bytestring @@ -92,7 +92,7 @@ build-depends: QuickCheck , aeson - , avro + , avro >=0.5 && <0.6 , base , binary , bytestring 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,21 @@ -{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} 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 +28,79 @@ 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) +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 key. +-- +-- Registers the schema in SchemaRegistry 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. -- -- 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) +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. +-- +-- Registers the schema in SchemaRegistry 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 @@ -24,7 +24,7 @@ import Control.Monad.IO.Class (MonadIO, liftIO) import Data.Aeson import Data.Aeson.Types (typeMismatch) -import Data.Avro.Schema (Schema, Type (..), typeName) +import Data.Avro.Schema.Schema (Schema (..), typeName) import Data.Bifunctor (bimap) import Data.Cache as C import Data.Hashable (Hashable)