template <typename T>
struct Serialize
{
    std::vector<uint8_t> operator()(const T&);
}

template <typename T>
struct Deserialize
{
    T operator()(const cppkafka::TopicPartition&, const cppkafka::Buffer&);
}

//==========================================================================
//                       Serializers/De-serializers
//==========================================================================
//------------------------ serializers.h -----------------------------------
#include <corokafka/corokafka.h>

// Declare functions inside corokafka namespace...
namespace Bloomberg { namespace corokafka {
    // This serializer is used for the key
    template <>
    struct Serialize<size_t> {
        std::vector<uint8_t> operator()(const size_t& value) {
            return {reinterpret_cast<const uint8_t*>(&value),
                    reinterpret_cast<const uint8_t*>(&value) + sizeof(size_t)};
        }
    };
    
    // This serializer is used for the headers and the payload
    template <>
    struct Serialize<std::string> {
        std::vector<uint8_t> operator()(const std::string& value) {
            return {value.begin(), value.end()};
        }
    };
    
    // This is the deserializer for the key
    template <>
    struct Deserialize<size_t> {
        size_t operator()(const cppkafka::TopicPartition&, 
                          const cppkafka::Buffer& buffer) {
            return *static_cast<const size_t*>((void*)buffer.get_data());
        }
    };
    
    // This is the deserializer for the header and payload
    template <>
    struct Deserialize<std::string> {
        std::string operator()(const cppkafka::TopicPartition&, 
                               const cppkafka::Buffer& buffer) {
            return {buffer.begin(), buffer.end()};
        }
    };
}} //namespace Bloomberg::corokafka

//==========================================================================
//                                Topic
//==========================================================================
//------------------------------ mytopic.h ---------------------------------
#include <serializers.h>

using MyTopic = corokafka::Topic<size_t, std::string, coroakafka::Headers<std::string>>;

