hw-kafka-client 2.1.1 → 2.1.2
raw patch · 8 files changed
+56/−48 lines, 8 files
Files
- example/ConsumerExample.hs +6/−2
- hw-kafka-client.cabal +1/−1
- src/Kafka/Consumer.hs +2/−2
- src/Kafka/Consumer/ConsumerProperties.hs +16/−14
- src/Kafka/Internal/Shared.hs +10/−9
- src/Kafka/Producer.hs +6/−6
- src/Kafka/Producer/ProducerProperties.hs +15/−12
- src/Kafka/Types.hs +0/−2
example/ConsumerExample.hs view
@@ -34,8 +34,12 @@ mapM_ (\_ -> do msg1 <- pollMessage kafka (Timeout 1000) putStrLn $ "Message: " <> show msg1- err <- commitAllOffsets OffsetCommit kafka- putStrLn $ "Offsets: " <> maybe "Committed." show err+ case msg1 of+ Right _ -> do+ err <- commitAllOffsets OffsetCommit kafka+ putStrLn $ "Offsets: " <> maybe "Committed." show err+ Left _ -> return ()+ ) [0 :: Integer .. 10] return $ Right ()
hw-kafka-client.cabal view
@@ -1,5 +1,5 @@ name: hw-kafka-client-version: 2.1.1+version: 2.1.2 homepage: https://github.com/haskell-works/hw-kafka-client bug-reports: https://github.com/haskell-works/hw-kafka-client/issues license: MIT
src/Kafka/Consumer.hs view
@@ -42,7 +42,7 @@ -- | Runs high-level kafka consumer. -- A callback provided is expected to call 'pollMessage' when convenient.-{-# DEPRECATED runConsumer "Use 'newConsumer'/'closeConsumer' instead" #-}+{-# DEPRECATED runConsumer "Use @newConsumer@/@closeConsumer@ instead" #-} runConsumer :: ConsumerProperties -> Subscription -> (KafkaConsumer -> IO (Either KafkaError a)) -- ^ A callback function to poll and handle messages@@ -200,7 +200,7 @@ ----------------------------------------------------------------------------- newConsumerConf :: ConsumerProperties -> IO KafkaConf-newConsumerConf (ConsumerProperties m _ cbs) = do+newConsumerConf ConsumerProperties {cpProps = m, cpCallbacks = cbs} = do conf <- kafkaConf (KafkaProps $ M.toList m) forM_ cbs (\setCb -> setCb conf) return conf
src/Kafka/Consumer/ConsumerProperties.hs view
@@ -9,7 +9,6 @@ import qualified Data.List as L import Data.Map (Map) import qualified Data.Map as M-import Data.Semigroup import Kafka.Consumer.Callbacks import Kafka.Consumer.Types import Kafka.Internal.Setup@@ -17,19 +16,22 @@ -- | Properties to create 'KafkaConsumer'. data ConsumerProperties = ConsumerProperties- { cpProps :: Map String String- , cpLogLevel :: Maybe KafkaLogLevel- , callbacks :: [KafkaConf -> IO ()]+ { cpProps :: Map String String+ , cpLogLevel :: Maybe KafkaLogLevel+ , cpCallbacks :: [KafkaConf -> IO ()] } -- | /Right biased/ so we prefer newer properties over older ones.-instance Semigroup ConsumerProperties where- (<>) (ConsumerProperties m1 ll1 cb1) (ConsumerProperties m2 ll2 cb2) =- ConsumerProperties (M.union m2 m1) (ll2 `mplus` ll1) (cb2 <> cb1)- instance Monoid ConsumerProperties where- mempty = ConsumerProperties M.empty Nothing []- mappend = (<>)+ mempty = ConsumerProperties+ { cpProps = M.empty+ , cpLogLevel = Nothing+ , cpCallbacks = []+ }+ {-# INLINE mempty #-}+ mappend (ConsumerProperties m1 ll1 cb1) (ConsumerProperties m2 ll2 cb2) =+ ConsumerProperties (M.union m2 m1) (ll2 `mplus` ll1) (cb2 `mplus` cb1)+ {-# INLINE mappend #-} brokersList :: [BrokerAddress] -> ConsumerProperties brokersList bs =@@ -51,12 +53,12 @@ extraProps $ M.fromList [("client.id", cid)] setCallback :: (KafkaConf -> IO ()) -> ConsumerProperties-setCallback cb = ConsumerProperties M.empty Nothing [cb]+setCallback cb = mempty { cpCallbacks = [cb] } -- | Sets the logging level. -- Usually is used with 'debugOptions' to configure which logs are needed. logLevel :: KafkaLogLevel -> ConsumerProperties-logLevel ll = ConsumerProperties M.empty (Just ll) []+logLevel ll = mempty { cpLogLevel = Just ll } -- | Sets the compression codec for the consumer. compression :: KafkaCompressionCodec -> ConsumerProperties@@ -74,13 +76,13 @@ -- | Any configuration options that are supported by /librdkafka/. -- The full list can be found <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md here> extraProps :: Map String String -> ConsumerProperties-extraProps m = ConsumerProperties m Nothing []+extraProps m = mempty { cpProps = m } {-# INLINE extraProps #-} -- | Any configuration options that are supported by /librdkafka/. -- The full list can be found <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md here> extraProp :: String -> String -> ConsumerProperties-extraProp k v = ConsumerProperties (M.singleton k v) Nothing []+extraProp k v = mempty { cpProps = M.singleton k v } {-# INLINE extraProp #-} -- | Sets debug features for the consumer.
@@ -1,9 +1,9 @@ module Kafka.Internal.Shared where -import Control.Concurrent (forkIO)+import Control.Concurrent (forkIO, rtsSupportsBoundThreads) import Control.Exception-import Control.Monad (void)+import Control.Monad (void, when) import qualified Data.ByteString as BS import qualified Data.ByteString.Internal as BSI import Foreign.C.Error@@ -13,13 +13,14 @@ import Kafka.Types runEventLoop :: HasKafka a => a -> CancellationToken -> Maybe Timeout -> IO ()-runEventLoop k ct timeout = void $ forkIO go- where- go = do- token <- CToken.status ct- case token of- Running -> pollEvents k timeout >> go- Cancelled -> return ()+runEventLoop k ct timeout =+ when rtsSupportsBoundThreads $ void $ forkIO go+ where+ go = do+ token <- CToken.status ct+ case token of+ Running -> pollEvents k timeout >> go+ Cancelled -> return () pollEvents :: HasKafka a => a -> Maybe Timeout -> IO () pollEvents a tm =
src/Kafka/Producer.hs view
@@ -36,7 +36,7 @@ -- | Runs Kafka Producer. -- The callback provided is expected to call 'produceMessage' -- or/and 'produceMessageBatch' to send messages to Kafka.-{-# DEPRECATED runProducer "Use 'newProducer'/'closeProducer' instead" #-}+{-# DEPRECATED runProducer "Use @newProducer@/@closeProducer@ instead" #-} runProducer :: ProducerProperties -> (KafkaProducer -> IO (Either KafkaError a)) -> IO (Either KafkaError a)@@ -54,18 +54,18 @@ -- | Creates a new kafka producer -- A newly created producer must be closed with 'closeProducer' function. newProducer :: MonadIO m => ProducerProperties -> m (Either KafkaError KafkaProducer)-newProducer (ProducerProperties kp tp ll cbs) = liftIO $ do- kc@(KafkaConf kc' ct) <- kafkaConf (KafkaProps $ M.toList kp)- tc <- topicConf (TopicProps $ M.toList tp)+newProducer pps = liftIO $ do+ kc@(KafkaConf kc' ct) <- kafkaConf (KafkaProps $ M.toList (ppKafkaProps pps))+ tc <- topicConf (TopicProps $ M.toList (ppTopicProps pps)) -- set callbacks- forM_ cbs (\setCb -> setCb kc)+ forM_ (ppCallbacks pps) (\setCb -> setCb kc) mbKafka <- newRdKafkaT RdKafkaProducer kc' case mbKafka of Left err -> return . Left $ KafkaError err Right kafka -> do- forM_ ll (rdKafkaSetLogLevel kafka . fromEnum)+ forM_ (ppLogLevel pps) (rdKafkaSetLogLevel kafka . fromEnum) let prod = KafkaProducer (Kafka kafka) kc tc runEventLoop prod ct (Just $ Timeout 100) >> return (Right prod)
src/Kafka/Producer/ProducerProperties.hs view
@@ -8,7 +8,6 @@ import qualified Data.List as L import Data.Map (Map) import qualified Data.Map as M-import Data.Semigroup import Kafka.Callbacks import Kafka.Internal.Setup import Kafka.Types@@ -18,17 +17,21 @@ { ppKafkaProps :: Map String String , ppTopicProps :: Map String String , ppLogLevel :: Maybe KafkaLogLevel- , callbacks :: [KafkaConf -> IO ()]+ , ppCallbacks :: [KafkaConf -> IO ()] } -- | /Right biased/ so we prefer newer properties over older ones.-instance Semigroup ProducerProperties where- (<>) (ProducerProperties k1 t1 ll1 cb1) (ProducerProperties k2 t2 ll2 cb2) =- ProducerProperties (M.union k2 k1) (M.union t2 t1) (ll2 `mplus` ll1) (cb2 `mplus` cb1)- instance Monoid ProducerProperties where- mempty = ProducerProperties M.empty M.empty Nothing []- mappend = (<>)+ mempty = ProducerProperties+ { ppKafkaProps = M.empty+ , ppTopicProps = M.empty+ , ppLogLevel = Nothing+ , ppCallbacks = []+ }+ {-# INLINE mempty #-}+ mappend (ProducerProperties k1 t1 ll1 cb1) (ProducerProperties k2 t2 ll2 cb2) =+ ProducerProperties (M.union k2 k1) (M.union t2 t1) (ll2 `mplus` ll1) (cb2 `mplus` cb1)+ {-# INLINE mappend #-} brokersList :: [BrokerAddress] -> ProducerProperties brokersList bs =@@ -36,12 +39,12 @@ in extraProps $ M.fromList [("bootstrap.servers", bs')] setCallback :: (KafkaConf -> IO ()) -> ProducerProperties-setCallback cb = ProducerProperties M.empty M.empty Nothing [cb]+setCallback cb = mempty { ppCallbacks = [cb] } -- | Sets the logging level. -- Usually is used with 'debugOptions' to configure which logs are needed. logLevel :: KafkaLogLevel -> ProducerProperties-logLevel ll = ProducerProperties M.empty M.empty (Just ll) []+logLevel ll = mempty { ppLogLevel = Just ll } compression :: KafkaCompressionCodec -> ProducerProperties compression c =@@ -54,7 +57,7 @@ -- | Any configuration options that are supported by /librdkafka/. -- The full list can be found <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md here> extraProps :: Map String String -> ProducerProperties-extraProps m = ProducerProperties m M.empty Nothing []+extraProps m = mempty { ppKafkaProps = m } -- | Suppresses producer disconnects logs. --@@ -67,7 +70,7 @@ -- | Any configuration options that are supported by /librdkafka/. -- The full list can be found <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md here> extraTopicProps :: Map String String -> ProducerProperties-extraTopicProps m = ProducerProperties M.empty m Nothing []+extraTopicProps m = mempty { ppTopicProps = m } -- | Sets debug features for the producer -- Usually is used with 'logLevel'.
src/Kafka/Types.hs view
@@ -6,13 +6,11 @@ import Control.Exception import Data.Int import Data.Typeable-import Kafka.Internal.CancellationToken import Kafka.Internal.RdKafka newtype BrokerId = BrokerId Int deriving (Show, Eq, Ord, Read)- newtype PartitionId = PartitionId Int deriving (Show, Eq, Read, Ord, Enum) newtype Millis = Millis Int64 deriving (Show, Read, Eq, Ord, Num)