kafka-client 0.7.0.0 → 0.7.0.1
raw patch · 22 files changed
+1325/−773 lines, 22 filesdep +containersdep +processdep +temporarydep ~basedep ~bytestringdep ~cerealPVP ok
version bump matches the API change (PVP)
Dependencies added: containers, process, temporary
Dependency ranges changed: base, bytestring, cereal, time
API changes (from Hackage documentation)
Files
- CHANGES.md +5/−0
- Kafka.hs +0/−163
- Kafka/Internal.hs +0/−11
- Kafka/Internal/Request.hs +0/−153
- Kafka/Internal/Response.hs +0/−61
- Kafka/Internal/Transport.hs +0/−88
- Kafka/Internal/Types.hs +0/−289
- README.md +27/−0
- examples/consoleConsumer.hs +36/−0
- examples/consoleProducer.hs +31/−0
- integration-test/IntegrationTest.hs +30/−0
- integration-test/KafkaSpec.hs +134/−0
- kafka-client.cabal +53/−8
- src/Kafka.hs +163/−0
- src/Kafka/Internal.hs +11/−0
- src/Kafka/Internal/Request.hs +153/−0
- src/Kafka/Internal/Response.hs +61/−0
- src/Kafka/Internal/Transport.hs +88/−0
- src/Kafka/Internal/Types.hs +289/−0
- test-lib/Test/Kafka/Arbitrary.hs +87/−0
- test-lib/Test/Kafka/Managed.hs +96/−0
- test/Kafka/Internal/TypesSpec.hs +61/−0
+ CHANGES.md view
@@ -0,0 +1,5 @@+0.7.0.0+=======++- Initial release.+
− Kafka.hs
@@ -1,163 +0,0 @@-{-# 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
@@ -1,11 +0,0 @@-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
@@ -1,153 +0,0 @@-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
@@ -1,61 +0,0 @@-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
@@ -1,88 +0,0 @@-{-# 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
@@ -1,289 +0,0 @@-{-# 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
+ README.md view
@@ -0,0 +1,27 @@+`kafka-client`+==============++This package provides a low-level Haskell client for [Apache Kafka][] 0.7.+**Kafka 0.8 and newer are not yet supported.**++Package documentation is available on [Hackage][] and [here][].++<dl>+<dt>+Why Kafka 0.7?+</dt>+<dd>+I needed to interact with a system running Kafka 0.7.+</dd>+<dt>+Will 0.8 or newer be supported?+</dt>+<dd>+There are currently no plans to do so. Unless I end up needing to also interact+with a Kafka 0.8 system, or receive a pull request, the answer is no.+</dd>+</dl>++ [Apache Kafka]: http://kafka.apache.org/+ [Hackage]: http://hackage.haskell.org/package/kafka-client+ [here]: http://abhinavg.net/kafka-client/
+ examples/consoleConsumer.hs view
@@ -0,0 +1,36 @@+{-# LANGUAGE OverloadedStrings #-}+module Main (main) where++import Control.Concurrent (threadDelay)+import Control.Monad+import qualified Data.ByteString.Char8 as B8+import System.Environment+import System.Exit++import Kafka++main :: IO ()+main = do+ args <- getArgs+ when (null args) $ do+ putStrLn "USAGE: consoleProducer TOPIC"+ exitFailure++ let topic = Topic (B8.pack $ head args)++ withConnection "localhost" 9092 $ \conn -> do+ startOffset:_ <- offsets conn (Offsets topic 0 OffsetsLatest 10)+ >>= either (fail . show) return+ loop conn (Fetch topic 0 startOffset (1024 * 1024))+ where+ loop conn fetchReq = do+ [FetchResponse{+ fetchMessages = messages+ , fetchNewOffset = newOffset+ }] <- fetch conn [fetchReq] >>= either (fail . show) return++ mapM_ B8.putStrLn messages+ when (null messages) $+ threadDelay (1000 * 1000)++ loop conn fetchReq{fetchOffset = newOffset}
+ examples/consoleProducer.hs view
@@ -0,0 +1,31 @@+{-# LANGUAGE OverloadedStrings #-}+module Main (main) where++import Control.Applicative+import Control.Monad+import qualified Data.ByteString.Char8 as B8+import System.Environment+import System.Exit+import qualified System.IO as IO++import Kafka++main :: IO ()+main = do+ args <- getArgs+ when (null args) $ do+ putStrLn "USAGE: consoleProducer TOPIC"+ exitFailure++ let topic = Topic (B8.pack $ head args)+ withConnection "localhost" 9092 $ \conn ->+ whileM_ (not <$> IO.isEOF) $ do+ line <- B8.getLine+ produce conn [Produce topic 0 [line]]++whileM_ :: Monad m => m Bool -> m a -> m ()+whileM_ cond action = loop+ where+ loop = do+ x <- cond+ when x (action >> loop)
+ integration-test/IntegrationTest.hs view
@@ -0,0 +1,30 @@+{-# LANGUAGE OverloadedStrings #-}+module Main (main) where++import Control.Exception+import Test.Hspec+import Test.Kafka.Managed++import qualified Network as N+import qualified System.IO.Temp as Temp++import qualified Kafka+import qualified KafkaSpec++getRandomUnusedPort :: IO Int+getRandomUnusedPort =+ bracket (N.listenOn $ N.PortNumber 0) N.sClose $ \socket -> do+ (N.PortNumber num) <- N.socketPort socket+ return (fromIntegral num)++main :: IO ()+main = do+ kafkaPort <- getRandomUnusedPort+ Temp.withSystemTempDirectory "kafka-logs" $ \logDir ->+ let kafkaConfig = KafkaConfig {+ kafkaLogDirectory = logDir+ , kafkaServerPort = kafkaPort+ }+ in withManagedKafka kafkaConfig $+ Kafka.withConnection "localhost" kafkaPort $+ hspec . KafkaSpec.spec
+ integration-test/KafkaSpec.hs view
@@ -0,0 +1,134 @@+{-# LANGUAGE OverloadedStrings #-}+module KafkaSpec (spec) where++import Control.Applicative ((<$>))+import Control.Concurrent (threadDelay)+import Control.Monad (replicateM)+import Data.IORef (modifyIORef', newIORef, readIORef)+import Test.Hspec+import Test.Hspec.QuickCheck (prop)+import Test.QuickCheck++import qualified Data.Set as Set++import Kafka+import Test.Kafka.Arbitrary++-- | Keep running the given generator until a value that satisfies the given+-- predicate is produced.+generateSuchThat :: Gen a -> (a -> IO Bool) -> IO a+generateSuchThat gen f = loop+ where+ loop = do+ a <- generate gen+ match <- f a+ if match+ then return a+ else loop++-- | Returns an IO action that will always return unique values from the given+-- generator.+uniqueGenerator :: Ord a => Gen a -> IO (IO a)+uniqueGenerator gen = do+ generated <- newIORef Set.empty+ let isNew a = Set.notMember a <$> readIORef generated+ return $ do+ a <- generateSuchThat gen isNew+ modifyIORef' generated (Set.insert a)+ return a++spec :: Socket -> Spec+spec conn = describe "Kafka" $ do++ -- This is hacky but it will ensure that none of the tests run under the+ -- same Kafka topic.+ getNewTopic <- runIO $ uniqueGenerator arbitrary++ prop "returns zero offset for empty topic" $ do+ topic <- getNewTopic+ let req = Offsets topic 0 OffsetsLatest 10+ offsets conn req `shouldReturn` Right [0]++ prop "can produce and fetch empty payloads" $ do+ topic <- getNewTopic+ produce conn [Produce topic 0 [""]]+ Right resp <- pollMessages conn $ Fetch topic 0 0 (1024 * 1024)+ fetchMessages resp `shouldBe` [""]++ prop "can produce and fetch empty lists of payloads" $ do+ topic <- getNewTopic+ produce conn [Produce topic 0 []]+ fetch conn [fetchRequest topic 0]+ `shouldReturn` Right [FetchResponse [] 0]++ prop "can fetch and produce" $ \(NonEmptyPayloadList messages) -> do+ topic <- getNewTopic+ produce conn [Produce topic 0 messages]++ Right resp <- pollMessages conn (fetchRequest topic 0)+ fetchMessages resp `shouldBe` messages++ let newOffset = fetchNewOffset resp++ offsets conn (Offsets topic 0 OffsetsLatest 10)+ `shouldReturn` Right [newOffset, 0]++ fetch conn [fetchRequest topic newOffset]+ `shouldReturn` Right [FetchResponse [] newOffset]++ prop "can multi-fetch and multi-produce" $ \(NonEmpty ps) -> do+ topics <- replicateM (length ps) getNewTopic+ let messageSets = map getPayloads ps+ produceReqs = zipWith+ (\t ms -> Produce t 0 ms) topics messageSets+ fetchReqs = map (`fetchRequest` 0) topics++ receivedAll (Left _) = False+ receivedAll (Right responses) = all isSuccess responses+ where+ isSuccess = not . null . fetchMessages++ produce conn produceReqs+ responses <- do+ result <- retryUntil receivedAll 10 500 (fetch conn fetchReqs)+ case result of+ Just (Right responses) -> return responses+ Just (Left e) -> fail (show e)+ Nothing -> fail "Failed to fetch messages."++ map fetchMessages responses `shouldBe` messageSets+ where+ fetchRequest topic offset = Fetch topic 0 offset (1024 * 1024)++-- | Polls the given transport with the given Fetch request until a non-empty+-- list of messages is available.+--+-- Throws an error if a non-empty list of messages is not received after+-- trying for 5 seconds.+pollMessages :: Transport t => t -> Fetch -> IO (Response FetchResponse)+pollMessages t req = do+ result <- retryUntil isSuccess 10 500 (fetch t [req])+ case result of+ Nothing -> fail "Failed to fetch messages."+ Just xs -> return $ head <$> xs+ where+ isSuccess (Left _) = False+ isSuccess (Right [resp]) = not . null . fetchMessages $ resp+ isSuccess (Right xs) =+ error $ "Too many messages in response: " ++ show xs++-- | Retries the given IO operation with the specified delay until the given+-- predicate on the result returns True.+--+-- Return Nothing if the maximum number of attempts was exceeded.+retryUntil :: (a -> Bool) -> Int -> Int -> IO a -> IO (Maybe a)+retryUntil predicate maxAttempts delayMillis m = loop 0+ where+ loop attempt+ | attempt >= maxAttempts = return Nothing+ | otherwise = do+ a <- m+ if predicate a+ then return (Just a)+ else threadDelay (delayMillis * 1000) >> loop (attempt + 1)+
kafka-client.cabal view
@@ -1,7 +1,7 @@ name : kafka-client-version : 0.7.0.0+version : 0.7.0.1 synopsis : Low-level Haskell client library for Apache Kafka 0.7.-homepage : https://github.com/abhinav/haskell-kafka-client+homepage : https://github.com/abhinav/kafka-client license : MIT license-file : LICENSE author : Abhinav Gupta@@ -14,6 +14,10 @@ with <http://kafka.apache.org/ Apache Kafka> 0.7. . __Kafka 0.8 and newer are not yet supported.__+extra-source-files:+ README.md+ CHANGES.md+ examples/*.hs library exposed-modules : Kafka@@ -22,8 +26,10 @@ , Kafka.Internal.Response , Kafka.Internal.Transport , Kafka.Internal.Types+ hs-source-dirs : src ghc-options : -Wall- build-depends : base >= 4.7 && < 4.8+ build-depends : base >= 4.7 && < 4.9+ , bytestring >= 0.10 , cereal >= 0.4 , digest >= 0.0.1@@ -35,17 +41,23 @@ -- Test suite-only dependencies: -- Workaround for ghc-mod #339. Uncomment while developing.- -- , hspec >= 2.0- -- , hspec-discover >= 2.1- -- , QuickCheck >= 2.5+ -- , containers+ -- , hspec+ -- , hspec-discover+ -- , process+ -- , temporary+ -- , QuickCheck default-language : Haskell2010 test-suite spec type : exitcode-stdio-1.0 ghc-options : -Wall- hs-source-dirs : test+ hs-source-dirs : test test-lib main-is : Spec.hs+ other-modules : Kafka.Internal.TypesSpec+ , Test.Kafka.Arbitrary build-depends : base+ , bytestring , cereal , hspec >= 2.0@@ -56,6 +68,39 @@ , kafka-client default-language : Haskell2010 +flag integration+ default : False+ description : Enable integration tests.++test-suite integration-test+ if flag(integration)+ buildable : True+ else+ buildable : False+ type : exitcode-stdio-1.0+ ghc-options : -Wall+ hs-source-dirs : integration-test test-lib+ main-is : IntegrationTest.hs+ other-modules : KafkaSpec+ , Test.Kafka.Arbitrary+ , Test.Kafka.Managed+ build-depends : base++ , bytestring+ , cereal+ , containers+ , hspec >= 2.0+ , hspec-discover >= 2.1+ , network >= 2.4+ , process >= 1.2+ , QuickCheck >= 2.5+ , temporary >= 1.2+ , time++ , kafka-client+ default-language : Haskell2010++ source-repository head type: git- location: git://github.com/abhinav/haskell-kafka-client.git+ location: git://github.com/abhinav/kafka-client.git
+ src/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
+ src/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
+ src/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+
+ src/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
+ src/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
+ src/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
+ test-lib/Test/Kafka/Arbitrary.hs view
@@ -0,0 +1,87 @@+{-# OPTIONS_GHC -fno-warn-orphans #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE StandaloneDeriving #-}+module Test.Kafka.Arbitrary+ ( Payload(..)+ , NonEmptyPayloadList(..)+ ) where++import Control.Applicative+import Test.QuickCheck++import qualified Data.ByteString as B+import qualified Data.ByteString.Char8 as B8+import qualified Data.Time as T++import qualified Kafka.Internal.Types as I++instance Arbitrary I.Error where+ arbitrary = elements [+ I.UnknownError+ , I.OffsetOutOfRangeError+ , I.InvalidMessageError+ , I.WrongPartitionError+ , I.InvalidFetchSizeError+ ]++instance Arbitrary I.Compression where+ arbitrary = elements [+ I.NoCompression+ , I.SnappyCompression+ , I.GzipCompression+ ]++instance Arbitrary I.OffsetsTime where+ arbitrary = oneof [+ elements [I.OffsetsLatest, I.OffsetsEarliest]+ , I.OffsetsBefore <$> someTime+ ]+ where+ someTime = T.UTCTime <$> someDay <*> someDayTime+ someDay =+ -- Note: We only support times after Unix epoch. Any dates before+ -- 1970 cannot be represented in Kafka 0.7.+ T.fromGregorian <$> choose (1970, 2400)+ <*> choose (1, 12)+ <*> choose (0, 31)+ someDayTime = T.secondsToDiffTime <$> choose (0, 86400)++instance Arbitrary I.RequestType where+ arbitrary = elements [+ I.ProduceRequestType+ , I.FetchRequestType+ , I.MultiFetchRequestType+ , I.MultiProduceRequestType+ , I.OffsetsRequestType+ ]++instance Arbitrary I.Topic where+ arbitrary = I.Topic . B8.pack <$> safeString+ where+ safeString = listOf1 $ elements ['a'..'z']++deriving instance Arbitrary I.Offset+deriving instance Arbitrary I.Partition+deriving instance Arbitrary I.Size+deriving instance Arbitrary I.Count++instance Arbitrary I.Message where+ arbitrary =+ I.Message <$> arbitrary+ <*> (getPayload <$> arbitrary)++instance Arbitrary I.MessageSet where+ arbitrary = I.MessageSet . map getPayload <$> arbitrary++newtype Payload = Payload { getPayload :: B.ByteString }+ deriving (Show)++instance Arbitrary Payload where+ arbitrary = Payload . B.pack <$> arbitrary++newtype NonEmptyPayloadList = NonEmptyPayloadList+ { getPayloads :: [B.ByteString]+ } deriving (Show)++instance Arbitrary NonEmptyPayloadList where+ arbitrary = NonEmptyPayloadList . map getPayload <$> listOf1 arbitrary
+ test-lib/Test/Kafka/Managed.hs view
@@ -0,0 +1,96 @@+{-# LANGUAGE RecordWildCards #-}+module Test.Kafka.Managed+ ( KafkaConfig(..)+ , withManagedKafka+ ) where++import Control.Applicative+import Control.Concurrent (threadDelay)+import Control.Exception (IOException, bracket, catch)+import System.Exit (ExitCode (..))++import qualified Network.Socket as N+import qualified System.IO as IO+import qualified System.IO.Temp as Temp+import qualified System.Process as P++data KafkaConfig = KafkaConfig {+ kafkaServerPort :: Int+ , kafkaLogDirectory :: FilePath+ } deriving (Show, Eq)++toProperties :: KafkaConfig -> [(String, String)]+toProperties KafkaConfig{..} =+ [ ("brokerid", "0")+ , ("port", show kafkaServerPort)+ , ("log.dir", kafkaLogDirectory)+ , ("num.partitions", "1")+ , ("enable.zookeeper", "false")+ , ("log.flush.interval", "1")+ , ("log.default.flush.interval.ms", "10")+ , ("log.default.flush.scheduler.interval.ms", "5")+ ]++writeProperties :: [(String, String)] -> IO.Handle -> IO ()+writeProperties entries h = IO.hPutStr h . unlines $ do+ (key, value) <- entries+ return $! key ++ "=" ++ value++withManagedKafka :: KafkaConfig -> IO a -> IO a+withManagedKafka config m =+ Temp.withSystemTempFile "kafka.proeprties" $ \configPath handle -> do+ writeProperties (toProperties config) handle+ IO.hClose handle+ bracket (startServer configPath) stopServer . const $ do+ waitUntilConnectable (kafkaServerPort config)+ m++-- | Keep trying to connect to the given port until the server accepts a+-- connection or we've made more than 10 attempts.+waitUntilConnectable :: Int -> IO ()+waitUntilConnectable port = loop (0 :: Int)+ where+ loop attempts+ -- TODO maxAttempts can probably be a parameter+ | attempts >= 10 = error "Can't connect to Kafka after 10 attempts."+ | otherwise = do+ addrInfo <- head <$> N.getAddrInfo+ (Just $ N.defaultHints {+ N.addrFlags = [N.AI_NUMERICSERV]+ , N.addrFamily = N.AF_INET+ , N.addrSocketType = N.Stream+ })+ (Just "localhost") (Just $ show port)+ socket <- N.socket N.AF_INET N.Stream N.defaultProtocol+ let connect = do+ N.connect socket (N.addrAddress addrInfo)+ N.close socket+ retry = do+ threadDelay (200 * 1000)+ loop (attempts + 1)+ connect `catchIO` const retry+ catchIO :: IO a -> (IOException -> IO a) -> IO a+ catchIO = catch++-- TODO Better stdout/stderr handling. Could return a ManagedKafka object here+-- that allows access to the logs in case of failure.+startServer :: FilePath -> IO P.ProcessHandle+startServer configPath =+ IO.withFile "/dev/null" IO.WriteMode $ \nullHandle -> do+ (_, _, _, h) <- P.createProcess (p nullHandle)+ return h+ where+ p nullHandle = (P.proc "kafka-server-start.sh" [configPath]) {+ P.std_out = P.UseHandle nullHandle+ }++stopServer :: P.ProcessHandle -> IO ()+stopServer h = do+ P.interruptProcessGroupOf h+ code <- P.waitForProcess h+ case code of+ ExitSuccess -> return ()+ ExitFailure 130 -> return () -- Exit code 130 == terminated with Ctrl-C+ ExitFailure exitCode -> putStrLn . unwords $+ ["Failed to stop Kafka server. Exit code", show exitCode]+
+ test/Kafka/Internal/TypesSpec.hs view
@@ -0,0 +1,61 @@+module Kafka.Internal.TypesSpec where++import Test.Hspec+import Test.QuickCheck++import qualified Data.Serialize as C++import Test.Kafka.Arbitrary ()++import qualified Kafka.Internal.Types as I++spec :: Spec+spec = do+ describe "Error" $+ it "serializes and deserializes" $+ property (checkSerialization :: Maybe I.Error -> Expectation)++ describe "Compression" $+ it "serializes and deserializes" $+ property (checkSerialization :: I.Compression -> Expectation)++ describe "OffsetsTime" $+ it "serializes and deserializes" $+ property (checkSerialization :: I.OffsetsTime -> Expectation)++ describe "RequestType" $+ it "serializes and deserializes" $+ property (checkSerialization :: I.RequestType -> Expectation)++ describe "Topic" $+ it "serializes and deserializes" $+ property (checkSerialization :: I.Topic -> Expectation)++ describe "Offset" $+ it "serializes and deserializes" $+ property (checkSerialization :: I.Offset -> Expectation)++ describe "Partition" $+ it "serializes and deserializes" $+ property (checkSerialization :: I.Partition -> Expectation)++ describe "Size" $+ it "serializes and deserializes" $+ property (checkSerialization :: I.Size -> Expectation)++ describe "Count" $+ it "serializes and deserializes" $+ property (checkSerialization :: I.Count -> Expectation)++ describe "Message" $+ it "serializes and deserializes" $+ property (checkSerialization :: I.Message -> Expectation)++ describe "MessageSet" $+ it "serializes and deserializes" $+ property (checkSerialization :: I.MessageSet -> Expectation)++-- | Check that the given serializable item serializes correctly in both+-- directions.+checkSerialization :: (C.Serialize a, Show a, Eq a) => a -> Expectation+checkSerialization a = C.decode (C.encode a) `shouldBe` Right a