diff --git a/example/ConsumerExample.hs b/example/ConsumerExample.hs --- a/example/ConsumerExample.hs +++ b/example/ConsumerExample.hs @@ -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 () diff --git a/hw-kafka-client.cabal b/hw-kafka-client.cabal --- a/hw-kafka-client.cabal +++ b/hw-kafka-client.cabal @@ -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 diff --git a/src/Kafka/Consumer.hs b/src/Kafka/Consumer.hs --- a/src/Kafka/Consumer.hs +++ b/src/Kafka/Consumer.hs @@ -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 diff --git a/src/Kafka/Consumer/ConsumerProperties.hs b/src/Kafka/Consumer/ConsumerProperties.hs --- a/src/Kafka/Consumer/ConsumerProperties.hs +++ b/src/Kafka/Consumer/ConsumerProperties.hs @@ -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 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 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. diff --git a/src/Kafka/Internal/Shared.hs b/src/Kafka/Internal/Shared.hs --- a/src/Kafka/Internal/Shared.hs +++ b/src/Kafka/Internal/Shared.hs @@ -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 = diff --git a/src/Kafka/Producer.hs b/src/Kafka/Producer.hs --- a/src/Kafka/Producer.hs +++ b/src/Kafka/Producer.hs @@ -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) diff --git a/src/Kafka/Producer/ProducerProperties.hs b/src/Kafka/Producer/ProducerProperties.hs --- a/src/Kafka/Producer/ProducerProperties.hs +++ b/src/Kafka/Producer/ProducerProperties.hs @@ -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 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 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'. diff --git a/src/Kafka/Types.hs b/src/Kafka/Types.hs --- a/src/Kafka/Types.hs +++ b/src/Kafka/Types.hs @@ -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)