diff --git a/Network/Kafka.hs b/Network/Kafka.hs --- a/Network/Kafka.hs +++ b/Network/Kafka.hs @@ -8,19 +8,19 @@ import Control.Exception (IOException) import Control.Exception.Lifted (catch) import Control.Lens -import Control.Monad (liftM) -import Control.Monad.Except (ExceptT(..), runExceptT, withExceptT, MonadError(..)) +import Control.Monad.Except (ExceptT(..), runExceptT, MonadError(..)) import Control.Monad.Trans (liftIO, lift) import Control.Monad.Trans.State +import Control.Monad.State.Class (MonadState) import Data.ByteString.Char8 (ByteString) import Data.List.NonEmpty (NonEmpty(..)) import qualified Data.List.NonEmpty as NE import Data.Monoid ((<>)) import qualified Data.Pool as Pool -import Data.Serialize.Get import System.IO -import qualified Data.ByteString.Char8 as B import qualified Data.Map as M +import Data.Set (Set) +import qualified Data.Set as Set import qualified Network import Prelude @@ -62,8 +62,6 @@ -- | Errors given from the Kafka monad. data KafkaClientError = -- | A response did not contain an offset. KafkaNoOffset - -- | Got a different form of a response than was requested. - | KafkaExpected KafkaExpectedResponse -- | A value could not be deserialized correctly. | KafkaDeserializationError String -- TODO: cereal is Stringly typed, should use tickle -- | Could not find a cached broker for the found leader. @@ -72,12 +70,6 @@ | KafkaIOException IOException deriving (Eq, Show) --- | Type of response to expect, used for 'KafkaExpected' error. -data KafkaExpectedResponse = ExpectedMetadata - | ExpectedFetch - | ExpectedProduce - deriving (Eq, Show) - -- | An abstract form of Kafka's time. Used for querying offsets. data KafkaTime = -- | The latest time on the broker. LatestTime @@ -90,7 +82,7 @@ , _palPartition :: Partition , _palLeader :: Leader } - deriving (Show) + deriving (Show, Eq, Ord) makeLenses ''PartitionAndLeader @@ -162,43 +154,33 @@ runKafka :: KafkaState -> Kafka a -> IO (Either KafkaClientError a) runKafka s k = runExceptT $ evalStateT k s --- | Make a request, incrementing the `_stateCorrelationId`. -makeRequest :: RequestMessage -> Kafka Request -makeRequest m = do - corid <- use stateCorrelationId - stateCorrelationId += 1 - conid <- use stateName - return $ Request (corid, ClientId conid, m) - -- | Catch 'IOException's and wrap them in 'KafkaIOException's. -tryKafkaIO :: IO a -> Kafka a -tryKafkaIO = tryKafka . liftIO - --- | Catch 'IOException's and wrap them in 'KafkaIOException's. tryKafka :: Kafka a -> Kafka a tryKafka = (`catch` \e -> lift . throwError $ KafkaIOException (e :: IOException)) -doRequest :: Handle -> Request -> Kafka Response -doRequest h r = do - rawLength <- tryKafkaIO $ do - B.hPut h $ requestBytes r - hFlush h - B.hGet h 4 - dataLength <- runGetKafka (liftM fromIntegral getWord32be) rawLength - resp <- tryKafkaIO $ B.hGet h dataLength - runGetKafka (getResponse dataLength) resp - -runGetKafka :: Get a -> ByteString -> Kafka a -runGetKafka g bs = lift $ withExceptT KafkaDeserializationError $ ExceptT $ return $ runGet g bs +-- | Make a request, incrementing the `_stateCorrelationId`. +makeRequest :: Handle -> ReqResp (Kafka a) -> Kafka a +makeRequest h reqresp = do + (clientId, correlationId) <- makeIds + eitherResp <- tryKafka $ doRequest clientId correlationId h reqresp + case eitherResp of + Left s -> throwError $ KafkaDeserializationError s + Right r -> return r + where + makeIds :: MonadState KafkaState m => m (ClientId, CorrelationId) + makeIds = do + corid <- use stateCorrelationId + stateCorrelationId += 1 + conid <- use stateName + return (ClientId conid, corid) --- | Send a metadata request +-- | Send a metadata request to any broker. metadata :: MetadataRequest -> Kafka MetadataResponse metadata request = withAnyHandle $ flip metadata' request --- | Send a metadata request +-- | Send a metadata request. metadata' :: Handle -> MetadataRequest -> Kafka MetadataResponse -metadata' h request = - makeRequest (MetadataRequest request) >>= doRequest h >>= expectResponse ExpectedMetadata _MetadataResponse +metadata' h request = makeRequest h $ MetadataRR request getTopicPartitionLeader :: TopicName -> Partition -> Kafka Broker getTopicPartitionLeader t p = do @@ -211,11 +193,11 @@ expect e f = lift . maybe (throwError e) return . f -- | Find a leader and partition for the topic. -brokerPartitionInfo :: TopicName -> Kafka [PartitionAndLeader] +brokerPartitionInfo :: TopicName -> Kafka (Set PartitionAndLeader) brokerPartitionInfo t = do let s = stateTopicMetadata . at t tmd <- findMetadataOrElse [t] s KafkaFailedToFetchMetadata - return $ pal <$> tmd ^. partitionsMetadata + return $ Set.fromList $ pal <$> tmd ^. partitionsMetadata where pal d = PartitionAndLeader t (d ^. partitionId) (d ^. partitionMetadataLeader) findMetadataOrElse :: [TopicName] -> Getting (Maybe a) KafkaState (Maybe a) -> KafkaClientError -> Kafka a @@ -230,45 +212,12 @@ Just x -> return x Nothing -> lift $ throwError err --- | Function to give an error when the response seems wrong. -expectResponse :: KafkaExpectedResponse -> Getting (Leftmost b) ResponseMessage b -> Response -> Kafka b -expectResponse e p = expect (KafkaExpected e) (firstOf $ responseMessage . p) - -- | Convert an abstract time to a serializable protocol value. protocolTime :: KafkaTime -> Time protocolTime LatestTime = Time (-1) protocolTime EarliestTime = Time (-2) protocolTime (OtherTime o) = o --- * Fetching - --- | Default: @-1@ -ordinaryConsumerId :: ReplicaId -ordinaryConsumerId = ReplicaId (-1) - --- | Construct a fetch request from the values in the state. -fetchRequest :: Offset -> Partition -> TopicName -> Kafka FetchRequest -fetchRequest o p topic = do - wt <- use stateWaitTime - ws <- use stateWaitSize - bs <- use stateBufferSize - return $ FetchReq (ordinaryConsumerId, wt, ws, [(topic, [(p, o, bs)])]) - --- | Execute a fetch request and get the raw fetch response. -fetch' :: Handle -> FetchRequest -> Kafka FetchResponse -fetch' h request = - makeRequest (FetchRequest request) >>= doRequest h >>= expectResponse ExpectedFetch _FetchResponse - --- | Execute a fetch request and get the raw fetch response. Round-robins the --- requests to addresses in the 'KafkaState'. -fetch :: FetchRequest -> Kafka FetchResponse -fetch request = withAnyHandle $ flip fetch' request - --- | Extract out messages with their topics from a fetch response. -fetchMessages :: FetchResponse -> [TopicAndMessage] -fetchMessages fr = (fr ^.. fetchResponseFields . folded) >>= tam - where tam a = TopicAndMessage (a ^. _1) <$> a ^.. _2 . folded . _4 . messageSetMembers . folded . setMessage - updateMetadatas :: [TopicName] -> Kafka () updateMetadatas ts = do md <- metadata $ MetadataReq ts @@ -316,7 +265,7 @@ let foundPool = conns ^. at address pool <- case foundPool of Nothing -> do - newPool <- tryKafkaIO $ mkPool address + newPool <- tryKafka $ liftIO $ mkPool address stateConnections .= (at address ?~ newPool $ conns) return newPool Just p -> return p @@ -365,12 +314,15 @@ -- | Get the first found offset. getLastOffset' :: Handle -> KafkaTime -> Partition -> TopicName -> Kafka Offset -getLastOffset' h m p t = - makeRequest (offsetRequest [(TopicAndPartition t p, PartitionOffsetRequestInfo m 1)]) >>= doRequest h >>= maybe (StateT . const $ throwError KafkaNoOffset) return . firstOf (responseMessage . _OffsetResponse . offsetResponseOffset p) +getLastOffset' h m p t = do + let offsetRR = OffsetRR $ offsetRequest [(TopicAndPartition t p, PartitionOffsetRequestInfo m 1)] + offsetResponse <- makeRequest h offsetRR + let maybeResp = firstOf (offsetResponseOffset p) offsetResponse + maybe (throwError KafkaNoOffset) return maybeResp -- | Create an offset request. -offsetRequest :: [(TopicAndPartition, PartitionOffsetRequestInfo)] -> RequestMessage +offsetRequest :: [(TopicAndPartition, PartitionOffsetRequestInfo)] -> OffsetRequest offsetRequest ts = - OffsetRequest $ OffsetReq (ReplicaId (-1), M.toList . M.unionsWith (<>) $ fmap f ts) + OffsetReq (ReplicaId (-1), M.toList . M.unionsWith (<>) $ fmap f ts) where f (TopicAndPartition t p, i) = M.singleton t [g p i] g p (PartitionOffsetRequestInfo kt mno) = (p, protocolTime kt, mno) diff --git a/Network/Kafka/Consumer.hs b/Network/Kafka/Consumer.hs new file mode 100644 --- /dev/null +++ b/Network/Kafka/Consumer.hs @@ -0,0 +1,40 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE FlexibleContexts #-} + +module Network.Kafka.Consumer where + +import Control.Applicative +import Control.Lens +import System.IO +import Prelude + +import Network.Kafka +import Network.Kafka.Protocol + +-- * Fetching + +-- | Default: @-1@ +ordinaryConsumerId :: ReplicaId +ordinaryConsumerId = ReplicaId (-1) + +-- | Construct a fetch request from the values in the state. +fetchRequest :: Offset -> Partition -> TopicName -> Kafka FetchRequest +fetchRequest o p topic = do + wt <- use stateWaitTime + ws <- use stateWaitSize + bs <- use stateBufferSize + return $ FetchReq (ordinaryConsumerId, wt, ws, [(topic, [(p, o, bs)])]) + +-- | Execute a fetch request and get the raw fetch response. +fetch' :: Handle -> FetchRequest -> Kafka FetchResponse +fetch' h request = makeRequest h $ FetchRR request + +fetch :: Offset -> Partition -> TopicName -> Kafka FetchResponse +fetch o p topic = do + broker <- getTopicPartitionLeader topic p + withBrokerHandle broker (\handle -> fetch' handle =<< fetchRequest o p topic) + +-- | Extract out messages with their topics from a fetch response. +fetchMessages :: FetchResponse -> [TopicAndMessage] +fetchMessages fr = (fr ^.. fetchResponseFields . folded) >>= tam + where tam a = TopicAndMessage (a ^. _1) <$> a ^.. _2 . folded . _4 . messageSetMembers . folded . setMessage diff --git a/Network/Kafka/Producer.hs b/Network/Kafka/Producer.hs --- a/Network/Kafka/Producer.hs +++ b/Network/Kafka/Producer.hs @@ -1,11 +1,14 @@ module Network.Kafka.Producer where +import Data.Bits ((.&.)) +import Data.ByteString.Char8 (ByteString) +import qualified Data.Digest.Murmur32 as Murmur32 import Control.Applicative import Control.Lens import Control.Monad.Trans (liftIO) -import Data.ByteString.Char8 (ByteString) -import qualified Data.Digest.Murmur32 as Murmur32 import Data.Monoid ((<>)) +import Data.Set (Set) +import qualified Data.Set as Set import System.IO import qualified Data.Map as M import System.Random (getStdRandom, randomR) @@ -19,8 +22,7 @@ -- | Execute a produce request and get the raw preduce response. produce :: Handle -> ProduceRequest -> Kafka ProduceResponse -produce handle request = - makeRequest (ProduceRequest request) >>= doRequest handle >>= expectResponse ExpectedProduce _ProduceResponse +produce handle request = makeRequest handle $ ProduceRR request -- | Construct a produce request with explicit arguments. produceRequest :: RequiredAcks -> Timeout -> [(TopicAndPartition, MessageSet)] -> ProduceRequest @@ -57,9 +59,13 @@ -- | Compute the partition for a record. This matches the way the official -- | clients compute the partition. -getPartitionByKey :: ByteString -> [PartitionAndLeader] -> Maybe PartitionAndLeader -getPartitionByKey key ps = let i = Murmur32.asWord32 $ Murmur32.hash32WithSeed 0x9747b28c key - in ps ^? ix (fromIntegral i) +getPartitionByKey :: ByteString -> Set PartitionAndLeader -> Maybe PartitionAndLeader +getPartitionByKey key ps = Set.toAscList ps ^? ix i + where murmur = Murmur32.asWord32 . Murmur32.hash32WithSeed 0x9747b28c + toPositive = (.&. 0x7fffffff) + numPartitions = length ps + x = fromIntegral $ toPositive $ murmur key + i = x `mod` numPartitions -- | Execute a produce request using the values in the state. send :: Leader -> [(TopicAndPartition, MessageSet)] -> Kafka ProduceResponse @@ -71,7 +77,7 @@ requestTimeout <- use stateRequestTimeout withBrokerHandle broker $ \handle -> produce handle $ produceRequest requiredAcks requestTimeout ts -getRandPartition :: [PartitionAndLeader] -> Kafka (Maybe PartitionAndLeader) +getRandPartition :: Set PartitionAndLeader -> Kafka (Maybe PartitionAndLeader) getRandPartition ps = liftIO $ (ps' ^?) . element <$> getStdRandom (randomR (0, length ps' - 1)) where ps' = ps ^.. folded . filtered (has $ palLeader . leaderId . _Just) @@ -97,3 +103,7 @@ -- | Construct a message from a string of bytes using default attributes. makeMessage :: ByteString -> Message makeMessage m = Message (defaultMessageCrc, defaultMessageMagicByte, defaultMessageAttributes, defaultMessageKey, Value (Just (KBytes m))) + +-- | Construct a message from a string of bytes using default attributes. +makeKeyedMessage :: ByteString -> ByteString -> Message +makeKeyedMessage k m = Message (defaultMessageCrc, defaultMessageMagicByte, defaultMessageAttributes, Key (Just (KBytes k)), Value (Just (KBytes m))) diff --git a/Network/Kafka/Protocol.hs b/Network/Kafka/Protocol.hs --- a/Network/Kafka/Protocol.hs +++ b/Network/Kafka/Protocol.hs @@ -1,14 +1,18 @@ {-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE GADTs #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE TemplateHaskell #-} {-# LANGUAGE Rank2Types #-} -module Network.Kafka.Protocol where +module Network.Kafka.Protocol + ( module Network.Kafka.Protocol + ) where import Control.Applicative import Control.Category (Category(..)) import Control.Lens -import Control.Monad (replicateM, liftM, liftM2, liftM3, liftM4, liftM5) +import Control.Monad (replicateM, liftM, liftM2, liftM3, liftM4, liftM5, unless) +import Control.Monad.IO.Class (MonadIO, liftIO) import Data.ByteString.Char8 (ByteString) import Data.ByteString.Lens (unpackedChars) import Data.Digest.CRC32 @@ -16,51 +20,50 @@ import Data.Serialize.Get import Data.Serialize.Put import GHC.Exts (IsString(..)) +import System.IO import Numeric.Lens import Prelude hiding ((.), id) import qualified Data.ByteString.Char8 as B import qualified Network +data ReqResp a where + MetadataRR :: MonadIO m => MetadataRequest -> ReqResp (m MetadataResponse) + ProduceRR :: MonadIO m => ProduceRequest -> ReqResp (m ProduceResponse) + FetchRR :: MonadIO m => FetchRequest -> ReqResp (m FetchResponse) + OffsetRR :: MonadIO m => OffsetRequest -> ReqResp (m OffsetResponse) + +doRequest' :: (Deserializable a, MonadIO m) => CorrelationId -> Handle -> Request -> m (Either String a) +doRequest' correlationId h r = do + rawLength <- liftIO $ do + B.hPut h $ requestBytes r + hFlush h + B.hGet h 4 + case runGet (liftM fromIntegral getWord32be) rawLength of + Left s -> return $ Left s + Right dataLength -> do + responseBytes <- liftIO $ B.hGet h dataLength + return $ flip runGet responseBytes $ do + correlationId' <- deserialize + unless (correlationId == correlationId') $ fail ("Expected " ++ show correlationId ++ " but got " ++ show correlationId') + isolate (dataLength - 4) deserialize + +doRequest :: MonadIO m => ClientId -> CorrelationId -> Handle -> ReqResp (m a) -> m (Either String a) +doRequest clientId correlationId h (MetadataRR req) = doRequest' correlationId h $ Request (correlationId, clientId, MetadataRequest req) +doRequest clientId correlationId h (ProduceRR req) = doRequest' correlationId h $ Request (correlationId, clientId, ProduceRequest req) +doRequest clientId correlationId h (FetchRR req) = doRequest' correlationId h $ Request (correlationId, clientId, FetchRequest req) +doRequest clientId correlationId h (OffsetRR req) = doRequest' correlationId h $ Request (correlationId, clientId, OffsetRequest req) + class Serializable a where serialize :: a -> Put class Deserializable a where deserialize :: Get a -data Response = Response { _responseCorrelationId :: CorrelationId, _responseMessage :: ResponseMessage } deriving (Show, Eq) - -getResponse :: Int -> Get Response -getResponse l = Response <$> deserialize <*> getResponseMessage (l - 4) - -newtype ConsumerMetadataResponse = ConsumerMetadataResp (KafkaError, Broker) deriving (Show, Eq, Deserializable) -- CoordinatorId CoordinatorHost CoordinatorPort --- newtype ErrorCode = ErrorCode int16 --- newtype CoordinatorId = CoordinatorId int32 --- newtype CoordinatorHost = CoordinatorHost string --- newtype CoordinatorPort = CoordinatorPort int32 - -getResponseMessage :: Int -> Get ResponseMessage -getResponseMessage l = liftM MetadataResponse (isolate l deserialize) - <|> liftM OffsetResponse (isolate l deserialize) - <|> liftM ProduceResponse (isolate l deserialize) - <|> liftM OffsetCommitResponse (isolate l deserialize) - <|> liftM OffsetFetchResponse (isolate l deserialize) - <|> liftM ConsumerMetadataResponse (isolate l deserialize) - -- MUST try FetchResponse last! - -- - -- As an optimization, Kafka might return a partial message - -- at the end of a MessageSet, so this will consume the rest - -- of the message at the end of the input. - -- - -- Strictly speaking, this might not actually be necessary. - -- Parsing a MessageSet is isolated to the byte count that's - -- at the beginning of a MessageSet. I don't want to spend - -- the time right now to prove that will always be safe, but - -- I'd like to at some point. - <|> liftM FetchResponse (isolate l deserialize) +newtype GroupCoordinatorResponse = GroupCoordinatorResp (KafkaError, Broker) deriving (Show, Eq, Deserializable) -newtype ApiKey = ApiKey Int16 deriving (Show, Eq, Deserializable, Serializable, Num) -- numeric ID for API (i.e. metadata req, produce req, etc.) -newtype ApiVersion = ApiVersion Int16 deriving (Show, Eq, Deserializable, Serializable, Num) -newtype CorrelationId = CorrelationId Int32 deriving (Show, Eq, Deserializable, Serializable, Num, Enum) +newtype ApiKey = ApiKey Int16 deriving (Show, Eq, Deserializable, Serializable, Num, Integral, Ord, Real, Enum) -- numeric ID for API (i.e. metadata req, produce req, etc.) +newtype ApiVersion = ApiVersion Int16 deriving (Show, Eq, Deserializable, Serializable, Num, Integral, Ord, Real, Enum) +newtype CorrelationId = CorrelationId Int32 deriving (Show, Eq, Deserializable, Serializable, Num, Integral, Ord, Real, Enum) newtype ClientId = ClientId KafkaString deriving (Show, Eq, Deserializable, Serializable, IsString) data RequestMessage = MetadataRequest MetadataRequest @@ -69,7 +72,7 @@ | OffsetRequest OffsetRequest | OffsetCommitRequest OffsetCommitRequest | OffsetFetchRequest OffsetFetchRequest - | ConsumerMetadataRequest ConsumerMetadataRequest + | GroupCoordinatorRequest GroupCoordinatorRequest deriving (Show, Eq) newtype MetadataRequest = MetadataReq [TopicName] deriving (Show, Eq, Serializable, Deserializable) @@ -96,9 +99,9 @@ newtype MetadataResponse = MetadataResp { _metadataResponseFields :: ([Broker], [TopicMetadata]) } deriving (Show, Eq, Deserializable) newtype Broker = Broker { _brokerFields :: (NodeId, Host, Port) } deriving (Show, Eq, Ord, Deserializable) -newtype NodeId = NodeId { _nodeId :: Int32 } deriving (Show, Eq, Ord, Deserializable, Num) +newtype NodeId = NodeId { _nodeId :: Int32 } deriving (Show, Eq, Deserializable, Num, Integral, Ord, Real, Enum) newtype Host = Host { _hostKString :: KafkaString } deriving (Show, Eq, Ord, Deserializable, IsString) -newtype Port = Port { _portInt :: Int32 } deriving (Show, Eq, Ord, Deserializable, Num) +newtype Port = Port { _portInt :: Int32 } deriving (Show, Eq, Deserializable, Num, Integral, Ord, Real, Enum) newtype TopicMetadata = TopicMetadata { _topicMetadataFields :: (KafkaError, TopicName, [PartitionMetadata]) } deriving (Show, Eq, Deserializable) newtype PartitionMetadata = PartitionMetadata { _partitionMetadataFields :: (KafkaError, Partition, Leader, Replicas, Isr) } deriving (Show, Eq, Deserializable) newtype Leader = Leader { _leaderId :: Maybe Int32 } deriving (Show, Eq, Ord) @@ -110,18 +113,18 @@ newtype OffsetFetchResponse = OffsetFetchResp [(TopicName, [(Partition, Offset, Metadata, KafkaError)])] deriving (Show, Eq, Deserializable) newtype OffsetRequest = OffsetReq (ReplicaId, [(TopicName, [(Partition, Time, MaxNumberOfOffsets)])]) deriving (Show, Eq, Serializable) -newtype Time = Time { _timeInt :: Int64 } deriving (Show, Eq, Serializable, Num, Bounded) -newtype MaxNumberOfOffsets = MaxNumberOfOffsets Int32 deriving (Show, Eq, Serializable, Num) +newtype Time = Time { _timeInt :: Int64 } deriving (Show, Eq, Serializable, Num, Integral, Ord, Real, Enum, Bounded) +newtype MaxNumberOfOffsets = MaxNumberOfOffsets Int32 deriving (Show, Eq, Serializable, Num, Integral, Ord, Real, Enum) newtype FetchRequest = FetchReq (ReplicaId, MaxWaitTime, MinBytes, [(TopicName, [(Partition, Offset, MaxBytes)])]) deriving (Show, Eq, Deserializable, Serializable) -newtype ReplicaId = ReplicaId Int32 deriving (Show, Eq, Num, Serializable, Deserializable) -newtype MaxWaitTime = MaxWaitTime Int32 deriving (Show, Eq, Num, Serializable, Deserializable) -newtype MinBytes = MinBytes Int32 deriving (Show, Eq, Num, Serializable, Deserializable) -newtype MaxBytes = MaxBytes Int32 deriving (Show, Eq, Num, Serializable, Deserializable) +newtype ReplicaId = ReplicaId Int32 deriving (Show, Eq, Num, Integral, Ord, Real, Enum, Serializable, Deserializable) +newtype MaxWaitTime = MaxWaitTime Int32 deriving (Show, Eq, Num, Integral, Ord, Real, Enum, Serializable, Deserializable) +newtype MinBytes = MinBytes Int32 deriving (Show, Eq, Num, Integral, Ord, Real, Enum, Serializable, Deserializable) +newtype MaxBytes = MaxBytes Int32 deriving (Show, Eq, Num, Integral, Ord, Real, Enum, Serializable, Deserializable) newtype ProduceRequest = ProduceReq (RequiredAcks, Timeout, @@ -129,26 +132,26 @@ deriving (Show, Eq, Serializable) newtype RequiredAcks = - RequiredAcks Int16 deriving (Show, Eq, Serializable, Deserializable, Num) + RequiredAcks Int16 deriving (Show, Eq, Serializable, Deserializable, Num, Integral, Ord, Real, Enum) newtype Timeout = - Timeout Int32 deriving (Show, Eq, Serializable, Deserializable, Num) + Timeout Int32 deriving (Show, Eq, Serializable, Deserializable, Num, Integral, Ord, Real, Enum) newtype Partition = - Partition Int32 deriving (Show, Ord, Eq, Serializable, Deserializable, Num) + Partition Int32 deriving (Show, Eq, Serializable, Deserializable, Num, Integral, Ord, Real, Enum) newtype MessageSet = MessageSet { _messageSetMembers :: [MessageSetMember] } deriving (Show, Eq) data MessageSetMember = MessageSetMember { _setOffset :: Offset, _setMessage :: Message } deriving (Show, Eq) -newtype Offset = Offset Int64 deriving (Show, Eq, Serializable, Deserializable, Num) +newtype Offset = Offset Int64 deriving (Show, Eq, Serializable, Deserializable, Num, Integral, Ord, Real, Enum) newtype Message = Message { _messageFields :: (Crc, MagicByte, Attributes, Key, Value) } deriving (Show, Eq, Deserializable) -newtype Crc = Crc Int32 deriving (Show, Eq, Serializable, Deserializable, Num) -newtype MagicByte = MagicByte Int8 deriving (Show, Eq, Serializable, Deserializable, Num) -newtype Attributes = Attributes Int8 deriving (Show, Eq, Serializable, Deserializable, Num) +newtype Crc = Crc Int32 deriving (Show, Eq, Serializable, Deserializable, Num, Integral, Ord, Real, Enum) +newtype MagicByte = MagicByte Int8 deriving (Show, Eq, Serializable, Deserializable, Num, Integral, Ord, Real, Enum) +newtype Attributes = Attributes Int8 deriving (Show, Eq, Serializable, Deserializable, Num, Integral, Ord, Real, Enum) newtype Key = Key { _keyBytes :: Maybe KafkaBytes } deriving (Show, Eq) newtype Value = Value { _valueBytes :: Maybe KafkaBytes } deriving (Show, Eq) @@ -159,10 +162,10 @@ | OffsetResponse OffsetResponse | OffsetCommitResponse OffsetCommitResponse | OffsetFetchResponse OffsetFetchResponse - | ConsumerMetadataResponse ConsumerMetadataResponse + | GroupCoordinatorResponse GroupCoordinatorResponse deriving (Show, Eq) -newtype ConsumerMetadataRequest = ConsumerMetadataReq ConsumerGroup deriving (Show, Eq, Serializable) +newtype GroupCoordinatorRequest = GroupCoordinatorReq ConsumerGroup deriving (Show, Eq, Serializable) newtype OffsetCommitRequest = OffsetCommitReq (ConsumerGroup, [(TopicName, [(Partition, Offset, Time, Metadata)])]) deriving (Show, Eq, Serializable) newtype OffsetFetchRequest = OffsetFetchReq (ConsumerGroup, [(TopicName, [Partition])]) deriving (Show, Eq, Serializable) @@ -259,7 +262,7 @@ apiKey (MetadataRequest{}) = ApiKey 3 apiKey (OffsetCommitRequest{}) = ApiKey 8 apiKey (OffsetFetchRequest{}) = ApiKey 9 -apiKey (ConsumerMetadataRequest{}) = ApiKey 10 +apiKey (GroupCoordinatorRequest{}) = ApiKey 10 instance Serializable RequestMessage where serialize (ProduceRequest r) = serialize r @@ -268,7 +271,7 @@ serialize (MetadataRequest r) = serialize r serialize (OffsetCommitRequest r) = serialize r serialize (OffsetFetchRequest r) = serialize r - serialize (ConsumerMetadataRequest r) = serialize r + serialize (GroupCoordinatorRequest r) = serialize r instance Serializable Int64 where serialize = putWord64be . fromIntegral instance Serializable Int32 where serialize = putWord32be . fromIntegral @@ -405,8 +408,6 @@ instance Deserializable Int8 where deserialize = liftM fromIntegral getWord8 -- * Generated lenses - -makeLenses ''Response makeLenses ''TopicName diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -6,3 +6,10 @@ Supports Kafka 0.8.x The protocol module is stable (the only changes will be to support changes in the Kafka protocol). The API is functional but subject to change. + +#### Limitations + +Some of the known limitations. + +* Does not yet support compression +* Does not yet support consumer groups diff --git a/milena.cabal b/milena.cabal --- a/milena.cabal +++ b/milena.cabal @@ -4,7 +4,7 @@ -- PVP summary: +-+------- breaking API changes -- | | +----- non-breaking API additions -- | | | +--- code changes with no API change -version: 0.4.0.1 +version: 0.5.0.0 synopsis: A Kafka client for Haskell. description: A Kafka client for Haskell. @@ -27,12 +27,13 @@ source-repository this type: git location: https://github.com/tylerholien/milena.git - tag: 0.4.0.1 + tag: 0.5.0.0 library default-language: Haskell2010 - ghc-options: -Wall + ghc-options: -Wall -fwarn-incomplete-uni-patterns exposed-modules: Network.Kafka, + Network.Kafka.Consumer, Network.Kafka.Protocol, Network.Kafka.Producer build-depends: base >=4.7 && <5, @@ -52,7 +53,7 @@ test-suite test default-language: Haskell2010 - ghc-options: -Wall -threaded + ghc-options: -Wall -threaded -fwarn-incomplete-uni-patterns build-depends: base, milena, mtl, diff --git a/test/tests.hs b/test/tests.hs --- a/test/tests.hs +++ b/test/tests.hs @@ -9,8 +9,9 @@ import Control.Monad.Except (catchError, throwError) import Control.Monad.Trans (liftIO) import Network.Kafka +import Network.Kafka.Consumer import Network.Kafka.Producer -import Network.Kafka.Protocol (Leader(..)) +import Network.Kafka.Protocol (ProduceResponse(..), KafkaError(..)) import Test.Tasty import Test.Tasty.Hspec import Test.Tasty.QuickCheck @@ -25,6 +26,10 @@ specs = do let topic = "milena-test" run = runKafka $ mkKafkaState "milena-test-client" ("localhost", 9092) + requireAllAcks = do + stateRequiredAcks .= -1 + stateWaitSize .= 1 + stateWaitTime .= 1000 byteMessages = fmap (TopicAndMessage topic . makeMessage . B.pack) describe "can talk to local Kafka server" $ do @@ -42,17 +47,33 @@ prop "can fetch messages" $ do result <- run $ do offset <- getLastOffset EarliestTime 0 topic - fetch =<< fetchRequest offset 0 topic + withAnyHandle (\handle -> fetch' handle =<< fetchRequest offset 0 topic) result `shouldSatisfy` isRight - prop "can roundtrip messages" $ \ms -> do + prop "can roundtrip messages" $ \ms key -> do let messages = byteMessages ms result <- run $ do + requireAllAcks info <- brokerPartitionInfo topic - leader <- maybe (Leader Nothing) _palLeader <$> getRandPartition info - offset <- getLastOffset LatestTime 0 topic - void $ send leader [(TopicAndPartition topic 0, groupMessagesToSet messages)] - fmap tamPayload . fetchMessages <$> (fetch =<< fetchRequest offset 0 topic) + let Just PartitionAndLeader { _palLeader = leader, _palPartition = partition } = getPartitionByKey (B.pack key) info + payload = [(TopicAndPartition topic partition, groupMessagesToSet messages)] + s = stateBrokers . at leader + [(_topicName, [(_, NoError, offset)])] <- _produceResponseFields <$> send leader payload + broker <- findMetadataOrElse [topic] s (KafkaInvalidBroker leader) + resp <- withBrokerHandle broker (\handle -> fetch' handle =<< fetchRequest offset partition topic) + return $ fmap tamPayload . fetchMessages $ resp + result `shouldBe` Right (tamPayload <$> messages) + + prop "can roundtrip keyed messages" $ \(NonEmpty ms) key -> do + let keyBytes = B.pack key + messages = fmap (TopicAndMessage topic . makeKeyedMessage keyBytes . B.pack) ms + result <- run $ do + requireAllAcks + produceResps <- produceMessages messages + let [[(_topicName, [(partition, NoError, offset)])]] = map _produceResponseFields produceResps + + resp <- fetch offset partition topic + return $ fmap tamPayload . fetchMessages $ resp result `shouldBe` Right (tamPayload <$> messages) describe "withAddressHandle" $ do