milena 0.5.3.0 → 0.5.4.0

raw patch · 5 files changed

+176/−20 lines, 5 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• CHANGELOG.md +10/−0
• Network/Kafka.hs +38/−0
• Network/Kafka/Protocol.hs +26/−7
• milena.cabal +3/−3
• test/tests.hs +99/−10

@@ -1,3 +1,13 @@+0.5.4.0++* commitOffset and fetchOffset to the API (https://github.com/adamflott/milena/pull/7)+* Add Delete Topic support+* Make tests pass on multiple invocations++0.5.3.0++* Add Create Topic support (https://github.com/adamflott/milena/pull/6). Thanks matth!+ 0.5.2.3  * Bump network upper bounds

@@ -202,6 +202,44 @@     CreateTopicsReq         ([(topic, partition, replication_factor, replica_assignment, config)], defaultRequestTimeout) +deleteTopic :: Kafka m => DeleteTopicsRequest -> m DeleteTopicsResponse+deleteTopic request = withAnyHandle $ flip deleteTopic' request++deleteTopic' :: Kafka m => Handle -> DeleteTopicsRequest -> m DeleteTopicsResponse+deleteTopic' h request = makeRequest h $ DeleteTopicsRR request++deleteTopicsRequest :: TopicName -> DeleteTopicsRequest+deleteTopicsRequest topic = DeleteTopicsReq ([topic], defaultRequestTimeout)+++fetchOffset :: Kafka m => OffsetFetchRequest -> m OffsetFetchResponse+fetchOffset request = withAnyHandle $ flip fetchOffset' request++fetchOffset' :: Kafka m => Handle -> OffsetFetchRequest -> m OffsetFetchResponse+fetchOffset' h request = makeRequest h $ OffsetFetchRR request++fetchOffsetRequest ::+     ConsumerGroup -> TopicName -> Partition -> OffsetFetchRequest+fetchOffsetRequest consumerGroup topic partition =+  OffsetFetchReq+        (consumerGroup, [(topic, [partition])])+++commitOffset :: Kafka m => OffsetCommitRequest -> m OffsetCommitResponse+commitOffset request = withAnyHandle $ flip commitOffset' request++commitOffset' ::+     Kafka m => Handle -> OffsetCommitRequest -> m OffsetCommitResponse+commitOffset' h request = makeRequest h $ OffsetCommitRR request++commitOffsetRequest ::+     ConsumerGroup -> TopicName -> Partition -> Offset -> OffsetCommitRequest+commitOffsetRequest consumerGroup topic partition offset =+  let time = -1+      metadata_ = Metadata "milena"+   in OffsetCommitReq+        (consumerGroup, [(topic, [(partition, offset, time, metadata_)])])+  getTopicPartitionLeader :: Kafka m => TopicName -> Partition -> m Broker getTopicPartitionLeader t p = do

@@ -2,24 +2,28 @@   ( module Network.Kafka.Protocol   ) where +import Prelude hiding ((.), id)++-- base import Control.Applicative import Control.Category (Category(..)) import Control.Exception (Exception)-import Control.Lens import Control.Monad (replicateM, liftM2, liftM3, liftM4, liftM5, unless)-import Control.Monad.IO.Class (MonadIO, liftIO) import Data.Bits ((.&.))+import Data.Int+import GHC.Exts (IsString(..))+import GHC.Generics (Generic)+import System.IO++-- Hackage+import Control.Lens+import Control.Monad.IO.Class (MonadIO, liftIO) import Data.ByteString.Char8 (ByteString) import Data.ByteString.Lens (unpackedChars) import Data.Digest.CRC32-import Data.Int import Data.Serialize.Get import Data.Serialize.Put-import GHC.Exts (IsString(..))-import GHC.Generics (Generic)-import System.IO 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)@@ -31,6 +35,9 @@   FetchRR    :: MonadIO m => FetchRequest    -> ReqResp (m FetchResponse)   OffsetRR   :: MonadIO m => OffsetRequest   -> ReqResp (m OffsetResponse)   TopicsRR   :: MonadIO m => CreateTopicsRequest -> ReqResp (m CreateTopicsResponse)+  DeleteTopicsRR :: MonadIO m => DeleteTopicsRequest -> ReqResp (m DeleteTopicsResponse)+  OffsetCommitRR :: MonadIO m => OffsetCommitRequest -> ReqResp (m OffsetCommitResponse)+  OffsetFetchRR :: MonadIO m => OffsetFetchRequest -> ReqResp (m OffsetFetchResponse)  doRequest' :: (Deserializable a, MonadIO m) => CorrelationId -> Handle -> Request -> m (Either String a) doRequest' correlationId h r = do@@ -53,6 +60,9 @@ 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) doRequest clientId correlationId h (TopicsRR req)   = doRequest' correlationId h $ Request (correlationId, clientId, CreateTopicsRequest req)+doRequest clientId correlationId h (DeleteTopicsRR req)   = doRequest' correlationId h $ Request (correlationId, clientId, DeleteTopicsRequest req)+doRequest clientId correlationId h (OffsetCommitRR req) = doRequest' correlationId h $ Request (correlationId, clientId, OffsetCommitRequest req)+doRequest clientId correlationId h (OffsetFetchRR req) = doRequest' correlationId h $ Request (correlationId, clientId, OffsetFetchRequest req)  class Serializable a where   serialize :: a -> Put@@ -75,6 +85,7 @@                     | OffsetFetchRequest OffsetFetchRequest                     | GroupCoordinatorRequest GroupCoordinatorRequest                     | CreateTopicsRequest CreateTopicsRequest+                    | DeleteTopicsRequest DeleteTopicsRequest                     deriving (Show, Generic, Eq)  newtype MetadataRequest = MetadataReq [TopicName] deriving (Show, Eq, Serializable, Generic, Deserializable)@@ -106,6 +117,10 @@   TopicsResp { _topicsResponseFields :: [(TopicName, KafkaError)] }   deriving (Show, Eq, Deserializable, Serializable, Generic) +newtype DeleteTopicsResponse =+  DeleteTopicsResp { _deleteTopicsResponseFields :: [(TopicName, KafkaError)] }+  deriving (Show, Eq, Deserializable, Serializable, Generic)+ newtype MetadataResponse = MetadataResp { _metadataResponseFields :: ([Broker], [TopicMetadata]) } deriving (Show, Eq, Deserializable, Generic) newtype Broker = Broker { _brokerFields :: (NodeId, Host, Port) } deriving (Show, Eq, Ord, Deserializable, Generic) newtype NodeId = NodeId { _nodeId :: Int32 } deriving (Show, Eq, Deserializable, Num, Integral, Ord, Real, Enum, Generic)@@ -181,6 +196,8 @@  newtype GroupCoordinatorRequest = GroupCoordinatorReq ConsumerGroup deriving (Show, Eq, Serializable, Generic) newtype CreateTopicsRequest = CreateTopicsReq ([(TopicName, Partition, ReplicationFactor, [(Partition, Replicas)], [(KafkaString, Metadata)])], Timeout) deriving (Show, Eq, Serializable, Generic)+newtype DeleteTopicsRequest = DeleteTopicsReq ([TopicName], Timeout) deriving (Show, Eq, Serializable, Generic)+ newtype OffsetCommitRequest = OffsetCommitReq (ConsumerGroup, [(TopicName, [(Partition, Offset, Time, Metadata)])]) deriving (Show, Eq, Serializable, Generic) newtype OffsetFetchRequest = OffsetFetchReq (ConsumerGroup, [(TopicName, [Partition])]) deriving (Show, Eq, Serializable, Generic) newtype ConsumerGroup = ConsumerGroup KafkaString deriving (Show, Eq, Serializable, Deserializable, IsString, Generic)@@ -286,6 +303,7 @@ apiKey OffsetFetchRequest{} = ApiKey 9 apiKey GroupCoordinatorRequest{} = ApiKey 10 apiKey CreateTopicsRequest{} = ApiKey 19+apiKey DeleteTopicsRequest{} = ApiKey 20  instance Serializable RequestMessage where   serialize (ProduceRequest r) = serialize r@@ -296,6 +314,7 @@   serialize (OffsetFetchRequest r) = serialize r   serialize (GroupCoordinatorRequest r) = serialize r   serialize (CreateTopicsRequest r) = serialize r+  serialize (DeleteTopicsRequest r) = serialize r  instance Serializable Int64 where serialize = putWord64be . fromIntegral instance Serializable Int32 where serialize = putWord32be . fromIntegral

