milena 0.5.0.2 → 0.5.1.0

raw patch · 8 files changed

+213/−73 lines, 8 filesdep +monad-controldep +zlibPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: monad-control, zlib

API changes (from Hackage documentation)

Files

• CHANGELOG.md +8/−0
• Network/Kafka.hs +27/−23
• Network/Kafka/Consumer.hs +3/−3
• Network/Kafka/Producer.hs +19/−10
• Network/Kafka/Protocol.hs +106/−20
• README.md +2/−2
• milena.cabal +6/−2
• test/tests.hs +42/−13

@@ -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

@@ -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

@@ -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)

@@ -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

@@ -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 

@@ -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

@@ -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

@@ -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