kafka-client (empty) → 0.7.0.0
raw patch · 10 files changed
+849/−0 lines, 10 filesdep +QuickCheckdep +basedep +bytestringsetup-changed
Dependencies added: QuickCheck, base, bytestring, cereal, digest, dlist, hspec, hspec-discover, kafka-client, network, snappy, time, zlib
Files
- Kafka.hs +163/−0
- Kafka/Internal.hs +11/−0
- Kafka/Internal/Request.hs +153/−0
- Kafka/Internal/Response.hs +61/−0
- Kafka/Internal/Transport.hs +88/−0
- Kafka/Internal/Types.hs +289/−0
- LICENSE +20/−0
- Setup.hs +2/−0
- kafka-client.cabal +61/−0
- test/Spec.hs +1/−0
+ Kafka.hs view
@@ -0,0 +1,163 @@+{-# LANGUAGE NamedFieldPuns #-}+-- |+-- Module : Kafka+-- Copyright : Abhinav Gupta 2015+-- License : MIT+--+-- Maintainer : mail@abhinavg.net+-- Stability : experimental+-- Portability : GHC+--+-- A library to interact with Apache Kafka 0.7.+--+module Kafka+ (+ -- * Main interface+ --+ -- | Requests to Kafka can be made using 'produce', 'fetch', and+ -- 'offsets'. For 'produce' and 'fetch', the functions automatically+ -- decide whether the request needs to be a single @Produce@/@Fetch@+ -- request or a @Multi*@ request.+ --+ -- The request operations send requests and receive responses using any+ -- type that is an instance of 'Transport'. 'withConnection' produces one+ -- such object.+ --+ withConnection+ , produce+ , fetch+ , offsets++ -- * Types++ , Produce(..)++ , Fetch(..)+ , FetchResponse(..)++ , Offsets(..)+ , OffsetsTime(..)++ , Topic(..)+ , Offset(..)+ , Partition(..)+ , Size(..)+ , Count(..)++ -- * Other++ , Error(..)+ , Response++ -- * Transport++ , Socket+ , Transport(..)+ ) where++import Control.Applicative+import Data.Monoid++import qualified Data.Serialize as C++import Kafka.Internal++-- | Receives the next response from the connection using the given+-- deserializer.+--+-- Returns either an 'Error' for Kafka errors or the deseriazlied result.+recvResponse+ :: Transport t+ => t -- ^ Transport from which the response will be read.+ -> C.Get a -- ^ Deserializer+ -> IO (Response a)+recvResponse transport getBody = do+ -- TODO this probably belongs in an internal module somewhere+ -- TODO use protocol error or something for parse errors instead of this+ -- TODO maybe also catch IO exceptions+ -- TODO alternatively, throw protocol and IO exceptions instead of failing+ -- like this.+ respLen <- C.runGet getLength <$> recv transport 4 >>= either fail return+ response <- recvExactly transport respLen+ either fail return $ C.runGet getResponse response+ where+ getLength = fromIntegral <$> C.getWord32be+ getResponse = do+ err <- C.get+ body <- getBody+ return $ maybe (Right body) Left err++-- | Sends the given 'Produce' requests to Kafka.+--+-- If multiple requests are supplied, a @MultiProduce@ request is made.+--+-- @+-- 'withConnection' \"localhost\" 9092 $ \\conn ->+-- produce conn [+-- 'Produce' ('Topic' \"my-topic\") ('Partition' 0) [\"foo\"]+-- , Produce \"another-topic\" 0+-- [\"multiple\", \"messages\"]+-- ]+-- @+--+-- Note that string literals may be used in place of 'Topic' (with the+-- @OverloadedStrings@ GHC extension), and integer literals may be used in+-- place of 'Partition'.+produce :: Transport t => t -> [Produce] -> IO ()+produce transport reqs = case reqs of+ [] -> return ()+ [x] -> send transport (C.runPut $ putProduceRequest x)+ xs -> send transport (C.runPut $ putMultiProduceRequest xs)++-- | 'Fetch'es messages from Kafka.+--+-- If multiple Fetch requests are supplied, a @MultiFetch@ request is made.+--+-- @+-- 'withConnection' \"localhost\" 9092 $ \\conn -> do+-- Right ['FetchResponse' messages newOffset] <- fetch conn [+-- 'Fetch' ('Topic' \"test-topic\") ('Partition' 0) (Offset 42) 1024+-- ]+-- {- Consume the messages here -}+-- response <- fetch conn ['Fetch' \"test-topic\" 0 newOffset 1024]+-- {- ... -}+-- @+--+-- Returns a list of 'FetchResponse's in the same order as the 'Fetch'+-- requests. Each response contains the messages returned for the+-- corresponding request and the new offset at which the next request should+-- be made for that request to get the messages that follow.+--+-- If a response for a request contains no messages, the specified+-- topic-partition pair has been exhausted.+--+-- Note that string literals may be used in place of 'Topic' (with the+-- @OverloadedStrings@ GHC extension), and integer literals may be used in+-- place of 'Offset'.+--+fetch :: Transport t => t -> [Fetch] -> IO (Response [FetchResponse])+fetch transport reqs = case reqs of+ [] -> return (Right mempty)+ [x] -> do+ send transport (C.runPut $ putFetchRequest x)+ fmap (:[]) <$> recvResponse transport (getFetchResponse x)+ xs -> do+ send transport (C.runPut $ putMultiFetchRequest xs)+ recvResponse transport (getMultiFetchResponse xs)++-- | Retrieve message offsets from Kafka.+--+-- @+-- 'withConnection' \"localhost\" 9092 $ \\conn -> do+-- Right [os] <- offsets conn (Offsets \"topic\" 0 OffsetsEarliest 1)+-- fetch conn ['Fetch' \"topic\" (Partition 0) os 10] >>= doSomething+-- @+--+-- Note that string literals may be used in place of 'Topic' (with the+-- @OverloadedStrings@ GHC extension), and integer literals may be used in+-- place of 'Count'.+--+offsets :: Transport t => t -> Offsets -> IO (Response [Offset])+offsets transport req = do+ send transport (C.runPut $ putOffsetsRequest req)+ recvResponse transport getOffsetsResponse
+ Kafka/Internal.hs view
@@ -0,0 +1,11 @@+module Kafka.Internal+ ( module Kafka.Internal.Request+ , module Kafka.Internal.Response+ , module Kafka.Internal.Transport+ , module Kafka.Internal.Types+ ) where++import Kafka.Internal.Request+import Kafka.Internal.Response+import Kafka.Internal.Transport+import Kafka.Internal.Types
+ Kafka/Internal/Request.hs view
@@ -0,0 +1,153 @@+module Kafka.Internal.Request (+ Produce(..)+ , putProduceRequest+ , putMultiProduceRequest++ , Fetch(..)+ , putFetchRequest+ , putMultiFetchRequest++ , Offsets(..)+ , putOffsetsRequest+ ) where++import Control.Applicative+import Data.ByteString (ByteString)+import Data.Foldable (Foldable)++import qualified Data.ByteString as B+import qualified Data.Foldable as Fold+import qualified Data.Serialize as C++import Kafka.Internal.Types++-- | 'Put's the items in the given Foldable in-order and prepends the result+-- with a 16-bit integer containing the total number of items that were put.+putWithCountPrefix :: Foldable f => f a -> C.Putter a -> C.Put+putWithCountPrefix xs put =+ let (count, p) = Fold.foldr' go (0, pure ()) xs+ in C.putWord16be count *> p+ where+ go a (c, p) = (c + 1, put a *> p)++-- | Prepends the given Put with a 32-bit integer containing the total size of+-- the given Put.+putWithLengthPrefix :: C.Put -> C.Put+putWithLengthPrefix p = do+ C.putWord32be $ fromIntegral (B.length bs)+ C.putByteString bs+ where+ bs = C.runPut p++-- | Encodes a request with the given 'RequestType'.+--+-- More specifically, this prepends the given put with the request type and+-- the size of the request.+encodeRequest :: RequestType -> C.Put -> C.Put+encodeRequest reqType p = putWithLengthPrefix $ C.put reqType >> p++-- | A request to send messages down a Kafka topic-partition pair.+--+-- Produce requests do not have a corresponding response. There is no way of+-- knowing in Kafka 0.7 if a message was successfully @Produce@d.+data Produce = Produce {+ produceTopic :: {-# UNPACK #-} !Topic+ -- ^ Kafka topic to which the messages will be sent.+ , producePartition :: {-# UNPACK #-} !Partition+ -- ^ Partition of the topic.+ , produceMessages :: [ByteString]+ -- ^ List of message payloads.+ --+ -- For those concerned with low-leveld details: These messages will be+ -- compressed using <https://code.google.com/p/snappy/ Snappy> compression.+ } deriving (Show, Read, Eq)++encodeProduce :: Produce -> C.Put+encodeProduce (Produce topic partition messages) = do+ C.put topic+ C.put partition+ putWithLengthPrefix (C.put (MessageSet messages))++-- | @Put@s the given single @Produce@ request.+putProduceRequest :: Produce -> C.Put+putProduceRequest =+ encodeRequest ProduceRequestType . encodeProduce++-- | @Put@s the given @MultiProduce@ request.+putMultiProduceRequest :: [Produce] -> C.Put+putMultiProduceRequest reqs =+ encodeRequest MultiProduceRequestType $+ putWithCountPrefix reqs encodeProduce++-- | A request to fetch messages from a particular Kafka topic-partition pair.+--+-- 'Kafka.FetchResponse' contains responses for this kind of request.+data Fetch = Fetch {+ fetchTopic :: {-# UNPACK #-} !Topic+ -- ^ Kafka topic from which messages will be fetched.+ , fetchPartition :: {-# UNPACK #-} !Partition+ -- ^ Partition of the topic.+ , fetchOffset :: {-# UNPACK #-} !Offset+ -- ^ Offset at which the fetch will start.+ --+ -- Kafka offloads the responsiblity of knowing this to the client. That+ -- means that if an offset is specified here that is not a real message+ -- start, Kafka will spit out garbage.+ --+ -- Use 'Kafka.offsets' to find valid offsets.+ , fetchSize :: {-# UNPACK #-} !Size+ -- ^ Maximum size of the returned messages.+ --+ -- Note, this is /not/ the number of messages. This is the maximum+ -- combined size of the returned /compressed/ messages.+ } deriving (Show, Read, Eq)++encodeFetch :: Fetch -> C.Put+encodeFetch (Fetch topic partition offset maxSize) = do+ C.put topic+ C.put partition+ C.put offset+ C.put maxSize++-- | @Put@s the given single @Fetch@ request.+putFetchRequest :: Fetch -> C.Put+putFetchRequest = encodeRequest FetchRequestType . encodeFetch++-- | @Put@s the given @MultiFetch@ request.+putMultiFetchRequest :: [Fetch] -> C.Put+putMultiFetchRequest reqs =+ encodeRequest MultiFetchRequestType $+ putWithCountPrefix reqs encodeFetch++-- | A request to retrieve offset information from Kafka.+--+-- The response for this kind of request is a list of 'Offset's.+data Offsets = Offsets {+ offsetsTopic :: {-# UNPACK #-} !Topic+ -- ^ Kafka topic from which offsets will be retrieved.+ , offsetsPartition :: {-# UNPACK #-} !Partition+ -- ^ Partition of the topic.+ , offsetsTime :: !OffsetsTime+ -- ^ Time around which offsets will be retrieved.+ --+ -- If you provide a time for this, keep in mind that the response will not+ -- contain the precise offset that occurred around that time. It will return+ -- up to @offsetsCount@ offsets in descending, each being the first offset+ -- of every segment file for the specified partition with a modified time+ -- less than the specified time, and possibly a "high water mark" for the+ -- last segment of the partition (if it was modified before the specified+ -- time) which specifies the offset at which the next message to that+ -- partition will be written.+ , offsetsCount :: {-# UNPACK #-} !Count+ -- ^ Maximum number of offsets that will be retrieved.+ } deriving (Show, Read, Eq)++-- | @Put@s the given @Offsets@ request.+putOffsetsRequest :: Offsets -> C.Put+putOffsetsRequest (Offsets topic partition time maxNumber) =+ encodeRequest OffsetsRequestType $ do+ C.put topic+ C.put partition+ C.put time+ C.put maxNumber+
+ Kafka/Internal/Response.hs view
@@ -0,0 +1,61 @@+module Kafka.Internal.Response+ ( Response+ , FetchResponse(..)+ , getFetchResponse+ , getMultiFetchResponse+ , getOffsetsResponse+ ) where++import Control.Applicative+import Control.Monad+import Data.ByteString (ByteString)+import Data.Monoid++import qualified Data.Serialize as C++import Kafka.Internal.Request+import Kafka.Internal.Types++-- | A response from Kafka can either be a failure or the value that was+-- expected.+type Response a = Either Error a++-- | Result of a single Kafka 'Kafka.Fetch'.+data FetchResponse = FetchResponse {+ fetchMessages :: [ByteString]+ -- ^ List of messages returned in the response for the 'Kafka.Fetch'+ -- request.+ , fetchNewOffset :: {-# UNPACK #-} !Offset+ -- ^ New offset at which the next 'Fetch' request should start reading in+ -- the same topic and partition to access the messages that follow the+ -- messages returned in this response.+ } deriving (Show, Read, Eq)++-- | Parses a single @FetchResponse@ for the given @Fetch@ request.+getFetchResponse :: Fetch -> C.Get FetchResponse+getFetchResponse Fetch{fetchOffset = oldOffset} = do+ startBytes <- C.remaining+ messages <- fromMessageSet . mconcat <$> many C.get+ endBytes <- C.remaining+ let newOffset = oldOffset + fromIntegral (startBytes - endBytes)+ return $ FetchResponse messages $! newOffset++-- | Parses @FetchResponse@s for each of the given @Fetch@ requests.+getMultiFetchResponse :: [Fetch] -> C.Get [FetchResponse]+getMultiFetchResponse requests = forM requests $ \request -> do+ size <- fromIntegral <$> C.getWord32be+ C.skip 2 -- skip error+ isolate' (size - 2) (getFetchResponse request)++-- | Parses the response for an @Offsets@ request.+getOffsetsResponse :: C.Get [Offset]+getOffsetsResponse = do+ numOffsets <- C.getWord32be+ replicateM (fromIntegral numOffsets) C.get++-- | A version of 'C.isolate' that does not fail the parse because of+-- unconsumed data.+isolate' :: Int -> C.Get a -> C.Get a+isolate' n m = do+ s <- C.getBytes n+ either fail return $ C.runGet m s
+ Kafka/Internal/Transport.hs view
@@ -0,0 +1,88 @@+{-# LANGUAGE NamedFieldPuns #-}+module Kafka.Internal.Transport+ ( Transport(..)+ , recvExactly++ , Socket+ , withConnection+ , connect+ , close+ ) where+++import Control.Applicative+import Data.ByteString (ByteString)+import Network.Socket (Socket, close)++import qualified Control.Exception as E+import qualified Data.ByteString as B+import qualified Data.ByteString.Char8 as B8+import qualified Network.Socket as N+import qualified Network.Socket.ByteString as S++-- | Types that provide a means to send and receive bytes.+class Transport t where+ -- | Send the given ByteString down the transport.+ --+ -- This must block until the request has been finished.+ send :: t -> ByteString -> IO ()+ -- | Read up to the given number of bytes from the stream.+ --+ -- The returned ByteString may be empty if the end of the stream was+ -- reached.+ recv :: t -> Int -> IO ByteString++instance Transport Socket where+ send = S.sendAll+ recv = S.recv++-- | Keep reading from the given transport until the given number of bytes are+-- read or the end of the stream is reached -- whichever comes first.+recvExactly :: Transport t => t -> Int -> IO ByteString+recvExactly t size = B.concat . reverse <$> loop [] 0+ where+ loop chunks bytesRead+ | bytesRead >= size = return chunks+ | otherwise = do+ chunk <- recv t (size - bytesRead)+ if B.null chunk+ then return chunks+ else loop (chunk:chunks) $! bytesRead + B.length chunk+ -- TODO this should return a bytestring of exactly the given size. Anything+ -- else should be an IO exception or a protocol error since we use this only+ -- in cases where we know the exact size of the response.++getAddrInfo :: ByteString -> Int -> IO N.AddrInfo+getAddrInfo host port = head <$>+ N.getAddrInfo (Just hints)+ (Just $ B8.unpack host)+ (Just $ show port)+ where+ hints = N.defaultHints {+ N.addrFamily = N.AF_INET+ , N.addrFlags = [N.AI_NUMERICSERV]+ , N.addrSocketType = N.Stream+ }++-- | Create a new connection.+--+-- Connects to the given hostname and port. Throws an 'N.IOException' in case+-- of failure.+connect :: ByteString -> Int -> IO Socket+connect host port = do+ addrInfo <- getAddrInfo host port+ socket <- N.socket N.AF_INET N.Stream N.defaultProtocol+ N.connect socket (N.addrAddress addrInfo) `E.onException` N.close socket+ return socket++-- | Open a connection, execute the given operation on it, and ensure it is+-- closed afterwards even if an exception was thrown.+--+-- > withConnection "localhost" 9092 $ \conn ->+-- > doStuff conn+-- > fail "something went wrong"+--+-- Throws an 'IOException' if we were unable to open the connection.+--+withConnection :: ByteString -> Int -> (Socket -> IO a) -> IO a+withConnection host port = E.bracket (connect host port) close
+ Kafka/Internal/Types.hs view
@@ -0,0 +1,289 @@+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE OverlappingInstances #-}+module Kafka.Internal.Types+ ( Error(..)+ , Compression(..)+ , OffsetsTime(..)+ , RequestType(..)++ , Topic(..)+ , Offset(..)+ , Partition(..)+ , Size(..)+ , Count(..)++ , Message(..)+ , MessageSet(..)+ ) where++import Control.Applicative+import Data.ByteString (ByteString)+import Data.ByteString.Lazy (fromStrict, toStrict)+import Data.Digest.CRC32 (crc32)+import Data.DList (DList)+import Data.Monoid+import Data.String+import Data.Time (UTCTime)+import Data.Time.Clock.POSIX+import Data.Word++import qualified Codec.Compression.GZip as GZip+import qualified Codec.Compression.Snappy as Snappy+import qualified Data.ByteString as B+import qualified Data.DList as DList+import qualified Data.Serialize as C++-- | Different errors returned by Kafka.+data Error+ = UnknownError -- -1+ -- ^ Unknown error+ | OffsetOutOfRangeError -- 1+ -- ^ Offset requested is invalid or no longer available on the server.+ | InvalidMessageError -- 2+ -- ^ A message failed to match its checksum.+ | WrongPartitionError -- 3+ -- ^ The requested partition doesn't exist.+ | InvalidFetchSizeError -- 4+ -- ^ The maximum size requested for fetching is smaller than the message+ -- being fetched.+ deriving (Show, Read, Eq, Ord)++instance C.Serialize (Maybe Error) where+ -- A note about extensions:+ -- We need FlexibleInstances to allow GHC to let us declare an instance+ -- for the type @Maybe Error@. We also need OverlappingInstances to tell+ -- it to pick this (more specific) instance of Maybe Error over the+ -- generic one provided by cereal.+ get = do+ code <- C.getWord16be+ case code of+ 0 -> return Nothing+ 1 -> return $ Just OffsetOutOfRangeError+ 2 -> return $ Just InvalidMessageError+ 3 -> return $ Just WrongPartitionError+ 4 -> return $ Just InvalidFetchSizeError+ -1 -> return $ Just UnknownError+ _ -> fail $ "Unknown error code: " ++ show code++ put Nothing = C.putWord16be 0+ put (Just OffsetOutOfRangeError) = C.putWord16be 1+ put (Just InvalidMessageError) = C.putWord16be 2+ put (Just WrongPartitionError) = C.putWord16be 3+ put (Just InvalidFetchSizeError) = C.putWord16be 4+ put (Just UnknownError) = C.putWord16be (-1)++-- | Methods of compression supported by Kafka.+data Compression+ = NoCompression+ -- ^ The message is uncompressed.+ | GzipCompression+ -- ^ The message is compressed using @gzip@ compression and may contain+ -- other messages in it.+ | SnappyCompression+ -- ^ The message is compressed using @snappy@ compression and may contain+ -- other messages in it.+ deriving (Show, Read, Eq, Ord)++instance C.Serialize Compression where+ put NoCompression = C.putWord8 0+ put GzipCompression = C.putWord8 1+ put SnappyCompression = C.putWord8 2++ get = C.getWord8 >>= \i -> case i of+ 0 -> return NoCompression+ 1 -> return GzipCompression+ 2 -> return SnappyCompression+ _ -> fail $ "Invalid compression code: " ++ show i++-- | Different times for which offsets may be retrieved using+-- 'Kafka.offsets'.+data OffsetsTime+ = OffsetsLatest+ -- ^ Retrieve the latest offsets+ | OffsetsEarliest+ -- ^ Retrieve the earliest offsets.+ | OffsetsBefore !UTCTime+ -- ^ Retrieve offsets before the given time.+ --+ -- Keep in mind that the response will not contain the precise offset that+ -- occurred around this time. It will return up to the specified count of+ -- offsets in descending, each being the first offset of every segment+ -- file for the specified partition with a modified time less than this+ -- time, and possibly a "high water mark" for the last segment of the+ -- partition (if it was modified before this time) which specifies the+ -- offset at which the next message to that partition will be written.+ deriving (Show, Read, Eq, Ord)++instance C.Serialize OffsetsTime where+ put OffsetsLatest = C.putWord64be (-1)+ put OffsetsEarliest = C.putWord64be (-2)+ put (OffsetsBefore t) =+ C.putWord64be . round . (* 1000) . utcTimeToPOSIXSeconds $ t++ -- Kafka doesn't care about anything beyond milliseconds while UTCTime can+ -- be more precise. We will lose some amount of precision if the same+ -- UTCTime is serialized and then deserialized.+ get = C.getWord64be >>= \i -> case i of+ -1 -> return OffsetsLatest+ -2 -> return OffsetsEarliest+ t -> return . OffsetsBefore .+ posixSecondsToUTCTime . (/ 1000) . realToFrac $ t++-- | The different request types supported by Kafka.+data RequestType+ = ProduceRequestType -- 0+ | FetchRequestType -- 1+ | MultiFetchRequestType -- 2+ | MultiProduceRequestType -- 3+ | OffsetsRequestType -- 4+ deriving (Show, Read, Eq, Ord)++instance C.Serialize RequestType where+ put ProduceRequestType = C.putWord16be 0+ put FetchRequestType = C.putWord16be 1+ put MultiFetchRequestType = C.putWord16be 2+ put MultiProduceRequestType = C.putWord16be 3+ put OffsetsRequestType = C.putWord16be 4++ get = C.getWord16be >>= \i -> case i of+ 0 -> return ProduceRequestType+ 1 -> return FetchRequestType+ 2 -> return MultiFetchRequestType+ 3 -> return MultiProduceRequestType+ 4 -> return OffsetsRequestType+ _ -> fail $ "Unknown request type: " ++ show i++-- | Represents a Kafka topic.+--+-- This is an instance of 'IsString' so a literal string may be used to create+-- a Topic with the @OverloadedStrings@ extension.+newtype Topic = Topic ByteString+ deriving (Show, Read, Eq, Ord, IsString)++instance C.Serialize Topic where+ put (Topic topic) = do+ C.putWord16be (fromIntegral $ B.length topic)+ C.putByteString topic+ get = do+ topicLength <- fromIntegral <$> C.getWord16be+ Topic <$> C.getByteString topicLength++-- | Represents an Offset in Kafka.+--+-- This is an instance of 'Num' so a literal number may be used to create an+-- Offset.+newtype Offset = Offset Word64+ deriving (Show, Read, Eq, Ord, Num)++instance C.Serialize Offset where+ put (Offset o) = C.putWord64be o+ get = Offset <$> C.getWord64be++-- | Represents a Kafka topic partition.+--+-- This is an instance of 'Num' so a literal number may be used to create a+-- Partition.+newtype Partition = Partition Word32+ deriving (Show, Read, Eq, Ord, Num)++instance C.Serialize Partition where+ put (Partition p) = C.putWord32be p+ get = Partition <$> C.getWord32be++-- | Represents a size.+--+-- This is an instance of 'Num' so a literal number may be used to create a+-- Size.+newtype Size = Size Word32+ deriving (Show, Read, Eq, Ord, Num)++instance C.Serialize Size where+ put (Size s) = C.putWord32be s+ get = Size <$> C.getWord32be++-- | Represents a Count.+--+-- This is an instance of 'Num' so a literal number may be used to create a+-- Size.+newtype Count = Count Word32+ deriving (Show, Read, Eq, Ord, Num)++instance C.Serialize Count where+ put (Count s) = C.putWord32be s+ get = Count <$> C.getWord32be++-- | Represents a Message being sent through Kafka.+data Message = Message {+ messageCompression :: !Compression+ -- ^ Compression used for the message.+ --+ -- If this is anything but 'NoCompression', this message contains other+ -- messages in it.+ , messagePayload :: !ByteString+ -- ^ Message payload.+ --+ -- If the message is using any compression scheme, the payload contains+ -- other messages in the same format.+ } deriving (Show, Read, Eq, Ord)++instance C.Serialize Message where+ put (Message compression payload) = do+ C.putWord32be (fromIntegral messageLength)+ C.putWord8 1 -- Magic+ C.put compression+ C.putWord32be (crc32 payload)+ C.putByteString payload+ where+ messageLength = B.length payload+ + 4 -- checksum+ + 1 -- compression+ + 1 -- magic++ get = C.getWord32be >>= \messageLength -> do+ magic <- C.getWord8+ compression <- case magic of+ 0 -> return NoCompression+ 1 -> C.get+ _ -> fail $ "Unknown magic code: " ++ show magic+ checksum <- C.getWord32be+ let remainingLength = fromIntegral $+ messageLength -+ ( 4 -- checksum+ + 1 -- magic+ + fromIntegral magic -- compression (1 byte if magic is 1)+ )+ payload <- C.getByteString remainingLength+ if crc32 payload == checksum+ then return (Message compression payload)+ else fail "Checksum did not match."+ -- TODO: This should probably be InvalidMessageError++-- | Represents a collection of message payloads.+--+-- These are compressed into a single message using Snappy compression when+-- being sent.+newtype MessageSet = MessageSet { fromMessageSet :: [ByteString] }+ deriving (Show, Read, Eq, Monoid)++instance C.Serialize MessageSet where+ put (MessageSet messages) = C.put (Message SnappyCompression payload)+ where+ payload = Snappy.compress . C.runPut $+ mapM_ (C.put . Message NoCompression) messages++ get = MessageSet . DList.toList <$> (C.get >>= readMessages)+ where+ readMessages :: Message -> C.Get (DList ByteString)+ readMessages (Message NoCompression payload) =+ return (DList.singleton payload)+ readMessages (Message compression payload) = do+ messages <- either fail return $+ C.runGet (many C.get) decompressedPayload+ mconcat <$> mapM readMessages messages+ where+ decompressedPayload = decompress payload+ decompress = case compression of+ NoCompression -> id+ SnappyCompression -> Snappy.decompress+ GzipCompression -> toStrict . GZip.decompress . fromStrict
+ LICENSE view
@@ -0,0 +1,20 @@+Copyright (c) 2015 Abhinav Gupta++Permission is hereby granted, free of charge, to any person obtaining+a copy of this software and associated documentation files (the+"Software"), to deal in the Software without restriction, including+without limitation the rights to use, copy, modify, merge, publish,+distribute, sublicense, and/or sell copies of the Software, and to+permit persons to whom the Software is furnished to do so, subject to+the following conditions:++The above copyright notice and this permission notice shall be included+in all copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.+IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY+CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,+TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE+SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ kafka-client.cabal view
@@ -0,0 +1,61 @@+name : kafka-client+version : 0.7.0.0+synopsis : Low-level Haskell client library for Apache Kafka 0.7.+homepage : https://github.com/abhinav/haskell-kafka-client+license : MIT+license-file : LICENSE+author : Abhinav Gupta+maintainer : Abhinav Gupta <mail@abhinavg.net>+category : Network+build-type : Simple+cabal-version : >=1.10+description :+ This package implements a low-level client client library to interface+ with <http://kafka.apache.org/ Apache Kafka> 0.7.+ .+ __Kafka 0.8 and newer are not yet supported.__++library+ exposed-modules : Kafka+ , Kafka.Internal+ , Kafka.Internal.Request+ , Kafka.Internal.Response+ , Kafka.Internal.Transport+ , Kafka.Internal.Types+ ghc-options : -Wall+ build-depends : base >= 4.7 && < 4.8+ , bytestring >= 0.10+ , cereal >= 0.4+ , digest >= 0.0.1+ , dlist >= 0.7+ , network >= 2.4+ , snappy >= 0.2+ , time >= 1.4+ , zlib >= 0.5++ -- Test suite-only dependencies:+ -- Workaround for ghc-mod #339. Uncomment while developing.+ -- , hspec >= 2.0+ -- , hspec-discover >= 2.1+ -- , QuickCheck >= 2.5+ default-language : Haskell2010++test-suite spec+ type : exitcode-stdio-1.0+ ghc-options : -Wall+ hs-source-dirs : test+ main-is : Spec.hs+ build-depends : base+ , bytestring+ , cereal+ , hspec >= 2.0+ , hspec-discover >= 2.1+ , QuickCheck >= 2.5+ , time++ , kafka-client+ default-language : Haskell2010++source-repository head+ type: git+ location: git://github.com/abhinav/haskell-kafka-client.git
+ test/Spec.hs view
@@ -0,0 +1,1 @@+{-# OPTIONS_GHC -F -pgmF hspec-discover #-}