{-# LANGUAGE DeriveDataTypeable #-}
{-# LANGUAGE RankNTypes #-}
import Haskakafka

import qualified Data.ByteString.Char8 as C8
import           Data.Maybe (fromMaybe)
import           Control.Monad (forM_)

import           System.Console.CmdArgs
import           Text.Show.Pretty (ppShow)


data BasicMode = Consumer | Producer | List | All
  deriving (Data, Typeable, Show, Eq)

data CArgs = CArgs
  { brokers'CArgs   :: Maybe String
  , topic'CArgs     :: Maybe String
  , mode'CArgs      :: BasicMode
  , partition'CArgs :: Maybe Int
  , pretty'CArgs :: Bool
  } deriving (Data, Typeable, Show, Eq)

producerExample ::
                String -> String -> Int ->
                (forall a. Show a => a -> String) -> IO ()
producerExample brokerNames ourTopic partition showFn = do
  let
      -- Optionally, we can configure certain parameters for Kafka
      kafkaConfig = [("socket.timeout.ms", "50000")]
      topicConfig = [("request.timeout.ms", "50000")]

      -- Payloads are just ByteStrings
      samplePayload = C8.pack "Hello world"

  -- withKafkaProducer opens a producer connection and gives us
  -- two objects for subsequent use.
  withKafkaProducer kafkaConfig topicConfig
                    brokerNames ourTopic
                    $ \kafka topic -> do

    -- Produce a single unkeyed message to partition
    let message = KafkaProduceMessage samplePayload
    _ <- produceMessage topic (KafkaSpecifiedPartition partition) message

    -- Produce a single keyed message
    let keyMessage = KafkaProduceKeyedMessage (C8.pack "Key") samplePayload
    _ <- produceKeyedMessage topic keyMessage

    -- We can also use the batch API for better performance
    let numMessages = 9
    _ <- produceMessageBatch topic
      KafkaUnassignedPartition
      (replicate numMessages message)

    putStrLn "Done producing messages, here was our config: "
    dumpConfFromKafka kafka >>= \d -> putStrLn $ "Kafka config: " ++ (showFn d)
    dumpConfFromKafkaTopic topic >>= \d -> putStrLn $ "Topic config: " ++ (showFn d)

consumerExample ::
                String -> String -> Int ->
                (forall a. Show a => a -> String) -> IO ()
consumerExample brokerNames ourTopic partition showFn = do
  let
      -- Optionally, we can configure certain parameters for Kafka
      kafkaConfig = [("socket.timeout.ms", "50000")]
      topicConfig = [("request.timeout.ms", "50000")]

  -- withKafkaConsumer opens a consumer connection and starts consuming
  withKafkaConsumer kafkaConfig topicConfig
                    brokerNames ourTopic
                    partition -- locked to a specific partition for each consumer
                    KafkaOffsetBeginning -- start reading from beginning
                    -- (alternatively, use
                    -- KafkaOffsetEnd, KafkaOffset or KafkaOffsetStored)
                    $ \kafka topic -> do

    -- Consume a single message at a time
    let timeoutMs = 3000
    me <- consumeMessage topic partition timeoutMs
    case me of
      (Left err) -> putStrLn $ "Uh oh, an error! " ++ (showFn err)
      (Right m) -> putStrLn $ "Woo, payload was " ++ (C8.unpack $ messagePayload m)

    -- For better performance, consume in batches
    let maxMessages = 10
    mes <- consumeMessageBatch topic partition timeoutMs maxMessages
    case mes of
      (Left err) -> putStrLn $
        "Something went wrong in batch consume! " ++ (showFn err)
      (Right ms) -> forM_ ms $ \msg -> do
        putStrLn $ "Woohoo, we got: " ++ (showFn msg)

    -- Be a little less noisy
    setLogLevel kafka KafkaLogCrit

metadataExample :: String -> (forall a. Show a => a -> String) -> IO ()
metadataExample brokerNames showFn = do
  -- we can also fetch metadata about our Kafka infrastructure
  let timeoutMs = 1000
  emd <- fetchBrokerMetadata [] brokerNames timeoutMs
  case emd of
    (Left err) -> putStrLn $ "Uh oh, error time: " ++ (showFn err)
    (Right md) -> putStrLn $ "Kafka metadata: " ++ (showFn md)

runExample ::
           BasicMode -> String -> String -> Int ->
           (forall a. (Show a) => a -> String) -> IO ()
runExample Consumer b t p pp = consumerExample b t p pp
runExample Producer b t p pp = producerExample b t p pp
runExample List     b _ _ pp = metadataExample b pp
runExample All      b t p pp = do
  -- consumerExample b t p pp
  producerExample b t p pp
  metadataExample b pp

parseExample :: CArgs -> IO ()
parseExample (CArgs b t m p pp) = runExample
  m
  (fromMaybe "localhost:9092" b)
  (fromMaybe "test_topic" t)
  (fromMaybe 0 p)
  (if pp then ppShow else show)

cargs :: CArgs
cargs = CArgs
  { brokers'CArgs = def
    &= typ "<brokers>"
    &= help "Comma separated list in format <hostname>:<port>,<hostname>:<port>"
    &= explicit
    &= name "brokers"
    &= name "b"
  , topic'CArgs = def
    &= typ "<topic>"
    &= help "Topic to fetch / produce"
    &= explicit
    &= name "topic"
    &= name "t"
  , mode'CArgs = enum
    [ Consumer &= help "Consumer mode" &= name "C"
    , Producer &= help "Producer mode" &= name "P"
    , List     &= help "Metadata list mode" &= name "L"
    , All      &= help "Run producer, consumer, and metadata list" &= name "A"
    ]
  , partition'CArgs = def
    &= typ "<num>"
    &= help "Partition (-1 for random partitioner when using producer)"
    &= explicit
    &= name "p"
  , pretty'CArgs = def
    &= help "Pretty print output"
    &= explicit
    &= name "pretty"
  } &= help "Fetch metadata, produce, and consume a message"
    &= program "basic example"

main :: IO ()
main = parseExample =<< cmdArgs cargs