diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,8 @@ +0.5.1.0 + +* Add GZip compression support (thanks to Pavel Kučera @pavelkucera) + +0.5.0.2 + +* New maintainer +* Bump dependency bounds diff --git a/Network/Kafka.hs b/Network/Kafka.hs --- a/Network/Kafka.hs +++ b/Network/Kafka.hs @@ -1,3 +1,4 @@ +{-# LANGUAGE ConstraintKinds #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE TemplateHaskell #-} {-# LANGUAGE FlexibleContexts #-} @@ -5,11 +6,12 @@ module Network.Kafka where import Control.Applicative -import Control.Exception (IOException) +import Control.Exception (Exception, IOException) import Control.Exception.Lifted (catch) import Control.Lens +import Control.Monad.Trans.Control (MonadBaseControl) +import Control.Monad.IO.Class (MonadIO, liftIO) 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) @@ -55,7 +57,7 @@ makeLenses ''KafkaState -- | The core Kafka monad. -type Kafka = StateT KafkaState (ExceptT KafkaClientError IO) +type Kafka m = (MonadState KafkaState m, MonadError KafkaClientError m, MonadIO m, MonadBaseControl IO m) type KafkaClientId = KafkaString @@ -70,6 +72,8 @@ | KafkaIOException IOException deriving (Eq, Show) +instance Exception KafkaClientError + -- | An abstract form of Kafka's time. Used for querying offsets. data KafkaTime = -- | The latest time on the broker. LatestTime @@ -151,15 +155,15 @@ (.:) = (.).(.) -- | Run the underlying Kafka monad. -runKafka :: KafkaState -> Kafka a -> IO (Either KafkaClientError a) +runKafka :: KafkaState -> StateT KafkaState (ExceptT KafkaClientError IO) a -> IO (Either KafkaClientError a) runKafka s k = runExceptT $ evalStateT k s -- | Catch 'IOException's and wrap them in 'KafkaIOException's. -tryKafka :: Kafka a -> Kafka a -tryKafka = (`catch` \e -> lift . throwError $ KafkaIOException (e :: IOException)) +tryKafka :: Kafka m => m a -> m a +tryKafka = (`catch` \e -> throwError $ KafkaIOException (e :: IOException)) -- | Make a request, incrementing the `_stateCorrelationId`. -makeRequest :: Handle -> ReqResp (Kafka a) -> Kafka a +makeRequest :: Kafka m => Handle -> ReqResp (m a) -> m a makeRequest h reqresp = do (clientId, correlationId) <- makeIds eitherResp <- tryKafka $ doRequest clientId correlationId h reqresp @@ -175,32 +179,32 @@ return (ClientId conid, corid) -- | Send a metadata request to any broker. -metadata :: MetadataRequest -> Kafka MetadataResponse +metadata :: Kafka m => MetadataRequest -> m MetadataResponse metadata request = withAnyHandle $ flip metadata' request -- | Send a metadata request. -metadata' :: Handle -> MetadataRequest -> Kafka MetadataResponse +metadata' :: Kafka m => Handle -> MetadataRequest -> m MetadataResponse metadata' h request = makeRequest h $ MetadataRR request -getTopicPartitionLeader :: TopicName -> Partition -> Kafka Broker +getTopicPartitionLeader :: Kafka m => TopicName -> Partition -> m Broker getTopicPartitionLeader t p = do let s = stateTopicMetadata . at t tmd <- findMetadataOrElse [t] s KafkaFailedToFetchMetadata leader <- expect KafkaFailedToFetchMetadata (firstOf $ findPartitionMetadata t . (folded . findPartition p) . partitionMetadataLeader) tmd use stateBrokers >>= expect (KafkaInvalidBroker leader) (view $ at leader) -expect :: KafkaClientError -> (a -> Maybe b) -> a -> Kafka b -expect e f = lift . maybe (throwError e) return . f +expect :: Kafka m => KafkaClientError -> (a -> Maybe b) -> a -> m b +expect e f = maybe (throwError e) return . f -- | Find a leader and partition for the topic. -brokerPartitionInfo :: TopicName -> Kafka (Set PartitionAndLeader) +brokerPartitionInfo :: Kafka m => TopicName -> m (Set PartitionAndLeader) brokerPartitionInfo t = do let s = stateTopicMetadata . at t tmd <- findMetadataOrElse [t] s KafkaFailedToFetchMetadata 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 +findMetadataOrElse :: Kafka m => [TopicName] -> Getting (Maybe a) KafkaState (Maybe a) -> KafkaClientError -> m a findMetadataOrElse ts s err = do maybeFound <- use s case maybeFound of @@ -210,7 +214,7 @@ maybeFound' <- use s case maybeFound' of Just x -> return x - Nothing -> lift $ throwError err + Nothing -> throwError err -- | Convert an abstract time to a serializable protocol value. protocolTime :: KafkaTime -> Time @@ -218,7 +222,7 @@ protocolTime EarliestTime = Time (-2) protocolTime (OtherTime o) = o -updateMetadatas :: [TopicName] -> Kafka () +updateMetadatas :: Kafka m => [TopicName] -> m () updateMetadatas ts = do md <- metadata $ MetadataReq ts let (brokers, tmds) = (md ^.. metadataResponseBrokers . folded, md ^.. topicsMetadata . folded) @@ -232,10 +236,10 @@ addTopicMetadata :: TopicMetadata -> M.Map TopicName TopicMetadata -> M.Map TopicName TopicMetadata addTopicMetadata tm = M.insert (tm ^. topicMetadataName) tm -updateMetadata :: TopicName -> Kafka () +updateMetadata :: Kafka m => TopicName -> m () updateMetadata t = updateMetadatas [t] -updateAllMetadata :: Kafka () +updateAllMetadata :: Kafka m => m () updateAllMetadata = updateMetadatas [] -- | Execute a Kafka action with a 'Handle' for the given 'Broker', updating @@ -247,7 +251,7 @@ -- Note that when the given action throws an exception, any state changes will -- be discarded. This includes both 'IOException's and exceptions thrown by -- 'throwError' from 'Control.Monad.Except'. -withBrokerHandle :: Broker -> (Handle -> Kafka a) -> Kafka a +withBrokerHandle :: Kafka m => Broker -> (Handle -> m a) -> m a withBrokerHandle broker = withAddressHandle (broker2address broker) -- | Execute a Kafka action with a 'Handle' for the given 'KafkaAddress', @@ -259,7 +263,7 @@ -- Note that when the given action throws an exception, any state changes will -- be discarded. This includes both 'IOException's and exceptions thrown by -- 'throwError' from 'Control.Monad.Except'. -withAddressHandle :: KafkaAddress -> (Handle -> Kafka a) -> Kafka a +withAddressHandle :: Kafka m => KafkaAddress -> (Handle -> m a) -> m a withAddressHandle address kafkaAction = do conns <- use stateConnections let foundPool = conns ^. at address @@ -286,7 +290,7 @@ -- Note that when the given action throws an exception, any state changes will -- be discarded. This includes both 'IOException's and exceptions thrown by -- 'throwError' from 'Control.Monad.Except'. -withAnyHandle :: (Handle -> Kafka a) -> Kafka a +withAnyHandle :: Kafka m => (Handle -> m a) -> m a withAnyHandle f = do (addy :| _) <- use stateAddresses x <- withAddressHandle addy f @@ -307,13 +311,13 @@ } -- | Get the first found offset. -getLastOffset :: KafkaTime -> Partition -> TopicName -> Kafka Offset +getLastOffset :: Kafka m => KafkaTime -> Partition -> TopicName -> m Offset getLastOffset m p t = do broker <- getTopicPartitionLeader t p withBrokerHandle broker (\h -> getLastOffset' h m p t) -- | Get the first found offset. -getLastOffset' :: Handle -> KafkaTime -> Partition -> TopicName -> Kafka Offset +getLastOffset' :: Kafka m => Handle -> KafkaTime -> Partition -> TopicName -> m Offset getLastOffset' h m p t = do let offsetRR = OffsetRR $ offsetRequest [(TopicAndPartition t p, PartitionOffsetRequestInfo m 1)] offsetResponse <- makeRequest h offsetRR diff --git a/Network/Kafka/Consumer.hs b/Network/Kafka/Consumer.hs --- a/Network/Kafka/Consumer.hs +++ b/Network/Kafka/Consumer.hs @@ -18,7 +18,7 @@ ordinaryConsumerId = ReplicaId (-1) -- | Construct a fetch request from the values in the state. -fetchRequest :: Offset -> Partition -> TopicName -> Kafka FetchRequest +fetchRequest :: Kafka m => Offset -> Partition -> TopicName -> m FetchRequest fetchRequest o p topic = do wt <- use stateWaitTime ws <- use stateWaitSize @@ -26,10 +26,10 @@ 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' :: Kafka m => Handle -> FetchRequest -> m FetchResponse fetch' h request = makeRequest h $ FetchRR request -fetch :: Offset -> Partition -> TopicName -> Kafka FetchResponse +fetch :: Kafka m => Offset -> Partition -> TopicName -> m FetchResponse fetch o p topic = do broker <- getTopicPartitionLeader topic p withBrokerHandle broker (\handle -> fetch' handle =<< fetchRequest o p topic) diff --git a/Network/Kafka/Producer.hs b/Network/Kafka/Producer.hs --- a/Network/Kafka/Producer.hs +++ b/Network/Kafka/Producer.hs @@ -1,3 +1,5 @@ +{-# LANGUAGE FlexibleContexts #-} + module Network.Kafka.Producer where import Data.Bits ((.&.)) @@ -21,7 +23,7 @@ -- * Producing -- | Execute a produce request and get the raw produce response. -produce :: Handle -> ProduceRequest -> Kafka ProduceResponse +produce :: Kafka m => Handle -> ProduceRequest -> m ProduceResponse produce handle request = makeRequest handle $ ProduceRR request -- | Construct a produce request with explicit arguments. @@ -31,18 +33,25 @@ where f (TopicAndPartition t p, i) = M.singleton t [(p, i)] -- | Send messages to partition calculated by 'partitionAndCollate'. -produceMessages :: [TopicAndMessage] -> Kafka [ProduceResponse] -produceMessages tams = do - m <- fmap (fmap groupMessagesToSet) <$> partitionAndCollate tams +produceMessages :: Kafka m => [TopicAndMessage] -> m [ProduceResponse] +produceMessages = prod (groupMessagesToSet NoCompression) + +-- | Send compressed messages to partition calculated by 'partitionAndCollate'. +produceCompressedMessages :: Kafka m => CompressionCodec -> [TopicAndMessage] -> m [ProduceResponse] +produceCompressedMessages c = prod (groupMessagesToSet c) + +prod :: Kafka m => ([TopicAndMessage] -> MessageSet) -> [TopicAndMessage] -> m [ProduceResponse] +prod g tams = do + m <- fmap (fmap g) <$> partitionAndCollate tams mapM (uncurry send) $ fmap M.toList <$> M.toList m -- | Create a protocol message set from a list of messages. -groupMessagesToSet :: [TopicAndMessage] -> MessageSet -groupMessagesToSet xs = MessageSet $ msm <$> xs +groupMessagesToSet :: CompressionCodec -> [TopicAndMessage] -> MessageSet +groupMessagesToSet c xs = MessageSet c $ msm <$> xs where msm = MessageSetMember (Offset (-1)) . _tamMessage -- | Group messages together with the leader they should be sent to. -partitionAndCollate :: [TopicAndMessage] -> Kafka (M.Map Leader (M.Map TopicAndPartition [TopicAndMessage])) +partitionAndCollate :: Kafka m => [TopicAndMessage] -> m (M.Map Leader (M.Map TopicAndPartition [TopicAndMessage])) partitionAndCollate ks = recurse ks M.empty where recurse [] accum = return accum recurse (x:xs) accum = do @@ -68,7 +77,7 @@ i = x `mod` numPartitions -- | Execute a produce request using the values in the state. -send :: Leader -> [(TopicAndPartition, MessageSet)] -> Kafka ProduceResponse +send :: Kafka m => Leader -> [(TopicAndPartition, MessageSet)] -> m ProduceResponse send l ts = do let s = stateBrokers . at l topicNames = map (_tapTopic . fst) ts @@ -77,7 +86,7 @@ requestTimeout <- use stateRequestTimeout withBrokerHandle broker $ \handle -> produce handle $ produceRequest requiredAcks requestTimeout ts -getRandPartition :: Set PartitionAndLeader -> Kafka (Maybe PartitionAndLeader) +getRandPartition :: Kafka m => Set PartitionAndLeader -> m (Maybe PartitionAndLeader) getRandPartition ps = liftIO $ (ps' ^?) . element <$> getStdRandom (randomR (0, length ps' - 1)) where ps' = ps ^.. folded . filtered (has $ palLeader . leaderId . _Just) @@ -98,7 +107,7 @@ -- | Default: @0@ defaultMessageAttributes :: Attributes -defaultMessageAttributes = 0 +defaultMessageAttributes = Attributes NoCompression -- | Construct a message from a string of bytes using default attributes. makeMessage :: ByteString -> Message diff --git a/Network/Kafka/Protocol.hs b/Network/Kafka/Protocol.hs --- a/Network/Kafka/Protocol.hs +++ b/Network/Kafka/Protocol.hs @@ -10,9 +10,11 @@ import Control.Applicative import Control.Category (Category(..)) +import Control.Exception (Exception) import Control.Lens -import Control.Monad (replicateM, liftM, liftM2, liftM3, liftM4, liftM5, unless) +import Control.Monad (replicateM, liftM2, liftM3, liftM4, liftM5, unless) import Control.Monad.IO.Class (MonadIO, liftIO) +import Data.Bits ((.&.)) import Data.ByteString.Char8 (ByteString) import Data.ByteString.Lens (unpackedChars) import Data.Digest.CRC32 @@ -24,6 +26,8 @@ import Numeric.Lens import Prelude hiding ((.), id) import qualified Data.ByteString.Char8 as B +import qualified Data.ByteString.Lazy.Char8 as LB (fromStrict, toStrict) +import qualified Codec.Compression.GZip as GZip (compress, decompress) import qualified Network data ReqResp a where @@ -38,7 +42,7 @@ B.hPut h $ requestBytes r hFlush h B.hGet h 4 - case runGet (liftM fromIntegral getWord32be) rawLength of + case runGet (fmap fromIntegral getWord32be) rawLength of Left s -> return $ Left s Right dataLength -> do responseBytes <- liftIO $ B.hGet h dataLength @@ -76,8 +80,11 @@ deriving (Show, Eq) newtype MetadataRequest = MetadataReq [TopicName] deriving (Show, Eq, Serializable, Deserializable) -newtype TopicName = TName { _tName :: KafkaString } deriving (Show, Eq, Ord, Deserializable, Serializable, IsString) +newtype TopicName = TName { _tName :: KafkaString } deriving (Eq, Ord, Deserializable, Serializable, IsString) +instance Show TopicName where + show = show . B.unpack . _kString. _tName + newtype KafkaBytes = KBytes { _kafkaByteString :: ByteString } deriving (Show, Eq, IsString) newtype KafkaString = KString { _kString :: ByteString } deriving (Show, Eq, Ord, IsString) @@ -138,8 +145,8 @@ newtype Partition = Partition Int32 deriving (Show, Eq, Serializable, Deserializable, Num, Integral, Ord, Real, Enum) -newtype MessageSet = - MessageSet { _messageSetMembers :: [MessageSetMember] } deriving (Show, Eq) +data MessageSet = MessageSet { _codec :: CompressionCodec, _messageSetMembers :: [MessageSetMember] } + deriving (Show, Eq) data MessageSetMember = MessageSetMember { _setOffset :: Offset, _setMessage :: Message } deriving (Show, Eq) @@ -149,9 +156,11 @@ Message { _messageFields :: (Crc, MagicByte, Attributes, Key, Value) } deriving (Show, Eq, Deserializable) +data CompressionCodec = NoCompression | Gzip deriving (Show, Eq) + 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) +data Attributes = Attributes { _compressionCodec :: CompressionCodec } deriving (Show, Eq) newtype Key = Key { _keyBytes :: Maybe KafkaBytes } deriving (Show, Eq) newtype Value = Value { _valueBytes :: Maybe KafkaBytes } deriving (Show, Eq) @@ -236,6 +245,8 @@ 16 -> return NotCoordinatorForConsumerCode _ -> fail $ "invalid error code: " ++ show x +instance Exception KafkaError + newtype Request = Request (CorrelationId, ClientId, RequestMessage) deriving (Show, Eq) instance Serializable Request where @@ -256,13 +267,13 @@ apiVersion _ = ApiVersion 0 -- everything is at version 0 right now apiKey :: RequestMessage -> ApiKey -apiKey (ProduceRequest{}) = ApiKey 0 -apiKey (FetchRequest{}) = ApiKey 1 -apiKey (OffsetRequest{}) = ApiKey 2 -apiKey (MetadataRequest{}) = ApiKey 3 -apiKey (OffsetCommitRequest{}) = ApiKey 8 -apiKey (OffsetFetchRequest{}) = ApiKey 9 -apiKey (GroupCoordinatorRequest{}) = ApiKey 10 +apiKey ProduceRequest{} = ApiKey 0 +apiKey FetchRequest{} = ApiKey 1 +apiKey OffsetRequest{} = ApiKey 2 +apiKey MetadataRequest{} = ApiKey 3 +apiKey OffsetCommitRequest{} = ApiKey 8 +apiKey OffsetFetchRequest{} = ApiKey 9 +apiKey GroupCoordinatorRequest{} = ApiKey 10 instance Serializable RequestMessage where serialize (ProduceRequest r) = serialize r @@ -293,12 +304,36 @@ putByteString bs instance Serializable MessageSet where - serialize (MessageSet ms) = do - let bytes = runPut $ mapM_ serialize ms + serialize (MessageSet codec messageSet) = do + let bytes = runPut $ mapM_ serialize (compress codec messageSet) l = fromIntegral (B.length bytes) :: Int32 serialize l putByteString bytes + where compress :: CompressionCodec -> [MessageSetMember] -> [MessageSetMember] + compress NoCompression ms = ms + compress c ms = [MessageSetMember (Offset (-1)) (message c ms)] + + message :: CompressionCodec -> [MessageSetMember] -> Message + message c ms = Message (0, 0, Attributes c, Key Nothing, value (compressor c) ms) + + compressor :: CompressionCodec -> (ByteString -> ByteString) + compressor c = case c of + Gzip -> LB.toStrict . GZip.compress . LB.fromStrict + _ -> fail "Unsupported compression codec" + + value :: (ByteString -> ByteString) -> [MessageSetMember] -> Value + value c ms = Value . Just . KBytes $ c (runPut $ mapM_ serialize ms) + +instance Serializable Attributes where + serialize = serialize . bits + where bits :: Attributes -> Int8 + bits = codecValue . _compressionCodec + + codecValue :: CompressionCodec -> Int8 + codecValue NoCompression = 0 + codecValue Gzip = 1 + instance Serializable KafkaBytes where serialize (KBytes bs) = do let l = fromIntegral (B.length bs) :: Int32 @@ -337,7 +372,11 @@ deserialize = do l <- deserialize :: Get Int32 ms <- isolate (fromIntegral l) getMembers - return $ MessageSet ms + + decompressed <- mapM decompress ms + + return $ MessageSet NoCompression (concat decompressed) + where getMembers :: Get [MessageSetMember] getMembers = do wasEmpty <- isEmpty @@ -345,6 +384,39 @@ then return [] else liftM2 (:) deserialize getMembers <|> (remaining >>= getBytes >> return []) + decompress :: MessageSetMember -> Get [MessageSetMember] + decompress m = if isCompressed m + then decompressSetMember m + else return [m] + + isCompressed :: MessageSetMember -> Bool + isCompressed = messageCompressed . _setMessage + + messageCompressed :: Message -> Bool + messageCompressed (Message (_, _, att, _, _)) = _compressionCodec att /= NoCompression + + decompressSetMember :: MessageSetMember -> Get [MessageSetMember] + decompressSetMember (MessageSetMember _ (Message (_, _, att, _, Value v))) = case v of + Just bytes -> decompressMessage (decompressor att) (_kafkaByteString bytes) + Nothing -> fail "Expecting a compressed message set, empty data set received" + + decompressor :: Attributes -> (ByteString -> ByteString) + decompressor att = case _compressionCodec att of + Gzip -> LB.toStrict . GZip.decompress . LB.fromStrict + _ -> fail "Unsupported compression codec." + + decompressMessage :: (ByteString -> ByteString) -> ByteString -> Get [MessageSetMember] + decompressMessage f = getDecompressedMembers . f + + getDecompressedMembers :: ByteString -> Get [MessageSetMember] + getDecompressedMembers "" = return [] -- a compressed empty message + getDecompressedMembers val = do + let res = runGetPartial deserialize val :: Result MessageSetMember + case res of + Fail err _ -> fail err + Partial _ -> fail "Could not consume all available data" + Done v vv -> fmap (v :) (getDecompressedMembers vv) + instance Deserializable MessageSetMember where deserialize = do o <- deserialize @@ -358,6 +430,20 @@ let l = Leader $ if x == -1 then Nothing else Just x return l +instance Deserializable Attributes where + deserialize = do + i <- deserialize :: Get Int8 + codec <- case compressionCodecFromValue i of + Just c -> return c + Nothing -> fail $ "Unknown compression codec value found in: " ++ show i + return $ Attributes codec + +compressionCodecFromValue :: Int8 -> Maybe CompressionCodec +compressionCodecFromValue i | eq 1 = Just Gzip + | eq 0 = Just NoCompression + | otherwise = Nothing + where eq y = i .&. y == y + instance Deserializable KafkaBytes where deserialize = do l <- deserialize :: Get Int32 @@ -402,10 +488,10 @@ instance (Deserializable a, Deserializable b, Deserializable c, Deserializable d, Deserializable e) => Deserializable ((,,,,) a b c d e) where deserialize = liftM5 (,,,,) deserialize deserialize deserialize deserialize deserialize -instance Deserializable Int64 where deserialize = liftM fromIntegral getWord64be -instance Deserializable Int32 where deserialize = liftM fromIntegral getWord32be -instance Deserializable Int16 where deserialize = liftM fromIntegral getWord16be -instance Deserializable Int8 where deserialize = liftM fromIntegral getWord8 +instance Deserializable Int64 where deserialize = fmap fromIntegral getWord64be +instance Deserializable Int32 where deserialize = fmap fromIntegral getWord32be +instance Deserializable Int16 where deserialize = fmap fromIntegral getWord16be +instance Deserializable Int8 where deserialize = fmap fromIntegral getWord8 -- * Generated lenses diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -3,7 +3,7 @@ A Kafka client for Haskell. -Supports Kafka 0.8.x and up (including 0.9.x). +Supports Kafka 0.8.x and up (including 0.10.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. @@ -11,5 +11,5 @@ Some of the known limitations. -* Does not yet support compression +* Does not yet support Snappy 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.5.0.2 +version: 0.5.1.0 synopsis: A Kafka client for Haskell. description: A Kafka client for Haskell. @@ -21,6 +21,8 @@ extra-source-files: README.md bug-reports: https://github.com/adamflott/milena/issues tested-with: GHC == 7.10.3, GHC == 8.0.1 +extra-source-files: + CHANGELOG.md source-repository head type: git @@ -45,13 +47,15 @@ network >=2.4 && <2.7, digest >=0.0.1.0 && <0.1, containers >=0.5 && <0.6, + monad-control >=1.0 && <1.1, random >=1.0 && <1.2, transformers >=0.3 && <0.6, lens >=4.4 && <4.16, resource-pool >=0.2.3.2 && <0.3, lifted-base >=0.2.3.6 && <0.3, murmur-hash >=0.1.0.8 && <0.2, - semigroups >=0.16.2.2 && <0.19 + semigroups >=0.16.2.2 && <0.19, + zlib >=0.6.1.2 && <0.7 test-suite test default-language: Haskell2010 diff --git a/test/tests.hs b/test/tests.hs --- a/test/tests.hs +++ b/test/tests.hs @@ -11,7 +11,7 @@ import Network.Kafka import Network.Kafka.Consumer import Network.Kafka.Producer -import Network.Kafka.Protocol (ProduceResponse(..), KafkaError(..)) +import Network.Kafka.Protocol (ProduceResponse(..), KafkaError(..), CompressionCodec(..)) import Test.Tasty import Test.Tasty.Hspec import Test.Tasty.QuickCheck @@ -37,11 +37,16 @@ result <- run . produceMessages $ byteMessages ms result `shouldSatisfy` isRight - prop "can produce multiple messages" $ \(ms, ms') -> do + prop "can produce compressed messages" $ \ms -> do + result <- run . produceCompressedMessages Gzip $ byteMessages ms + result `shouldSatisfy` isRight + + prop "can produce multiple messages" $ \(ms, ms', ms'') -> do result <- run $ do r1 <- produceMessages $ byteMessages ms r2 <- produceMessages $ byteMessages ms' - return $ r1 ++ r2 + r3 <- produceCompressedMessages Gzip $ byteMessages ms'' + return $ r1 ++ r2 ++ r3 result `shouldSatisfy` isRight prop "can fetch messages" $ do @@ -55,25 +60,49 @@ result <- run $ do requireAllAcks info <- brokerPartitionInfo 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 + + case getPartitionByKey (B.pack key) info of + Just PartitionAndLeader { _palLeader = leader, _palPartition = partition } -> do + let payload = [(TopicAndPartition topic partition, groupMessagesToSet NoCompression 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 + + Nothing -> fail "Could not deduce partition" + result `shouldBe` Right (tamPayload <$> messages) + prop "can roundtrip compressed messages" $ \(NonEmpty ms) -> do + let messages = byteMessages ms + result <- run $ do + requireAllAcks + produceResps <- produceCompressedMessages Gzip messages + + case map _produceResponseFields produceResps of + [[(_topicName, [(partition, NoError, offset)])]] -> do + resp <- fetch offset partition topic + return $ fmap tamPayload . fetchMessages $ resp + + _ -> fail "Unexpected produce response" + + 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 + case map _produceResponseFields produceResps of + [[(_topicName, [(partition, NoError, offset)])]] -> do + resp <- fetch offset partition topic + return $ fmap tamPayload . fetchMessages $ resp + + _ -> fail "Unexpected produce response" + result `shouldBe` Right (tamPayload <$> messages) describe "withAddressHandle" $ do