content: module KafkaConsumer where
  
  {-| Author: Pablo Perez Garcia
      This code example is build in top of hw-kafka-client library https://hackage.haskell.org/package/hw-kafka-client
      Before start codgin remember that you need install in your system [librdkafka]
      git clone https://github.com/edenhill/librdkafka
      cd librdkafka
      ./configure
      make
      sudo make install
  -}
  import Control.Arrow     ((&&&))
  import Control.Exception (bracket)
  import Data.Monoid       ((<>))
  import Kafka.Consumer
  import qualified Data.Text as TEXT
  import Data.ByteString
  import qualified Data.ByteString.Char8 as CH
  import qualified System.IO as SIO
  import ConfigurationUtils
  import Text.Read (lift)
  import Control.Concurrent (threadDelay)
  
  {-| IO monad to run the consumer, using the function [newConsumer] where we pass the ConsumerProperties and the Subscription
      it return a Either of [KafkaError] or [KafkaConsumer] -}
  startConsumer :: IO ()
  startConsumer = do consumerSubscription <- createConsumerSubscription
                     consumerProps <- createConsumerProp
                     kafkaConsumerEither <- newConsumer consumerProps consumerSubscription
                     response <- case kafkaConsumerEither of
                                Left err -> do return $ Left err
                                Right kafkaConsumer -> consumeMessages kafkaConsumer
                     print response
  
  {-| We use [ConsumerProperties] adding information of the broker host, consumer group id, commit strategy and log level-}
  createConsumerProp :: IO ConsumerProperties
  createConsumerProp = do  groupIdName <- getGroupId
                           bootstrapServer <- getBootstrapServer
                           return $ brokersList [BrokerAddress bootstrapServer]
                                  <> groupId (ConsumerGroupId groupIdName)
                                  <> noAutoCommit
                                  <> logLevel KafkaLogInfo
  
  {-| Create a subscription using [createTopic] functions and is merged using [<>] operator with another subscription created
      when we run [OffsetReset] function to define the Offset strategy -}
  createConsumerSubscription :: IO Subscription
  createConsumerSubscription = do topic <- getTopic
                                  return $ createTopic topic <> offsetReset Earliest
  
  {-| Create a subscription using [topics] functions which expect a [TopicName] type created passing a topic namex -}
  createTopic:: String  -> Subscription
  createTopic topic = topics [TopicName topic]
  
  {-| Having a Kafka consumer We use [mapM_] function to start iterating 15 times, polling for messages.
      We then commit offset using [commitAllOffsets] which return a maybe of Kafka error which in case
      is filled contains the error description
      In Haskell as a declarative paradigm language has not loops operators, so in order to iterate to
      get more message from the topic we do a recursive call of the function.-}
  consumeMessages :: KafkaConsumer -> IO (Either KafkaError String)
  consumeMessages kafkaConsumer = do message <- consumeMessage kafkaConsumer
                                     SIO.putStrLn $ "Message: " <> show message
                                     maybeError <- commitAllOffsets OffsetCommit kafkaConsumer
                                     SIO.putStrLn $ "Offsets: " <> maybe "Committed." show maybeError
                                     SIO.putStrLn $ "---------------------------------------"
                                     _ <- consumeMessages kafkaConsumer -- Recursive call.
                                     _ <- closeConsumer kafkaConsumer
                                     return $ Right "All events processed"
  
  {-| Having a Kafka consumer We use [pollMessage] function passing the consumer and a timeout.
      It return an either of KafkaError or ConsumerRecord..-}
  consumeMessage :: KafkaConsumer -> IO (Either KafkaError String)
  consumeMessage kafkaConsumer = do eitherConsumerRecord <- pollMessage kafkaConsumer (Timeout 5000)
                                    eitherResponse <- case eitherConsumerRecord of
                                                Right consumerRecord -> getConsumerRecordValue consumerRecord
                                    return eitherResponse
  
  {-| Function to extract the value from the ConsumerRecord -}
  getConsumerRecordValue :: ConsumerRecord (Maybe ByteString) (Maybe ByteString) -> IO (Either KafkaError String)
  getConsumerRecordValue(ConsumerRecord _ _ _ _ _ value) = case value of
                                                    Just value -> return $ Right $ CH.unpack value
                                                    Nothing -> return $ Left $ KafkaError "No data find in Customer record"
  
  