hw-kafka-avro (empty) → 1.1.0
raw patch · 11 files changed
+548/−0 lines, 11 filesdep +QuickCheckdep +aesondep +avrosetup-changed
Dependencies added: QuickCheck, aeson, avro, base, binary, bytestring, cache, containers, errors, hashable, hspec, http-client, hw-kafka-avro, mtl, pure-zlib, semigroups, servant, servant-client, text, transformers, unordered-containers
Files
- LICENSE +30/−0
- README.md +5/−0
- Setup.hs +2/−0
- example/Main.hs +34/−0
- example/Message.hs +40/−0
- hw-kafka-avro.cabal +108/−0
- src/Kafka/Avro.hs +7/−0
- src/Kafka/Avro/Decode.hs +57/−0
- src/Kafka/Avro/Encode.hs +62/−0
- src/Kafka/Avro/SchemaRegistry.hs +201/−0
- test/Spec.hs +2/−0
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright Alexey Raga (c) 2016++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of Alexey Raga nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ README.md view
@@ -0,0 +1,5 @@+# kafka-avro-serialiser++Avro serialiser/deserialiser for Kafka messages. Uses SchemaRegistry for schema compatibility and discoverability functionality.++This library is meant to be compatible (on both sending and receiving sides) with Java kafka/avro serialiser (written by Confluent).
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ example/Main.hs view
@@ -0,0 +1,34 @@+{-# LANGUAGE OverloadedStrings, ScopedTypeVariables #-}+module Main where++import Control.Monad.Trans.Except+import Data.Monoid+import qualified Data.Aeson as J+import Data.Avro as A+import Data.Avro.Schema as S+import qualified Data.Avro.Types as AT++import Data.Int+import Data.Text+import Kafka.Avro+import Message++exampleMessage = TestMessage 1 "Example" True 12345678++data AppError = EncError EncodeError | DecError DecodeError+ deriving (Show)++main :: IO ()+main = do+ sr <- schemaRegistry "http://localhost:8081"+ res <- runExceptT $ roundtrip sr+ print res++roundtrip :: SchemaRegistry -> ExceptT AppError IO TestMessage+roundtrip sr = do+ 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
+ example/Message.hs view
@@ -0,0 +1,40 @@+{-# LANGUAGE OverloadedStrings #-}+module Message+( TestMessage(..)+) where+--++import Data.Avro+import Data.Int+import Data.Text+import Data.Avro.Schema+import qualified Data.Avro.Types as AT++data TestMessage = TestMessage Int64 Text Bool Int64 deriving (Show, Eq, Ord)++testMessageSchema =+ let fld nm = Field nm [] Nothing Nothing+ in Record "TestMessage" (Just "hw.kafka.avro.test") [] Nothing Nothing+ [ fld "id" Long Nothing+ , fld "name" String Nothing+ , fld "is_active" Boolean Nothing+ , fld "timestamp" Long Nothing+ ]++instance FromAvro TestMessage where+ fromAvro (AT.Record _ r) =+ TestMessage <$> r .: "id"+ <*> r .: "name"+ <*> r .: "is_active"+ <*> r .: "timestamp"+ fromAvro v = badValue v "TestMessage"++instance ToAvro TestMessage where+ toAvro (TestMessage i s d t) =+ record testMessageSchema+ [ "id" .= i+ , "name" .= s+ , "is_active" .= d+ , "timestamp" .= t+ ]+ schema = pure testMessageSchema
+ hw-kafka-avro.cabal view
@@ -0,0 +1,108 @@+name: hw-kafka-avro+version: 1.1.0+synopsis: Avro support for Kafka infrastructure+description: Please see README.md+homepage: https://github.com/haskell-works/hw-kafka-avro#readme+license: BSD3+license-file: LICENSE+author: Alexey Raga+maintainer: alexey.raga@gmail.com+copyright: Alexey Raga+category: Services+build-type: Simple+extra-source-files: README.md+cabal-version: >=1.10++flag examples+ default: False+ manual: True+ description: Also compile examples++library+ hs-source-dirs: src+ exposed-modules: Kafka.Avro,+ Kafka.Avro.Decode,+ Kafka.Avro.Encode,+ Kafka.Avro.SchemaRegistry++ build-depends: aeson,+ avro,+ base >= 4.7 && < 5,+ binary,+ bytestring,+ cache,+ containers,+ errors,+ hashable,+ http-client,+ mtl,+ pure-zlib >= 0.6,+ semigroups,+ servant,+ servant-client,+ text,+ transformers,+ unordered-containers+ default-language: Haskell2010++executable kafka-avro-example+ hs-source-dirs: example+ main-is: Main.hs+ other-modules: Message+ ghc-options: -threaded -rtsopts -with-rtsopts=-N+ build-depends: aeson+ , avro+ , base+ , binary+ , bytestring+ , cache+ , containers+ , errors+ , hashable+ , hw-kafka-avro+ , http-client+ , mtl+ , pure-zlib >= 0.6+ , semigroups+ , servant+ , servant-client+ , text+ , transformers+ , unordered-containers+ default-language: Haskell2010+ if flag(examples)+ buildable: True+ else+ buildable: False++test-suite kafka-avro-test+ type: exitcode-stdio-1.0+ hs-source-dirs: test+ main-is: Spec.hs+ build-depends: QuickCheck+ , aeson+ , avro+ , base+ , binary+ , bytestring+ , cache+ , containers+ , errors+ , hashable+ , hw-kafka-avro+ , hspec+ , http-client+ , mtl+ , pure-zlib >= 0.6+ , semigroups+ , servant+ , servant-client+ , text+ , transformers+ , unordered-containers+ ghc-options: -threaded -rtsopts -with-rtsopts=-N+ default-language: Haskell2010++source-repository head+ type: git+ location: https://github.com/haskell-works/hw-kafka-avro
+ src/Kafka/Avro.hs view
@@ -0,0 +1,7 @@+module Kafka.Avro+( module X+) where++import Kafka.Avro.SchemaRegistry as X+import Kafka.Avro.Decode as X+import Kafka.Avro.Encode as X
+ src/Kafka/Avro/Decode.hs view
@@ -0,0 +1,57 @@+module Kafka.Avro.Decode+(+ DecodeError(..)+, decodeWithSchema+) where++import Control.Monad.IO.Class (MonadIO)+import Data.Avro as A (FromAvro, Result (..), decode)+import Data.Avro.Schema (Schema)+import Data.Bits (shiftL)+import Data.ByteString.Lazy (ByteString)+import qualified Data.ByteString.Lazy as BL hiding (zipWith)+import Data.Int+import Kafka.Avro.SchemaRegistry++data DecodeError = DecodeRegistryError SchemaRegistryError+ | BadPayloadNoSchemaId+ | DecodeError 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 =+ case schemaData of+ Left err -> return $ Left err+ Right (sid, payload) -> do+ res <- leftMap DecodeRegistryError <$> loadSchema sr sid+ return $ res >>= decode payload+ where+ schemaData = maybe (Left BadPayloadNoSchemaId) Right (extractSchemaId bs)+ decode p s = resultToEither s (A.decode s p)++extractSchemaId :: ByteString -> Maybe (SchemaId, ByteString)+extractSchemaId bs = do+ (_ , b0) <- BL.uncons bs+ (w1, b1) <- BL.uncons b0+ (w2, b2) <- BL.uncons b1+ (w3, b3) <- BL.uncons b2+ (w4, b4) <- BL.uncons b3+ let ints = fromIntegral <$> [w4, w3, w2, w1] :: [Int32]+ let int = sum $ zipWith shiftL ints [0, 8, 16, 24]+ return (SchemaId int, b4)++leftMap :: (e -> e') -> Either e r -> Either e' r+leftMap _ (Right r) = Right r+leftMap f (Left e) = Left (f e)++resultToEither :: Schema -> A.Result a -> Either DecodeError a+resultToEither sc res = case res of+ Success a -> Right a+ Error msg -> Left $ DecodeError sc msg
+ src/Kafka/Avro/Encode.hs view
@@ -0,0 +1,62 @@+{-# LANGUAGE OverloadedStrings #-}+module Kafka.Avro.Encode+( encodeKey, encodeValue+, encodeWithSchema+, EncodeError(..)+) where++import Control.Monad.IO.Class (MonadIO)+import Data.Avro as A (ToAvro, encode, schemaOf)+import Data.Avro.Schema (Schema)+import qualified Data.Binary as B+import Data.Bits (shiftL)+import Data.ByteString.Lazy (ByteString)+import qualified Data.ByteString.Lazy as BL hiding (zipWith)+import Data.Monoid+import Kafka.Avro.SchemaRegistry++data EncodeError = EncodeRegistryError SchemaRegistryError+ deriving (Show, Eq)++-- | Encodes a provided value as a message key.+--+-- Registers the schema in SchemaRegistry with "<subject>-key" subject.+encodeKey :: (MonadIO m, ToAvro a)+ => SchemaRegistry+ -> Subject+ -> a+ -> m (Either EncodeError ByteString)+encodeKey sr (Subject subj) a =+ let keySubj = Subject (subj <> "-key")+ in encodeWithSchema sr keySubj a++-- | Encodes a provided value as a message value.+--+-- Registers the schema in SchemaRegistry with "<subject>-value" subject.+encodeValue :: (MonadIO m, ToAvro a)+ => SchemaRegistry+ -> Subject+ -> a+ -> m (Either EncodeError ByteString)+encodeValue sr (Subject subj) a =+ let valSubj = Subject (subj <> "-value")+ in encodeWithSchema sr valSubj a++-- | 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)+ case mbSid of+ Left err -> return . Left . EncodeRegistryError $ err+ Right sid -> return . Right $ appendSchemaId sid (encode a)+++appendSchemaId :: SchemaId -> ByteString -> ByteString+appendSchemaId (SchemaId sid) bs =+ -- add a "magic byte" followed by schema id+ BL.cons (toEnum 0) (B.encode sid) <> bs
+ src/Kafka/Avro/SchemaRegistry.hs view
@@ -0,0 +1,201 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeOperators #-}++module Kafka.Avro.SchemaRegistry+( schemaRegistry, loadSchema, sendSchema+, SchemaId(..), Subject(..)+, SchemaRegistry, SchemaRegistryError(..)+, RegisteredSchema(..)+) where++import Control.Monad.IO.Class+import Control.Monad.Trans.Except (ExceptT (..), runExceptT, withExceptT)+import Data.Aeson+import Data.Avro.Schema (Schema, Type (..), typeName)+import Data.Cache as C+import Data.Hashable+import Data.Int+import Data.Proxy+import Data.Text (Text, append, cons)+import qualified Data.Text.Encoding as Text+import qualified Data.Text.Lazy.Encoding as LText+import GHC.Exception+import GHC.Generics (Generic)+import Network.HTTP.Client (Manager, defaultManagerSettings, newManager)+import Servant.API+import Servant.Client++newtype SchemaId = SchemaId Int32 deriving (Eq, Ord, Show, Hashable)+newtype SchemaName = SchemaName Text deriving (Eq, Ord, Show, Hashable)++newtype Subject = Subject Text deriving (Eq, Show, Generic, Hashable)++newtype RegisteredSchema = RegisteredSchema Schema deriving (Generic, Show)++data SchemaRegistry = SchemaRegistry+ { srCache :: Cache SchemaId Schema+ , srReverseCache :: Cache (Subject, SchemaName) SchemaId+#if MIN_VERSION_servant(0,9,1)+ , srClientEnv :: ClientEnv+#else+ , srManager :: Manager+ , srBaseUrl :: BaseUrl+#endif+ }++data SchemaRegistryError = SchemaRegistryConnectError String+ | SchemaDecodeError SchemaId String+ | SchemaRegistryLoadError SchemaId+ | SchemaRegistryError SchemaId+ | SchemaRegistrySendError String+ deriving (Show, Eq)++schemaRegistry :: MonadIO m => String -> m SchemaRegistry+schemaRegistry url = liftIO $+ SchemaRegistry+ <$> newCache Nothing+ <*> newCache Nothing+#if MIN_VERSION_servant(0,9,1)+ <*> (ClientEnv <$> newManager defaultManagerSettings <*> parseBaseUrl url)+#else+ <*> newManager defaultManagerSettings+ <*> parseBaseUrl url+#endif++loadSchema :: MonadIO m => SchemaRegistry -> SchemaId -> m (Either SchemaRegistryError Schema)+loadSchema sr sid = do+ sc <- cachedSchema sr sid+ case sc of+ Just s -> return (Right s)+ Nothing -> do+#if MIN_VERSION_servant(0,9,1)+ res <- loadSchemaFromSR (srClientEnv sr) sid+#else+ res <- loadSchemaFromSR (srManager sr) (srBaseUrl sr) sid+#endif+ _ <- traverse (cacheSchema sr sid) res+ return res++sendSchema :: MonadIO m => SchemaRegistry -> Subject -> Schema -> m (Either SchemaRegistryError SchemaId)+sendSchema sr subj sc = do+ sid <- cachedId sr subj schemaName+ case sid of+ Just sid' -> return (Right sid')+ Nothing -> do+#if MIN_VERSION_servant(0,9,1)+ res <- sendSchemaToSR (srClientEnv sr) subj sc+#else+ res <- sendSchemaToSR (srManager sr) (srBaseUrl sr) subj sc+#endif+ _ <- traverse (cacheId sr subj schemaName) res+ _ <- traverse (\sid' -> cacheSchema sr sid' sc) res+ return res+ where+ schemaName = fullTypeName sc++------------------ PRIVATE: HELPERS --------------------------------------------++type API = "schemas" :> "ids" :> Capture "id" Int32 :> Get '[JSON] RegisteredSchema+ :<|> "subjects" :> Capture "subject" Subject :> "versions" :> ReqBody '[JSON] RegisteredSchema :> Post '[JSON] SchemaId+api :: Proxy API+api = Proxy++#if MIN_VERSION_servant(0,9,1)+--ExceptT ServantError IO SchemaId+getSchemaById :: Int32 -> ClientM RegisteredSchema+putSchema :: Subject -> RegisteredSchema -> ClientM SchemaId+#else+getSchemaById :: Int32 -> Manager -> BaseUrl -> ClientM RegisteredSchema+putSchema :: Subject -> RegisteredSchema -> Manager -> BaseUrl -> ClientM SchemaId+#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 =+ runExceptT $ withExceptT toSRError $ runServant env $ putSchema subj (RegisteredSchema s)+#else+sendSchemaToSR :: MonadIO m => Manager -> BaseUrl -> Subject -> Schema -> m (Either SchemaRegistryError SchemaId)+sendSchemaToSR m u subj s =+ liftExceptT . withExceptT toSRError $ putSchema subj (RegisteredSchema s) m u+#endif+ where+ toSRError msg = case msg of+ ConnectionError ex -> SchemaRegistryConnectError (show ex)+ DecodeFailure de _ _ -> SchemaRegistrySendError de+ err -> SchemaRegistrySendError (show err)++#if MIN_VERSION_servant(0,9,1)+loadSchemaFromSR :: MonadIO m => ClientEnv -> SchemaId -> m (Either SchemaRegistryError Schema)+loadSchemaFromSR env sid@(SchemaId i) =+ runExceptT (withExceptT toSRError $ runServant env $ unwrapResponse <$> getSchemaById i)+#else+loadSchemaFromSR :: MonadIO m => Manager -> BaseUrl -> SchemaId -> m (Either SchemaRegistryError Schema)+loadSchemaFromSR m u sid@(SchemaId i) =+ liftExceptT (withExceptT toSRError $ unwrapResponse <$> getSchemaById i m u)+#endif+ 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++---------------------------------------------------------------------+fullTypeName :: Schema -> SchemaName+fullTypeName r = SchemaName $ case r of+ Record{} -> maybe (typeName r)+ (\ns -> ns `append` ('.' `cons` typeName r))+ (namespace r)+ _ -> typeName r++cachedSchema :: MonadIO m => SchemaRegistry -> SchemaId -> m (Maybe Schema)+cachedSchema sr k = liftIO $ C.lookup (srCache sr) k+{-# INLINE cachedSchema #-}++cacheSchema :: MonadIO m => SchemaRegistry -> SchemaId -> Schema -> m ()+cacheSchema sr k v = liftIO $ C.insert (srCache sr) k v+{-# INLINE cacheSchema #-}++cachedId :: MonadIO m => SchemaRegistry -> Subject -> SchemaName -> m (Maybe SchemaId)+cachedId sr subj scn = liftIO $ C.lookup (srReverseCache sr) (subj, scn)+{-# INLINE cachedId #-}++cacheId :: MonadIO m => SchemaRegistry -> Subject -> SchemaName -> SchemaId -> m ()+cacheId sr subj scn sid = liftIO $ C.insert (srReverseCache sr) (subj, scn) sid+{-# INLINE cacheId #-}++instance FromJSON RegisteredSchema where+ parseJSON (Object v) =+ withObject "expected schema" (\obj -> do+ sch <- obj .: "schema"+ maybe mempty (return . RegisteredSchema) (decode $ LText.encodeUtf8 sch)+ ) (Object v)++ parseJSON _ = mempty++instance ToJSON RegisteredSchema where+ toJSON (RegisteredSchema v) = object ["schema" .= LText.decodeUtf8 (encode $ toJSON v)]++instance FromJSON SchemaId where+ parseJSON (Object v) = SchemaId <$> v .: "id"+ parseJSON _ = mempty++instance ToHttpApiData Subject where+ toUrlPiece (Subject s) = toUrlPiece s++#if MIN_VERSION_servant(0,9,1)+runServant :: MonadIO m => ClientEnv -> ClientM a -> ExceptT ServantError m a+runServant env cli = ExceptT $ liftIO (runClientM cli env)+#else+liftExceptT :: MonadIO m => ExceptT l IO r -> m (Either l r)+liftExceptT = liftIO . runExceptT+#endif
+ test/Spec.hs view
@@ -0,0 +1,2 @@+main :: IO ()+main = putStrLn "Test suite not yet implemented"