milena 0.4.0.1 → 0.5.0.0
raw patch · 7 files changed
+189/−157 lines, 7 files
Files
- Network/Kafka.hs +33/−81
- Network/Kafka/Consumer.hs +40/−0
- Network/Kafka/Producer.hs +18/−8
- Network/Kafka/Protocol.hs +58/−57
- README.md +7/−0
- milena.cabal +5/−4
- test/tests.hs +28/−7
Network/Kafka.hs view
@@ -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)
+ Network/Kafka/Consumer.hs view
@@ -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
Network/Kafka/Producer.hs view
@@ -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)))
Network/Kafka/Protocol.hs view
@@ -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
README.md view
@@ -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
milena.cabal view
@@ -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,
test/tests.hs view
@@ -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