// Create a topic which will be shared between producer and consumer.
static MyTopic myTopic("my-topic", corokafka::Header<std::string>("Header1");

//==========================================================================
//                          Producer setup
//==========================================================================
#include <mytopic.h>

// Define a delivery report callback (optional - see documentation for all available callbacks)
void deliveryReportCallback(const corokafka::ProducerMetadata& metadata,
                            const corokafka::SentMessage& msg)
{
    std::cout << "Produced message for [" << metadata.getInternalName()
              << ":" << msg.getTopic() << "]"
              << std::endl;
    if (msg) {
        if (msg.getError()) {
            return;
        }
        //print message contents
        std::cout << "Message key: " << *reinterpret_cast<const size_t*>(msg.getKeyBuffer().get_data())
                  << " Message content: [" << (std::string)msg.getPayloadBuffer() << "]"
                  << " Message partition: " << msg.getPartition()
                  << " Message offset: " << msg.getOffset()
                  << std::endl;
    }
}

// Define a log callback (optional - see documentation for all available callbacks)
void logCallback(const corokafka::Metadata& metadata,
                 cppkafka::LogLevel level,
                 const std::string& facility,
                 const std::string& message)
{
    std::cout << "Log from: " << metadata.getInternalName()
              << " facility: " << facility
              << " text: " << message << std::endl;
}

// Create a topic configuration (optional)
std::initializer_list<cppkafka::ConfigurationOption > topicOptions = {
    { "produce.offset.report",  true },
    { "request.required.acks", -1 }
};

// Create a producer configuration (only 'metadata.broker.list' setting is mandatory)
std::initializer_list<cppkafka::ConfigurationOption > producerOptions = {
    { "metadata.broker.list", "broker_url:port" },
    { "api.version.request", true },
    { "internal.producer.retries", 5 }
};

// Associate the topic and producer configuration with a topic name.
// Note: any number of producer configs/topics can be created.
corokafka::ProducerConfiguration config(myTopic, producerOptions, topicOptions);

// Add the callbacks
config.setDeliveryReportCallback(deliveryReportCallback);
config.setLogCallback(logCallback);

// Create the connector config (optional)
corokafka::ConnectorConfiguration connectorConfiguration(
    {{ "internal.connector.poll.interval.ms", 100 }}
);

// Combine all the configs together
corokafka::ConfigurationBuilder builder;
builder(config).(connectorConfiguration);

// Create the connector
corokafka::Connector connector(std::move(builder));

// Produce 10 messages
corokafka::ProducerManager& producer = connector.producer(); //get handle on producer

// Produce 10 messages. 'i' represents the 'key'
for (size_t i = 0; i < 10; ++i) { 
    //produce a message synchronously (async production also available)
    producer.send(myTopic, 0, i, std::string("Hello world"), std::string("This is some header"));
}

//==========================================================================
//                 Message worker queue and processor
//==========================================================================
#include <mytopic.h>

std::mutex messageMutex;
std::deque<corokafka::ReceivedMessage<size_t, std::string>> messages;

void messageProcessor()
{
    while (1)
    {
        if (!messages.empty()) {
            std::lock_guard<std::mutex> lock(messageMutex);
            corokafka::ReceivedMessage<size_t, std::string> message = std::move(messages.front());
            messages.pop_front();
                       
            //get the headers
            const ckf::HeaderPack &headers = message.getHeaders();
            std::string headerName;
            std::string headerValue;
            if (headers) {
                headerName = headers.getAt<std::string>(0).name();
                //get header value (type unsafe)
                headerValue = headers.getAt<std::string>(0).value();
                //get header value (type-safe)
                //headerValue = message.getHeaderAt<0>();
            }
            std::ostringstream oss;
            oss << "Received message from topic: " << message.getTopic()
                << " partition: " << message.getPartition()
                << " offset: " << message.getOffset()
                << " key: " << message.getKey()
                << " header: " << headerName << " :: " << headerValue
                << " payload: " << message.getPayload()
                << std::endl;
            std::cout << oss.str() << std::endl;
            
            //commit the message. This can also be done automatically (see consumer configuration).
            message.commit();
        }
        // Pause a little. Typically this could be implemented with a condition variable being signalled
        // when a message gets enqueued.
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
    }
}

//==========================================================================
//                         Consumer setup
//==========================================================================
#include <corokafka/corokafka.h>

// Define a log callback (optional - see documentation for all available callbacks)
void logCallback(const corokafka::Metadata& metadata,
                 cppkafka::LogLevel level,
                 const std::string& facility,
                 const std::string& message)
{
    std::cout << "Log from: " << metadata.getInternalName()
              << " facility: " << facility
              << " text: " << message << std::endl;
}

// Define a receiver callback (mandatory)
void receiverCallback(MyTopic::receivedMessage message)
{
    if (!message) return; //invalid message
     
    if (message.getError()) {
        std::cout << "Enqueued message from topic: " << message.getTopic()
                  << " with error: " << message.getError().to_string()
                  << std::endl;
    }
    else {
        std::cout << "Enqueued message from topic: " << message.getTopic()
                  << " partition: " << message.getPartition()
                  << " offset: " << message.getOffset()
                  << " key: " << message.getKey()
                  << " payload: " << (std::string)message.getPayload()
                  << std::endl;
    }

    // Post message unto a worker queue...
    std::lock_guard<std::mutex> lock(messageMutex);
    messages.emplace_back(std::move(message));
}

// Create the producer and topic config
std::initializer_list<cppkafka::ConfigurationOption > consumerOptions = {
    { "metadata.broker.list", "broker_url:port" },
    { "group.id", "my-group" },
    { "api.version.request", true },
    { "enable.partition.eof", true },
    { "enable.auto.commit", true },
    { "auto.commit.interval.ms", 1000 },
    { "auto.offset.reset", "earliest" },
    { "partition.assignment.strategy", "roundrobin" },
    { "internal.consumer.pause.on.start", false }
};

//create the consumer configuration
corokafka::ConsumerConfiguration config(myTopic, consumerOptions, {}, receiverCallback);

//add the callbacks
config.setLogCallback(logCallback);

// Optionally set initial partition assignment (4 partitions per topic)
config.assignInitialPartitions(corokafka::PartitionStrategy::Dynamic,
    {
    {topic, 0, cppkafka::TopicPartition::OFFSET_BEGINNING},
    {topic, 1, cppkafka::TopicPartition::OFFSET_BEGINNING},
    {topic, 2, cppkafka::TopicPartition::OFFSET_BEGINNING},
    {topic, 3, cppkafka::TopicPartition::OFFSET_BEGINNING}
    });

// Create the connector (this will subscribe all consumers and start receiving messages)
corokafka::Connector connector(corokafka::ConfigurationBuilder(config));

// Start processing messages in the background...
std::thread t(messageProcessor);