doc: Add documentation on Producer and remaining modules

Author: sir4ur0n

README.md

@@ -13,9 +13,9 @@ HaskellWorks Kafka ecosystem is described here: <https://github.com/haskell-work
 
 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 @@ To run an example please compile with the `examples` flag.
 
 ```haskell
 import Control.Exception (bracket)
-import Data.Monoid ((<>))
-import Kafka
 import Kafka.Consumer
 
 -- Global consumer properties
@@ -67,12 +65,11 @@ runConsumerExample = do
 -------------------------------------------------------------------
 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 ()
 ```
 

src/Kafka/Callbacks.hs

@@ -15,9 +15,9 @@ import Kafka.Types (KafkaError(..), KafkaLogLevel(..))
 --
 -- 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 @@ errorCallback callback k =
 --
 -- 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 @@ logCallback callback k =
 --
 -- Basic usage:
 --
--- > setCallback (statsCallback myStatsCallback)
+-- > 'setCallback' ('statsCallback' myStatsCallback)
 -- >
 -- > myStatsCallback :: String -> IO ()
 -- > myStatsCallback stats = print $ show stats

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 @@ pollMessage c@(KafkaConsumer _ (KafkaConf _ qr _)) (Timeout ms) = liftIO $ do
 -- 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 @@ pollMessageBatch c@(KafkaConsumer _ (KafkaConf _ qr _)) (Timeout ms) (BatchSize
   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.

src/Kafka/Consumer/Callbacks.hs

@@ -31,10 +31,10 @@ rebalanceCallback callback kc@(KafkaConf conf _ _) = rdKafkaConfSetRebalanceCb c
 -- | 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

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.Types          (BrokerAddress (..), ClientId (..), KafkaC
 
 import Kafka.Consumer.Callbacks as X
 
-data CallbackPollMode = CallbackPollModeSync | CallbackPollModeAsync deriving (Show, Eq)
-
--- | Properties to create 'KafkaConsumer'.
+-- | 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 'Kafka.Consumer.Types.KafkaConsumer'.
 data ConsumerProperties = ConsumerProperties
   { cpProps            :: Map Text Text
   , cpLogLevel         :: Maybe KafkaLogLevel
@@ -61,93 +73,96 @@ instance Monoid ConsumerProperties where
   mappend = (Sem.<>)
   {-# INLINE mappend #-}
 
+-- | Set the <https://kafka.apache.org/documentation/#bootstrap.servers list of brokers> 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 <https://kafka.apache.org/documentation/#auto.commit.interval.ms auto commit interval> and enables <https://kafka.apache.org/documentation/#enable.auto.commit auto commit>.
 autoCommit :: Millis -> ConsumerProperties
 autoCommit (Millis ms) = extraProps $
   M.fromList
     [ ("enable.auto.commit", "true")
     , ("auto.commit.interval.ms", Text.pack $ show ms)
     ]
 
--- | Disables auto commit for the consumer
+-- | Disable <https://kafka.apache.org/documentation/#enable.auto.commit auto commit> 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 <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md enable.auto.offset.store> for more information.
 noAutoOffsetStore :: ConsumerProperties
 noAutoOffsetStore =
   extraProps $ M.fromList [("enable.auto.offset.store", "false")]
 
--- | Consumer group id
+-- | Set the consumer <https://kafka.apache.org/documentation/#group.id group id>.
 groupId :: ConsumerGroupId -> ConsumerProperties
 groupId (ConsumerGroupId cid) =
   extraProps $ M.fromList [("group.id", cid)]
 
+-- | Set the <https://kafka.apache.org/documentation/#client.id consumer identifier>.
 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 <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md compression.codec> for the consumer.
 compression :: KafkaCompressionCodec -> ConsumerProperties
 compression c =
   extraProps $ M.singleton "compression.codec" (kafkaCompressionCodecToText c)
 
--- | Suppresses consumer disconnects logs.
+-- | Suppresses consumer <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md log.connection.close>.
 --
 -- 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 <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md here>
 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 <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md here>
 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 <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md debug> 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 <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md queued.max.messages.kbytes>
 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 }

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 @@ import           Kafka.Types          (TopicName (..))
 -- @
 -- 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)
 

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 @@ import Kafka.Types          (Millis (..), PartitionId (..), TopicName (..))
 
 -- | 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 :: (Traversable t, Applicative f, Monad m)
 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'))

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

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(..)