milena 0.2.0.0 → 0.3.0.0

raw patch · 5 files changed

+34/−10 lines, 5 filesdep +lifted-basedep +listsafedep +murmur-hashdep ~lensdep ~mtlPVP ok

version bump matches the API change (PVP)

Dependencies added: lifted-base, listsafe, murmur-hash

Dependency ranges changed: lens, mtl

API changes (from Hackage documentation)

Files

• Network/Kafka.hs +1/−0
• Network/Kafka/Producer.hs +15/−3
• Network/Kafka/Protocol.hs +6/−0
• milena.cabal +10/−5
• test/tests.hs +2/−2

@@ -1,5 +1,6 @@ {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE FlexibleContexts #-}  module Network.Kafka where 

@@ -1,10 +1,13 @@ module Network.Kafka.Producer where +import Prelude hiding ((!!)) import Control.Applicative import Control.Lens import Control.Monad.Trans (liftIO, lift) import Control.Monad.Trans.Either import Data.ByteString.Char8 (ByteString)+import qualified Data.Digest.Murmur32 as Murmur32+import Data.List.Safe ((!!)) import Data.Monoid ((<>)) import System.IO import qualified Data.Map as M@@ -43,13 +46,22 @@       where recurse [] accum = return accum             recurse (x:xs) accum = do               topicPartitionsList <- brokerPartitionInfo $ _tamTopic x-              pal <- getPartition topicPartitionsList+              let maybeKey = x ^. tamMessage . messageKey . keyBytes+              pal <- case maybeKey of+                Nothing -> getRandPartition topicPartitionsList+                Just key -> return $ getPartitionByKey (_kafkaByteString key) topicPartitionsList               let leader = maybe (Leader Nothing) _palLeader pal                   tp = TopicAndPartition <$> pal ^? folded . palTopic <*> pal ^? folded . palPartition                   b = M.singleton leader $ maybe M.empty (`M.singleton` [x]) tp                   accum' = M.unionWith (M.unionWith (<>)) accum b               recurse xs accum' +-- | 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 !! i+ -- | Execute a produce request using the values in the state. send :: Leader -> [(TopicAndPartition, MessageSet)] -> Kafka ProduceResponse send l ts = do@@ -80,8 +92,8 @@         Just x -> return x         Nothing -> lift $ left $ err -getPartition :: [PartitionAndLeader] -> Kafka (Maybe PartitionAndLeader)-getPartition ps =+getRandPartition :: [PartitionAndLeader] -> Kafka (Maybe PartitionAndLeader)+getRandPartition ps =     liftIO $ (ps' ^?) . element <$> getStdRandom (randomR (0, length ps' - 1))         where ps' = ps ^.. folded . filtered (has $ palLeader . leaderId . _Just) 

@@ -491,6 +491,12 @@ fetchResponseMessageMembers :: Fold FetchResponse MessageSetMember fetchResponseMessageMembers = fetchResponseMessages . messageSetMembers . folded +messageKey :: Lens' Message Key+messageKey = messageFields . _4++messageKeyBytes :: Fold Message ByteString+messageKeyBytes = messageKey . keyBytes . folded . kafkaByteString+ messageValue :: Lens' Message Value messageValue = messageFields . _5 

@@ -4,9 +4,10 @@ -- PVP summary:      +-+------- breaking API changes --                   | | +----- non-breaking API additions --                   | | | +--- code changes with no API change-version:             0.2.0.0+version:             0.3.0.0 synopsis:            A Kafka client for Haskell.-description:         Kafka client for Haskell (not recommended for production use).+description:+    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. license:             BSD3 license-file:        LICENSE author:              Tyler Holien@@ -24,7 +25,7 @@ source-repository this   type: git   location: https://github.com/tylerholien/milena.git-  tag: 0.2.0.0+  tag: 0.3.0.0  library   default-language: Haskell2010@@ -42,14 +43,18 @@                        either        >=4.3      && <4.4,                        random        >=1.0      && <1.2,                        transformers  >=0.3      && <0.5,-                       lens          >=4.4      && <4.8,-                       resource-pool >=0.2.3.2  && <0.3+                       lens          >=4.4      && <4.13,+                       resource-pool >=0.2.3.2  && <0.3,+                       lifted-base   >=0.2.3.6  && <0.3,+                       murmur-hash   >=0.1.0.8  && <0.2,+                       listsafe      >=0.1.0.1  && <0.2  test-suite test   default-language: Haskell2010   ghc-options:         -Wall -threaded   build-depends:       base,                        milena,+                       mtl,                        network,                        QuickCheck,                        bytestring,

@@ -14,7 +14,7 @@ main :: IO () main = hspec $ do   let topic = "milena-test"-      run = runKafka ("kafka1.dev", 9093) $ defaultState "milena-test-client"+      run = runKafka ("localhost", 9092) $ defaultState "milena-test-client"       byteMessages = fmap (TopicAndMessage topic . makeMessage . B.pack)    describe "can talk to local Kafka server" $ do@@ -32,7 +32,7 @@       let messages = byteMessages ms       result <- run $ do         info <- brokerPartitionInfo topic-        leader <- maybe (Leader Nothing) _palLeader <$> getPartition info+        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)