content: module Haskakafka.ConsumerExample
  
  where
  
  import           Control.Arrow                  ((&&&))
  import           Haskakafka
  import           Haskakafka.Consumer
  import           Haskakafka.InternalRdKafkaEnum ()
  
  iterator :: [Integer]
  iterator = [0 .. 20]
  
  runConsumerExample :: IO ()
  runConsumerExample = do
      res <- runConsumer
                (ConsumerGroupId "test_group")
                []
                (BrokersString "localhost:9092")
                [TopicName "^hl-test*"]
                processMessages
      print $ show res
  
  consumerExample :: IO ()
  consumerExample = do
      print "creating kafka conf"
      conf <- newKafkaConsumerConf (ConsumerGroupId "test_group") []
  
      -- unnecessary, demo only
      setRebalanceCallback conf printingRebalanceCallback
  
      res <- runConsumerConf
                 conf
                 (BrokersString "localhost:9092")
                 [TopicName "^hl-test*"]
                 processMessages
  
      print $ show res
  
  -------------------------------------------------------------------
  processMessages :: Kafka -> IO (Either KafkaError ())
  processMessages kafka = do
      mapM_ (\_ -> do
                     msg1 <- pollMessage kafka 1000
                     print $ show msg1) iterator
      return $ Right ()
  
  printingRebalanceCallback :: Kafka -> KafkaError -> [KafkaTopicPartition] -> IO ()
  printingRebalanceCallback k e ps = do
      print $ show e
      print "partitions: "
      mapM_ (print . show . (ktpTopicName &&& ktpPartition &&& ktpOffset)) ps
      case e of
          KafkaResponseError RdKafkaRespErrAssignPartitions -> do
              err <- assign k ps
              print $ "Assign result: " ++ show err
          KafkaResponseError RdKafkaRespErrRevokePartitions -> do
              err <- assign k []
              print $ "Revoke result: " ++ show err
          x ->
              print "UNKNOWN (and unlikely!)" >> print (show x)
  
  
  