Merge pull request #140 from Sir4ur0n/doc/KafkaConsumer

doc(kafka): Add documentation

Author: AlexeyRaga

src/Kafka/Consumer.hs

@@ -1,8 +1,58 @@
 {-# LANGUAGE LambdaCase        #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE TupleSections     #-}
+
+-----------------------------------------------------------------------------
+-- |
+-- Module to consume messages from Kafka topics.
+-- 
+-- Here's an example of code to consume messages from a topic:
+-- 
+-- @
+-- import Control.Exception (bracket)
+-- import Data.Monoid ((<>))
+-- import Kafka
+-- import Kafka.Consumer
+-- 
+-- -- Global consumer properties
+-- consumerProps :: 'ConsumerProperties'
+-- consumerProps = 'brokersList' ['BrokerAddress' "localhost:9092"]
+--              <> 'groupId' ('ConsumerGroupId' "consumer_example_group")
+--              <> 'noAutoCommit'
+--              <> 'logLevel' 'KafkaLogInfo'
+-- 
+-- -- Subscription to topics
+-- consumerSub :: 'Subscription'
+-- consumerSub = 'topics' ['TopicName' "kafka-client-example-topic"]
+--            <> 'offsetReset' 'Earliest'
+-- 
+-- -- Running an example
+-- runConsumerExample :: IO ()
+-- runConsumerExample = do
+--     res <- bracket mkConsumer clConsumer runHandler
+--     print res
+--     where
+--       mkConsumer = 'newConsumer' consumerProps consumerSub
+--       clConsumer (Left err) = return (Left err)
+--       clConsumer (Right kc) = (maybe (Right ()) Left) \<$\> 'closeConsumer' kc
+--       runHandler (Left err) = return (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 ()
+-- @
+-----------------------------------------------------------------------------
 module Kafka.Consumer
-( module X
+( KafkaConsumer
+, module X
 , runConsumer
 , newConsumer
 , assign, assignment, subscription
@@ -14,7 +64,6 @@ module Kafka.Consumer
 , storeOffsets, storeOffsetMessage
 , closeConsumer
 -- ReExport Types
-, KafkaConsumer
 , RdKafkaRespErrT (..)
 )
 where
@@ -48,7 +97,7 @@ import Kafka.Types                       as X
 
 -- | 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
@@ -64,6 +113,7 @@ runConsumer cp sub f =
     runHandler (Left err) = return (Left err)
     runHandler (Right kc) = f kc
 
+-- | Create a `KafkaConsumer`. This consumer must be correctly released using 'closeConsumer'.
 newConsumer :: MonadIO m
             => ConsumerProperties
             -> Subscription
@@ -94,6 +144,7 @@ newConsumer props (Subscription ts tp) = liftIO $ do
               runConsumerLoop kafka (Just $ Timeout 100)) >> return (Right kafka)
             Just err -> closeConsumer kafka >> return (Left err)
 
+-- | Polls a single message
 pollMessage :: MonadIO m
             => KafkaConsumer
             -> Timeout -- ^ the timeout, in milliseconds
@@ -104,7 +155,7 @@ pollMessage c@(KafkaConsumer _ (KafkaConf _ qr _)) (Timeout ms) = liftIO $ do
     Nothing -> rdKafkaConsumerPoll (getRdKafka c) ms >>= fromMessagePtr
     Just q  -> rdKafkaConsumeQueue q (fromIntegral ms) >>= fromMessagePtr
 
--- | Polls up to BatchSize messages.
+-- | Polls up to 'BatchSize' messages.
 -- Unlike 'pollMessage' this function does not return usual "timeout" errors.
 -- An empty batch is returned when there are no messages available.
 --
@@ -205,6 +256,7 @@ resumePartitions (KafkaConsumer (Kafka k) _) ps = liftIO $ do
   mapM_ (\(TopicName topicName, PartitionId partitionId) -> rdKafkaTopicPartitionListAdd pl (Text.unpack topicName) partitionId) ps
   KafkaResponseError <$> rdKafkaResumePartitions k pl
 
+-- | Seek a particular offset for each provided 'TopicPartition'
 seek :: MonadIO m => KafkaConsumer -> Timeout -> [TopicPartition] -> m (Maybe KafkaError)
 seek (KafkaConsumer (Kafka k) _) (Timeout timeout) tps = liftIO $
   either Just (const Nothing) <$> seekAll
@@ -244,7 +296,7 @@ position (KafkaConsumer (Kafka k) _) tps = liftIO $ do
 --
 -- Events will cause application provided callbacks to be called.
 --
--- The \p timeout_ms argument specifies the maximum amount of time
+-- The 'Timeout' argument specifies the maximum amount of time
 -- (in milliseconds) that the call will block waiting for events.
 --
 -- This function is called on each 'pollMessage' and, if runtime allows
@@ -260,6 +312,8 @@ pollConsumerEvents k timeout =
   void . withCallbackPollEnabled k $ pollConsumerEvents' k timeout
 
 -- | Closes the consumer.
+-- 
+-- See 'newConsumer'
 closeConsumer :: MonadIO m => KafkaConsumer -> m (Maybe KafkaError)
 closeConsumer (KafkaConsumer (Kafka k) (KafkaConf _ qr statusVar)) = liftIO $
   -- because closing the consumer will raise callbacks,
@@ -280,7 +334,7 @@ newConsumerConf ConsumerProperties {cpProps = m, cpCallbacks = cbs} = do
 
 -- | Subscribes to a given list of topics.
 --
--- Wildcard (regex) topics are supported by the librdkafka assignor:
+-- Wildcard (regex) topics are supported by the /librdkafka/ assignor:
 -- any topic name in the topics list that is prefixed with @^@ will
 -- be regex-matched to the full list of topics in the cluster and matching
 -- topics will be added to the subscription list.

src/Kafka/Consumer/Subscription.hs

@@ -17,6 +17,18 @@ import           Data.Text            (Text)
 import           Kafka.Consumer.Types (OffsetReset (..))
 import           Kafka.Types          (TopicName (..))
 
+-- | A consumer subscription to a topic.
+--
+-- ==== __Examples__
+--
+-- Typically you don't call the constructor directly, but combine settings:
+--
+-- @
+-- consumerSub :: 'Subscription'
+-- consumerSub = 'topics' ['TopicName' "kafka-client-example-topic"]
+--         <> 'offsetReset' 'Earliest'
+--         <> 'extraSubscriptionProps' (fromList [("prop1", "value 1"), ("prop2", "value 2")])
+-- @
 data Subscription = Subscription (Set TopicName) (Map Text Text)
 
 instance Sem.Semigroup Subscription where
@@ -32,15 +44,18 @@ instance Monoid Subscription where
   mappend = (Sem.<>)
   {-# INLINE mappend #-}
 
+-- | Build a subscription by giving the list of topic names only
 topics :: [TopicName] -> Subscription
 topics ts = Subscription (Set.fromList ts) M.empty
 
+-- | Build a subscription by giving the offset reset parameter only
 offsetReset :: OffsetReset -> Subscription
 offsetReset o =
   let o' = case o of
              Earliest -> "earliest"
              Latest   -> "latest"
    in Subscription Set.empty (M.fromList [("auto.offset.reset", o')])
 
+-- | Build a subscription by giving extra properties only
 extraSubscriptionProps :: Map Text Text -> Subscription
 extraSubscriptionProps = Subscription Set.empty

src/Kafka/Consumer/Types.hs

@@ -30,14 +30,17 @@ where
 
 import Data.Bifoldable      (Bifoldable (..))
 import Data.Bifunctor       (Bifunctor (..))
-import Data.Bitraversable   (Bitraversable (..), bimapM, bisequenceA)
+import Data.Bitraversable   (Bitraversable (..), bimapM, bisequence)
 import Data.Int             (Int64)
 import Data.Text            (Text)
 import Data.Typeable        (Typeable)
 import GHC.Generics         (Generic)
 import Kafka.Internal.Setup (HasKafka (..), HasKafkaConf (..), Kafka (..), KafkaConf (..))
 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.
 data KafkaConsumer = KafkaConsumer
   { kcKafkaPtr  :: !Kafka
   , kcKafkaConf :: !KafkaConf
@@ -51,8 +54,17 @@ instance HasKafkaConf KafkaConsumer where
   getKafkaConf = kcKafkaConf
   {-# INLINE getKafkaConf #-}
 
+-- | Consumer group ID. Different consumers with the same consumer group ID will get assigned different partitions of each subscribed topic. 
+-- 
+-- See <https://kafka.apache.org/documentation/#group.id Kafka documentation on consumer group>
 newtype ConsumerGroupId = ConsumerGroupId { unConsumerGroupId :: Text } deriving (Show, Ord, Eq, Generic)
+
+-- | A message offset in a partition
 newtype Offset          = Offset { unOffset :: Int64 } deriving (Show, Eq, Ord, Read, Generic)
+
+-- | Where to reset the offset when there is no initial offset in Kafka
+-- 
+-- See <https://kafka.apache.org/documentation/#auto.offset.reset Kafka documentation on offset reset>
 data OffsetReset        = Earliest | Latest deriving (Show, Eq, Generic)
 
 -- | A set of events which happen during the rebalancing process
@@ -67,6 +79,7 @@ data RebalanceEvent =
   | RebalanceRevoke [(TopicName, PartitionId)]
   deriving (Eq, Show, Generic)
 
+-- | The partition offset
 data PartitionOffset =
     PartitionOffsetBeginning
   | PartitionOffsetEnd
@@ -75,11 +88,13 @@ data PartitionOffset =
   | PartitionOffsetInvalid
   deriving (Eq, Show, Generic)
 
+-- | Partitions subscribed by a consumer
 data SubscribedPartitions
-  = SubscribedPartitions [PartitionId]
-  | SubscribedPartitionsAll
+  = SubscribedPartitions [PartitionId] -- ^ Subscribe only to those partitions
+  | SubscribedPartitionsAll            -- ^ Subscribe to all partitions
   deriving (Show, Eq, Generic)
 
+-- | Consumer record timestamp 
 data Timestamp =
     CreateTime !Millis
   | LogAppendTime !Millis
@@ -119,8 +134,8 @@ data ConsumerRecord k v = ConsumerRecord
   , crPartition :: !PartitionId  -- ^ Kafka partition this message was received from
   , crOffset    :: !Offset       -- ^ Offset within the 'crPartition' Kafka partition
   , crTimestamp :: !Timestamp    -- ^ Message timestamp
-  , crKey       :: !k
-  , crValue     :: !v
+  , crKey       :: !k            -- ^ Message key
+  , crValue     :: !v            -- ^ Message value
   }
   deriving (Eq, Show, Read, Typeable, Generic)
 
@@ -148,53 +163,56 @@ instance Bitraversable ConsumerRecord where
   bitraverse f g r = (\k v -> bimap (const k) (const v) r) <$> f (crKey r) <*> g (crValue r)
   {-# INLINE bitraverse #-}
 
+{-# DEPRECATED crMapKey "Isn't concern of this library. Use 'first'" #-}
 crMapKey :: (k -> k') -> ConsumerRecord k v -> ConsumerRecord k' v
 crMapKey = first
 {-# INLINE crMapKey #-}
 
+{-# DEPRECATED crMapValue "Isn't concern of this library. Use 'second'" #-}
 crMapValue :: (v -> v') -> ConsumerRecord k v -> ConsumerRecord k v'
 crMapValue = second
 {-# INLINE crMapValue #-}
 
+{-# DEPRECATED crMapKV "Isn't concern of this library. Use 'bimap'" #-}
 crMapKV :: (k -> k') -> (v -> v') -> ConsumerRecord k v -> ConsumerRecord k' v'
 crMapKV = bimap
 {-# INLINE crMapKV #-}
 
-{-# DEPRECATED sequenceFirst "Isn't concern of this library. Use 'bitraverse id pure'" #-}
+{-# DEPRECATED sequenceFirst "Isn't concern of this library. Use @'bitraverse' 'id' 'pure'@" #-}
 sequenceFirst :: (Bitraversable t, Applicative f) => t (f k) v -> f (t k v)
 sequenceFirst = bitraverse id pure
 {-# INLINE sequenceFirst #-}
 
-{-# DEPRECATED traverseFirst "Isn't concern of this library. Use 'bitraverse f pure'"  #-}
+{-# DEPRECATED traverseFirst "Isn't concern of this library. Use @'bitraverse' f 'pure'@"  #-}
 traverseFirst :: (Bitraversable t, Applicative f)
               => (k -> f k')
               -> t k v
               -> f (t k' v)
 traverseFirst f = bitraverse f pure
 {-# INLINE traverseFirst #-}
 
-{-# DEPRECATED traverseFirstM "Isn't concern of this library. Use 'bitraverse id pure <$> bitraverse f pure r'"  #-}
+{-# DEPRECATED traverseFirstM "Isn't concern of this library. Use @'bitraverse' 'id' 'pure' '<$>' 'bitraverse' f 'pure' r@"  #-}
 traverseFirstM :: (Bitraversable t, Applicative f, Monad m)
                => (k -> m (f k'))
                -> t k v
                -> m (f (t k' v))
 traverseFirstM f r = bitraverse id pure <$> bitraverse f pure r
 {-# INLINE traverseFirstM #-}
 
-{-# DEPRECATED traverseM "Isn't concern of this library. Use 'sequenceA <$> traverse f r'"  #-}
+{-# DEPRECATED traverseM "Isn't concern of this library. Use @'sequenceA' '<$>' 'traverse' f r@"  #-}
 traverseM :: (Traversable t, Applicative f, Monad m)
           => (v -> m (f v'))
           -> t v
           -> m (f (t v'))
 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 @'bisequenceA' '<$>' 'bimapM' f g r@"  #-}
 bitraverseM :: (Bitraversable t, Applicative f, Monad m)
             => (k -> m (f k'))
             -> (v -> m (f v'))
             -> t k v
             -> m (f (t k' v'))
-bitraverseM f g r = bisequenceA <$> bimapM f g r
+bitraverseM f g r = bisequence <$> bimapM f g r
 {-# INLINE bitraverseM #-}
 

src/Kafka/Producer.hs

@@ -41,7 +41,7 @@ import Kafka.Types                       as X
 -- | 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)