@@ -1,13 +1,13 @@ cabal-version: 1.12 --- This file has been generated from package.yaml by hpack version 0.31.1.+-- This file has been generated from package.yaml by hpack version 0.31.2. -- -- see: https://github.com/sol/hpack ----- hash: 8060a8b3f6bf4aad48a041ce65dd55877fe0f1bd5353f32359bae70b010f2843+-- hash: a7bdec96a93aec94bc09e9d89b957ff87d484a0d2f73237f8816de0a56cd03b4  name:           milena-version:        0.5.3.0+version:        0.5.4.0 synopsis:       A Kafka client for Haskell. description:    A Kafka client for Haskell.                 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.

@@ -2,23 +2,42 @@  module Main where -import Data.Functor+import Prelude++-- base import Data.Either (isRight, isLeft) import qualified Data.List.NonEmpty as NE++-- Hackage import Control.Lens 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 (ProduceResponse(..), KafkaError(..), CompressionCodec(..), CreateTopicsResponse(..)) import Test.Tasty import Test.Tasty.Hspec import Test.Tasty.QuickCheck import qualified Data.ByteString.Char8 as B -import Prelude+-- local+import Network.Kafka+import Network.Kafka.Consumer+import Network.Kafka.Producer+import Network.Kafka.Protocol (ProduceResponse(..),+                               KafkaError(..),+                               CompressionCodec(..),+                               CreateTopicsResponse(..),+                               DeleteTopicsResponse(..),+                               Offset(..),+                               OffsetCommitResponse(..),+                               OffsetFetchResponse(..),+                               ConsumerGroup(..),+                               Partition(..),+                               KafkaString(..),+                               Metadata(..),+                               TopicName(..),+                              ) ++ main :: IO () main = testSpec "the specs" specs >>= defaultMain @@ -32,6 +51,13 @@         stateWaitTime .= 1000       byteMessages = fmap (TopicAndMessage topic . makeMessage . B.pack) +  let cleanup :: TopicName -> IO ()+      cleanup topicName = do+          run $ do+              stateAddresses %= NE.cons ("localhost", 9092)+              deleteTopic (deleteTopicsRequest topicName)+          pure ()+   describe "can talk to local Kafka server" $ do     prop "can produce messages" $ \ms -> do       result <- run . produceMessages $ byteMessages ms@@ -129,13 +155,76 @@         use stateAddresses       result `shouldBe` fmap NE.nub result -  describe "create topics" $++  let newTopicName = "milena-test-13-partitions"+  describe "create topics" $ do     it "create topics with multiple partitions" $ do-      let t = "milena-test-13-partitions"       result <- run $ do         stateAddresses %= NE.cons ("localhost", 9092)-        createTopic (createTopicsRequest t 13 1 [] [])-      result `shouldBe` (Right $ TopicsResp [(t, NoError)])+        createTopic (createTopicsRequest newTopicName 13 1 [] [])+      result `shouldBe` (Right $ TopicsResp [(newTopicName, NoError)])++    it "create already existing topic" $ do+      result <- run $ do+        stateAddresses %= NE.cons ("localhost", 9092)+        createTopic (createTopicsRequest newTopicName 13 1 [] [])+      result `shouldBe` (Right $ TopicsResp [(newTopicName, TopicAlreadyExists)])++  describe "delete topics" $+    it "delete topics" $ do+      result <- run $ do+        stateAddresses %= NE.cons ("localhost", 9092)+        deleteTopic (deleteTopicsRequest newTopicName)+      result `shouldBe` (Right $ DeleteTopicsResp [(newTopicName, NoError)])+++  let commitOffsetTopicName = "commit-offset"+      t = commitOffsetTopicName++  Test.Tasty.Hspec.afterAll_ (cleanup commitOffsetTopicName) $ do+    describe "can commit messages" $ do+      it "create a topic" $ do+        topicCreation <- run $ do+          stateAddresses %= NE.cons ("localhost", 9092)+          createTopic (createTopicsRequest t 3 1 [] [])+        topicCreation `shouldBe` (Right $ TopicsResp [(t, NoError)])++      it "commit offset 5 to partition 0 for consumer group \"group1\"" $ do+        commitOff <- run $ do+            stateAddresses %= NE.cons ("localhost", 9092)+            commitOffset (commitOffsetRequest (ConsumerGroup "group1") t 0 (Offset 5))+        commitOff `shouldBe` Right (OffsetCommitResp [(t,[(Partition 0,NoError)])])++      it "commit offset 15 to partition 1 for consumer group \"group1\"" $ do+        commitOff <- run $ do+            stateAddresses %= NE.cons ("localhost", 9092)+            commitOffset (commitOffsetRequest (ConsumerGroup "group1") t 1 (Offset 15))+        commitOff `shouldBe` Right (OffsetCommitResp [(t,[(Partition 1,NoError)])])++      it "commit offset 10 to partition 2 for consumer group \"group2\"" $ do+        commitOff <- run $ do+            stateAddresses %= NE.cons ("localhost", 9092)+            commitOffset (commitOffsetRequest (ConsumerGroup "group2") t 2 10)+        commitOff `shouldSatisfy` isRight++      it "fetch offset from partition 0 for \"group1\"" $ do+        fetchOff <- run $ do+            stateAddresses %= NE.cons ("localhost", 9092)+            fetchOffset (fetchOffsetRequest (ConsumerGroup "group1") t 0)+        fetchOff `shouldBe` Right (OffsetFetchResp [(t,[(Partition 0, Offset 5,Metadata (KString {_kString = ""}),NoError)])])++      it "fetch offset from partition 0 for \"group2\"" $ do+        fetchOff <- run $ do+            stateAddresses %= NE.cons ("localhost", 9092)+            fetchOffset (fetchOffsetRequest (ConsumerGroup "group2") t 0)+        fetchOff `shouldBe` Right (OffsetFetchResp [(t,[(Partition 0, Offset (-1),Metadata (KString {_kString = ""}),UnknownTopicOrPartition)])])++      it "note that getLastOffset is unchanged" $ do+        getLastOff <- run $ do+            stateAddresses %= NE.cons ("localhost", 9092)+            getLastOffset EarliestTime 0 t+        getLastOff `shouldSatisfy` isRight+        getLastOff `shouldBe` (Right $ Offset 0)  prop :: Testable prop => String -> prop -> SpecWith () prop s = it s . property