diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -13,9 +13,9 @@ High level consumers are supported by `librdkafka` starting from version 0.9. High-level consumers provide an abstraction for consuming messages from multiple -partitions and topics. They are also address scalability (up to a number of partitions) +partitions and topics. They also address scalability (up to a number of partitions) by providing automatic rebalancing functionality. When a new consumer joins a consumer -group the set of consumers attempt to "rebalance" the load to assign partitions to each consumer. +group the set of consumers attempts to "rebalance" the load to assign partitions to each consumer. ### Consumer example @@ -36,8 +36,6 @@ ```haskell import Control.Exception (bracket) -import Data.Monoid ((<>)) -import Kafka import Kafka.Consumer -- Global consumer properties @@ -67,12 +65,11 @@ ------------------------------------------------------------------- processMessages :: KafkaConsumer -> IO (Either KafkaError ()) processMessages kafka = do - mapM_ (\_ -> do - msg1 <- pollMessage kafka (Timeout 1000) - putStrLn $ "Message: " <> show msg1 - err <- commitAllOffsets OffsetCommit kafka - putStrLn $ "Offsets: " <> maybe "Committed." show err - ) [0 .. 10] + replicateM_ 10 $ do + msg <- pollMessage kafka (Timeout 1000) + putStrLn $ "Message: " <> show msg + err <- commitAllOffsets OffsetCommit kafka + putStrLn $ "Offsets: " <> maybe "Committed." show err return $ Right () ``` @@ -231,10 +228,10 @@ $ docker-compose up ``` -After that, tests can be run as usual: +After that, integration tests can switched on with using 'it' flag: ``` -$ cabal test --test-show-details=direct +$ cabal test --test-show-details=direct --flag it ``` ## Credits diff --git a/hw-kafka-client.cabal b/hw-kafka-client.cabal --- a/hw-kafka-client.cabal +++ b/hw-kafka-client.cabal @@ -1,7 +1,7 @@ cabal-version: 2.2 name: hw-kafka-client -version: 3.1.1 +version: 3.1.2 synopsis: Kafka bindings for Haskell description: Apache Kafka bindings backed by the librdkafka C library. . @@ -32,6 +32,11 @@ manual: True default: False +flag it + description: Run integration tests + manual: True + default: False + library hs-source-dirs: src ghc-options: -Wall @@ -103,6 +108,8 @@ , text , transformers build-tool-depends: hspec-discover:hspec-discover + if !(flag(it)) + buildable: False other-modules: Kafka.IntegrationSpec Kafka.TestEnv default-language: Haskell2010 diff --git a/src/Kafka/Callbacks.hs b/src/Kafka/Callbacks.hs --- a/src/Kafka/Callbacks.hs +++ b/src/Kafka/Callbacks.hs @@ -15,9 +15,9 @@ -- -- Basic usage: -- --- > setCallback (errorCallback myErrorCallback) +-- > 'setCallback' ('errorCallback' myErrorCallback) -- > --- > myErrorCallback :: KafkaError -> String -> IO () +-- > myErrorCallback :: 'KafkaError' -> String -> IO () -- > myErrorCallback kafkaError message = print $ show kafkaError <> "|" <> message errorCallback :: HasKafkaConf k => (KafkaError -> String -> IO ()) -> k -> IO () errorCallback callback k = @@ -30,9 +30,9 @@ -- -- Basic usage: -- --- > setCallback (logCallback myLogCallback) +-- > 'setCallback' ('logCallback' myLogCallback) -- > --- > myLogCallback :: KafkaLogLevel -> String -> String -> IO () +-- > myLogCallback :: 'KafkaLogLevel' -> String -> String -> IO () -- > myLogCallback level facility message = print $ show level <> "|" <> facility <> "|" <> message logCallback :: HasKafkaConf k => (KafkaLogLevel -> String -> String -> IO ()) -> k -> IO () logCallback callback k = @@ -45,7 +45,7 @@ -- -- Basic usage: -- --- > setCallback (statsCallback myStatsCallback) +-- > 'setCallback' ('statsCallback' myStatsCallback) -- > -- > myStatsCallback :: String -> IO () -- > myStatsCallback stats = print $ show stats diff --git a/src/Kafka/Consumer.hs b/src/Kafka/Consumer.hs --- a/src/Kafka/Consumer.hs +++ b/src/Kafka/Consumer.hs @@ -10,8 +10,7 @@ -- -- @ -- import Control.Exception (bracket) --- import Data.Monoid ((<>)) --- import Kafka +-- import Control.Monad (replicateM_) -- import Kafka.Consumer -- -- -- Global consumer properties @@ -33,21 +32,20 @@ -- print res -- where -- mkConsumer = 'newConsumer' consumerProps consumerSub --- clConsumer (Left err) = return (Left err) +-- clConsumer (Left err) = pure (Left err) -- clConsumer (Right kc) = (maybe (Right ()) Left) \<$\> 'closeConsumer' kc --- runHandler (Left err) = return (Left err) +-- runHandler (Left err) = pure (Left err) -- runHandler (Right kc) = processMessages kc -- -- -- Example polling 10 times before stopping -- processMessages :: 'KafkaConsumer' -> IO (Either 'KafkaError' ()) -- processMessages kafka = do --- mapM_ (\_ -> do --- msg1 <- 'pollMessage' kafka ('Timeout' 1000) --- putStrLn $ "Message: " <> show msg1 --- err <- 'commitAllOffsets' 'OffsetCommit' kafka --- putStrLn $ "Offsets: " <> maybe "Committed." show err --- ) [0 .. 10] --- return $ Right () +-- replicateM_ 10 $ do +-- msg <- 'pollMessage' kafka ('Timeout' 1000) +-- putStrLn $ "Message: " <> show msg +-- err <- 'commitAllOffsets' 'OffsetCommit' kafka +-- putStrLn $ "Offsets: " <> maybe "Committed." show err +-- pure $ Right () -- @ ----------------------------------------------------------------------------- module Kafka.Consumer @@ -159,7 +157,7 @@ -- Unlike 'pollMessage' this function does not return usual "timeout" errors. -- An empty batch is returned when there are no messages available. -- --- This API is not available when 'userPolls' is set. +-- This API is not available when 'CallbackPollMode' is set to 'CallbackPollModeSync'. pollMessageBatch :: MonadIO m => KafkaConsumer -> Timeout @@ -169,7 +167,7 @@ pollConsumerEvents c Nothing mbq <- readIORef qr case mbq of - Nothing -> return [Left $ KafkaBadSpecification "userPolls is set when calling pollMessageBatch."] + Nothing -> return [Left $ KafkaBadSpecification "Calling pollMessageBatch while CallbackPollMode is set to CallbackPollModeSync."] Just q -> rdKafkaConsumeBatchQueue q ms b >>= traverse fromMessagePtr -- | Commit message's offset on broker for the message's partition. diff --git a/src/Kafka/Consumer/Callbacks.hs b/src/Kafka/Consumer/Callbacks.hs --- a/src/Kafka/Consumer/Callbacks.hs +++ b/src/Kafka/Consumer/Callbacks.hs @@ -31,10 +31,10 @@ -- | Sets a callback that is called when rebalance is needed. -- -- The results of automatic or manual offset commits will be scheduled --- for this callback and is served by `pollMessage`. +-- for this callback and is served by 'Kafka.Consumer.pollMessage'. -- -- If no partitions had valid offsets to commit this callback will be called --- with `KafkaError` == `KafkaResponseError` `RdKafkaRespErrNoOffset` which is not to be considered +-- with 'KafkaResponseError' 'RdKafkaRespErrNoOffset' which is not to be considered -- an error. offsetCommitCallback :: (KafkaConsumer -> KafkaError -> [TopicPartition] -> IO ()) -> KafkaConf -> IO () offsetCommitCallback callback kc@(KafkaConf conf _ _) = rdKafkaConfSetOffsetCommitCb conf realCb 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 @@ -1,5 +1,9 @@ {-# LANGUAGE OverloadedStrings #-} +----------------------------------------------------------------------------- +-- | +-- Module with consumer properties types and functions. +----------------------------------------------------------------------------- module Kafka.Consumer.ConsumerProperties ( ConsumerProperties(..) , CallbackPollMode(..) @@ -34,9 +38,17 @@ import Kafka.Consumer.Callbacks as X -data CallbackPollMode = CallbackPollModeSync | CallbackPollModeAsync deriving (Show, Eq) +-- | Whether the callback polling should be done synchronously or not. +data CallbackPollMode = + -- | You have to poll the consumer frequently to handle new messages + -- as well as rebalance and keep alive events. + -- This enables lowering the footprint and having full control over when polling + -- happens, at the cost of manually managing those events. + CallbackPollModeSync + -- | Handle polling rebalance and keep alive events for you in a background thread. + | CallbackPollModeAsync deriving (Show, Eq) --- | Properties to create 'KafkaConsumer'. +-- | Properties to create 'Kafka.Consumer.Types.KafkaConsumer'. data ConsumerProperties = ConsumerProperties { cpProps :: Map Text Text , cpLogLevel :: Maybe KafkaLogLevel @@ -61,11 +73,13 @@ mappend = (Sem.<>) {-# INLINE mappend #-} +-- | Set the to contact to connect to the Kafka cluster. brokersList :: [BrokerAddress] -> ConsumerProperties brokersList bs = let bs' = Text.intercalate "," ((\(BrokerAddress x) -> x) <$> bs) in extraProps $ M.fromList [("bootstrap.servers", bs')] +-- | Set the and enables . autoCommit :: Millis -> ConsumerProperties autoCommit (Millis ms) = extraProps $ M.fromList @@ -73,81 +87,82 @@ , ("auto.commit.interval.ms", Text.pack $ show ms) ] --- | Disables auto commit for the consumer +-- | Disable for the consumer. noAutoCommit :: ConsumerProperties noAutoCommit = extraProps $ M.fromList [("enable.auto.commit", "false")] --- | Disables auto offset store for the consumer +-- | Disable auto offset store for the consumer. +-- +-- See for more information. noAutoOffsetStore :: ConsumerProperties noAutoOffsetStore = extraProps $ M.fromList [("enable.auto.offset.store", "false")] --- | Consumer group id +-- | Set the consumer . groupId :: ConsumerGroupId -> ConsumerProperties groupId (ConsumerGroupId cid) = extraProps $ M.fromList [("group.id", cid)] +-- | Set the . clientId :: ClientId -> ConsumerProperties clientId (ClientId cid) = extraProps $ M.fromList [("client.id", cid)] +-- | Set the consumer callback. +-- +-- For examples of use, see: +-- +-- * 'errorCallback' +-- * 'logCallback' +-- * 'statsCallback' setCallback :: (KafkaConf -> IO ()) -> ConsumerProperties setCallback cb = mempty { cpCallbacks = [cb] } --- | Sets the logging level. +-- | Set the logging level. -- Usually is used with 'debugOptions' to configure which logs are needed. logLevel :: KafkaLogLevel -> ConsumerProperties logLevel ll = mempty { cpLogLevel = Just ll } --- | Sets the compression codec for the consumer. +-- | Set the for the consumer. compression :: KafkaCompressionCodec -> ConsumerProperties compression c = extraProps $ M.singleton "compression.codec" (kafkaCompressionCodecToText c) --- | Suppresses consumer disconnects logs. +-- | Suppresses consumer . -- -- It might be useful to turn this off when interacting with brokers --- with an aggressive connection.max.idle.ms value. +-- with an aggressive @connection.max.idle.ms@ value. suppressDisconnectLogs :: ConsumerProperties suppressDisconnectLogs = extraProps $ M.fromList [("log.connection.close", "false")] --- | Any configuration options that are supported by /librdkafka/. +-- | Set any configuration options that are supported by /librdkafka/. -- The full list can be found extraProps :: Map Text Text -> ConsumerProperties extraProps m = mempty { cpProps = m } {-# INLINE extraProps #-} --- | Any configuration options that are supported by /librdkafka/. +-- | Set any configuration option that is supported by /librdkafka/. -- The full list can be found extraProp :: Text -> Text -> ConsumerProperties extraProp k v = mempty { cpProps = M.singleton k v } {-# INLINE extraProp #-} --- | Sets debug features for the consumer. --- Usually is used with 'consumerLogLevel'. +-- | Set features for the consumer. +-- Usually is used with 'logLevel'. debugOptions :: [KafkaDebug] -> ConsumerProperties debugOptions [] = extraProps M.empty debugOptions d = let points = Text.intercalate "," (kafkaDebugToText <$> d) in extraProps $ M.fromList [("debug", points)] +-- | Set queuedMaxMessagesKBytes :: Int -> ConsumerProperties queuedMaxMessagesKBytes kBytes = extraProp "queued.max.messages.kbytes" (Text.pack $ show kBytes) {-# INLINE queuedMaxMessagesKBytes #-} --- | Sets the callback poll mode. --- --- The default 'CallbackPollModeAsync' mode handles polling rebalance --- and keep alive events for you --- in a background thread. --- --- With 'CallbacPollModeSync' the user will poll the consumer --- frequently to handle new messages as well as rebalance and keep alive events. --- 'CallbacPollModeSync' lets you can simplify --- hw-kafka-client's footprint and have full control over when polling --- happens at the cost of having to manage this yourself. +-- | Set the callback poll mode. Default value is 'CallbackPollModeAsync'. callbackPollMode :: CallbackPollMode -> ConsumerProperties callbackPollMode mode = mempty { cpCallbackPollMode = mode } diff --git a/src/Kafka/Consumer/Subscription.hs b/src/Kafka/Consumer/Subscription.hs --- a/src/Kafka/Consumer/Subscription.hs +++ b/src/Kafka/Consumer/Subscription.hs @@ -1,5 +1,9 @@ {-# LANGUAGE OverloadedStrings #-} +----------------------------------------------------------------------------- +-- | +-- Module with subscription types and functions. +----------------------------------------------------------------------------- module Kafka.Consumer.Subscription ( Subscription(..) , topics @@ -26,8 +30,8 @@ -- @ -- consumerSub :: 'Subscription' -- consumerSub = 'topics' ['TopicName' "kafka-client-example-topic"] --- <> 'offsetReset' 'Earliest' --- <> 'extraSubscriptionProps' (fromList [("prop1", "value 1"), ("prop2", "value 2")]) +-- <> 'offsetReset' 'Earliest' +-- <> 'extraSubscriptionProps' (fromList [("prop1", "value 1"), ("prop2", "value 2")]) -- @ data Subscription = Subscription (Set TopicName) (Map Text Text) diff --git a/src/Kafka/Consumer/Types.hs b/src/Kafka/Consumer/Types.hs --- a/src/Kafka/Consumer/Types.hs +++ b/src/Kafka/Consumer/Types.hs @@ -1,5 +1,10 @@ {-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE DeriveGeneric #-} + +----------------------------------------------------------------------------- +-- | +-- Module holding consumer types. +----------------------------------------------------------------------------- module Kafka.Consumer.Types ( KafkaConsumer(..) , ConsumerGroupId(..) @@ -40,7 +45,7 @@ -- | The main type for Kafka consumption, used e.g. to poll and commit messages. -- --- Its constructor is intentionally not exposed, instead, one should used 'newConsumer' to acquire such a value. +-- Its constructor is intentionally not exposed, instead, one should use 'Kafka.Consumer.newConsumer' to acquire such a value. data KafkaConsumer = KafkaConsumer { kcKafkaPtr :: !Kafka , kcKafkaConf :: !KafkaConf @@ -207,7 +212,7 @@ traverseM f r = sequenceA <$> traverse f r {-# INLINE traverseM #-} -{-# DEPRECATED bitraverseM "Isn't concern of this library. Use @'bisequenceA' '<$>' 'bimapM' f g r@" #-} +{-# DEPRECATED bitraverseM "Isn't concern of this library. Use @'Data.Bitraversable.bisequenceA' '<$>' 'bimapM' f g r@" #-} bitraverseM :: (Bitraversable t, Applicative f, Monad m) => (k -> m (f k')) -> (v -> m (f v')) diff --git a/src/Kafka/Dump.hs b/src/Kafka/Dump.hs --- a/src/Kafka/Dump.hs +++ b/src/Kafka/Dump.hs @@ -1,3 +1,8 @@ +----------------------------------------------------------------------------- +-- | +-- Module providing various functions to dump information. These may be useful for +-- debug/investigation but should probably not be used on production applications. +----------------------------------------------------------------------------- module Kafka.Dump ( hPrintSupportedKafkaConf , hPrintKafka diff --git a/src/Kafka/Metadata.hs b/src/Kafka/Metadata.hs --- a/src/Kafka/Metadata.hs +++ b/src/Kafka/Metadata.hs @@ -1,5 +1,10 @@ {-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE OverloadedStrings #-} + +----------------------------------------------------------------------------- +-- | +-- Module with metadata types and functions. +----------------------------------------------------------------------------- module Kafka.Metadata ( KafkaMetadata(..), BrokerMetadata(..), TopicMetadata(..), PartitionMetadata(..) , WatermarkOffsets(..) diff --git a/src/Kafka/Producer.hs b/src/Kafka/Producer.hs --- a/src/Kafka/Producer.hs +++ b/src/Kafka/Producer.hs @@ -1,14 +1,67 @@ {-# LANGUAGE TupleSections #-} {-# LANGUAGE LambdaCase #-} + +----------------------------------------------------------------------------- +-- | +-- Module to produce messages to Kafka topics. +-- +-- Here's an example of code to produce messages to a topic: +-- +-- @ +-- import Control.Exception (bracket) +-- import Control.Monad (forM_) +-- import Data.ByteString (ByteString) +-- import Kafka.Producer +-- +-- -- Global producer properties +-- producerProps :: 'ProducerProperties' +-- producerProps = 'brokersList' ['BrokerAddress' "localhost:9092"] +-- <> 'logLevel' 'KafkaLogDebug' +-- +-- -- Topic to send messages to +-- targetTopic :: 'TopicName' +-- targetTopic = 'TopicName' "kafka-client-example-topic" +-- +-- -- Run an example +-- runProducerExample :: IO () +-- runProducerExample = +-- bracket mkProducer clProducer runHandler >>= print +-- where +-- mkProducer = 'newProducer' producerProps +-- clProducer (Left _) = pure () +-- clProducer (Right prod) = 'closeProducer' prod +-- runHandler (Left err) = pure $ Left err +-- runHandler (Right prod) = sendMessages prod +-- +-- -- Example sending 2 messages and printing the response from Kafka +-- sendMessages :: 'KafkaProducer' -> IO (Either 'KafkaError' ()) +-- sendMessages prod = do +-- err1 <- 'produceMessage' prod (mkMessage Nothing (Just "test from producer") ) +-- forM_ err1 print +-- +-- err2 <- 'produceMessage' prod (mkMessage (Just "key") (Just "test from producer (with key)")) +-- forM_ err2 print +-- +-- pure $ Right () +-- +-- mkMessage :: Maybe ByteString -> Maybe ByteString -> 'ProducerRecord' +-- mkMessage k v = 'ProducerRecord' +-- { 'prTopic' = targetTopic +-- , 'prPartition' = 'UnassignedPartition' +-- , 'prKey' = k +-- , 'prValue' = v +-- } +-- @ +----------------------------------------------------------------------------- module Kafka.Producer -( module X +( KafkaProducer +, module X , runProducer , newProducer , produceMessage, produceMessageBatch , produceMessage' , flushProducer , closeProducer -, KafkaProducer , RdKafkaRespErrT (..) ) where 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 @@ -1,5 +1,9 @@ {-# LANGUAGE OverloadedStrings #-} +----------------------------------------------------------------------------- +-- | +-- Module with producer properties types and functions. +----------------------------------------------------------------------------- module Kafka.Producer.ProducerProperties ( ProducerProperties(..) , brokersList @@ -27,7 +31,7 @@ import Kafka.Producer.Callbacks --- | Properties to create 'KafkaProducer'. +-- | Properties to create 'Kafka.Producer.Types.KafkaProducer'. data ProducerProperties = ProducerProperties { ppKafkaProps :: Map Text Text , ppTopicProps :: Map Text Text @@ -52,11 +56,19 @@ mappend = (Sem.<>) {-# INLINE mappend #-} +-- | Set the to contact to connect to the Kafka cluster. brokersList :: [BrokerAddress] -> ProducerProperties brokersList bs = let bs' = Text.intercalate "," ((\(BrokerAddress x) -> x) <$> bs) in extraProps $ M.fromList [("bootstrap.servers", bs')] +-- | Set the producer callback. +-- +-- For examples of use, see: +-- +-- * 'errorCallback' +-- * 'logCallback' +-- * 'statsCallback' setCallback :: (KafkaConf -> IO ()) -> ProducerProperties setCallback cb = mempty { ppCallbacks = [cb] } @@ -65,14 +77,17 @@ logLevel :: KafkaLogLevel -> ProducerProperties logLevel ll = mempty { ppLogLevel = Just ll } +-- | Set the for the producer. compression :: KafkaCompressionCodec -> ProducerProperties compression c = extraProps $ M.singleton "compression.codec" (kafkaCompressionCodecToText c) +-- | Set the for the topic. topicCompression :: KafkaCompressionCodec -> ProducerProperties topicCompression c = extraTopicProps $ M.singleton "compression.codec" (kafkaCompressionCodecToText c) +-- | Set the . sendTimeout :: Timeout -> ProducerProperties sendTimeout (Timeout t) = extraTopicProps $ M.singleton "message.timeout.ms" (Text.pack $ show t) @@ -90,12 +105,12 @@ suppressDisconnectLogs = extraProps $ M.fromList [("log.connection.close", "false")] --- | Any configuration options that are supported by /librdkafka/. --- The full list can be found +-- | Any *topic* configuration options that are supported by /librdkafka/. +-- The full list can be found extraTopicProps :: Map Text Text -> ProducerProperties extraTopicProps m = mempty { ppTopicProps = m } --- | Sets debug features for the producer +-- | Sets features for the producer -- Usually is used with 'logLevel'. debugOptions :: [KafkaDebug] -> ProducerProperties debugOptions [] = extraProps M.empty diff --git a/src/Kafka/Producer/Types.hs b/src/Kafka/Producer/Types.hs --- a/src/Kafka/Producer/Types.hs +++ b/src/Kafka/Producer/Types.hs @@ -2,6 +2,11 @@ {-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE DerivingStrategies #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} + +----------------------------------------------------------------------------- +-- | +-- Module holding producer types. +----------------------------------------------------------------------------- module Kafka.Producer.Types ( KafkaProducer(..) , ProducerRecord(..) @@ -18,7 +23,9 @@ import Kafka.Internal.Setup (HasKafka (..), HasKafkaConf (..), HasTopicConf (..), Kafka (..), KafkaConf (..), TopicConf (..)) import Kafka.Types (KafkaError (..), TopicName (..)) --- | Main pointer to Kafka object, which contains our brokers +-- | The main type for Kafka message production, used e.g. to send messages. +-- +-- Its constructor is intentionally not exposed, instead, one should used 'Kafka.Producer.newProducer' to acquire such a value. data KafkaProducer = KafkaProducer { kpKafkaPtr :: !Kafka , kpKafkaConf :: !KafkaConf @@ -45,8 +52,11 @@ , prValue :: Maybe ByteString } deriving (Eq, Show, Typeable, Generic) +-- | data ProducePartition = - SpecifiedPartition {-# UNPACK #-} !Int -- the partition number of the topic + -- | The partition number of the topic + SpecifiedPartition {-# UNPACK #-} !Int + -- | Let the Kafka broker decide the partition | UnassignedPartition deriving (Show, Eq, Ord, Typeable, Generic) @@ -54,8 +64,12 @@ newtype ImmediateError = ImmediateError KafkaError deriving newtype (Eq, Show) +-- | The result of sending a message to the broker, useful for callbacks data DeliveryReport + -- | The message was successfully sent at this offset = DeliverySuccess ProducerRecord Offset + -- | The message could not be sent | DeliveryFailure ProducerRecord KafkaError + -- | An error occurred, but /librdkafka/ did not attach any sent message | NoMessageError KafkaError deriving (Show, Eq, Generic) diff --git a/src/Kafka/Types.hs b/src/Kafka/Types.hs --- a/src/Kafka/Types.hs +++ b/src/Kafka/Types.hs @@ -2,6 +2,11 @@ {-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE OverloadedStrings #-} + +----------------------------------------------------------------------------- +-- | +-- Module holding types shared by consumer and producer modules. +----------------------------------------------------------------------------- module Kafka.Types ( BrokerId(..) , PartitionId(..) @@ -52,7 +57,7 @@ | System -- ^ Topics starting with a double underscore "\__" (@__consumer_offsets@, @__confluent.support.metrics@, etc.) are considered "system" topics deriving (Show, Read, Eq, Ord, Generic) --- | Topic name to be consumed +-- | Topic name to consume/produce messages -- -- Wildcard (regex) topics are supported by the /librdkafka/ assignor: -- any topic name in the topics list that is prefixed with @^@ will