diff --git a/Kafka.hs b/Kafka.hs new file mode 100644 --- /dev/null +++ b/Kafka.hs @@ -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 diff --git a/Kafka/Internal.hs b/Kafka/Internal.hs new file mode 100644 --- /dev/null +++ b/Kafka/Internal.hs @@ -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 diff --git a/Kafka/Internal/Request.hs b/Kafka/Internal/Request.hs new file mode 100644 --- /dev/null +++ b/Kafka/Internal/Request.hs @@ -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 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 + diff --git a/Kafka/Internal/Response.hs b/Kafka/Internal/Response.hs new file mode 100644 --- /dev/null +++ b/Kafka/Internal/Response.hs @@ -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 diff --git a/Kafka/Internal/Transport.hs b/Kafka/Internal/Transport.hs new file mode 100644 --- /dev/null +++ b/Kafka/Internal/Transport.hs @@ -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 diff --git a/Kafka/Internal/Types.hs b/Kafka/Internal/Types.hs new file mode 100644 --- /dev/null +++ b/Kafka/Internal/Types.hs @@ -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 diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/kafka-client.cabal b/kafka-client.cabal new file mode 100644 --- /dev/null +++ b/kafka-client.cabal @@ -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 +category : Network +build-type : Simple +cabal-version : >=1.10 +description : + This package implements a low-level client client library to interface + with 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 diff --git a/test/Spec.hs b/test/Spec.hs new file mode 100644 --- /dev/null +++ b/test/Spec.hs @@ -0,0 +1,1 @@ +{-# OPTIONS_GHC -F -pgmF hspec-discover #-}