#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Copyright 2015 Cyan, Inc.
# Copyright 2018 Ciena Corporation

import sys
import logging

from twisted.internet import defer, task

from afkak.client import KafkaClient
from afkak.consumer import OFFSET_LATEST, Consumer

log = logging.getLogger('consumer_example')


@defer.inlineCallbacks
def ready_client(reactor, netloc, topic):
    """
    Connect to a Kafka broker and wait for the named topic to exist.
    This assumes that ``auto.create.topics.enable`` is set in the broker
    configuration.

    :raises: `KafkaUnavailableError` if unable to connect.
    """
    client = KafkaClient(netloc, reactor=reactor)

    e = True
    while e:
        yield client.load_metadata_for_topics(topic)
        e = client.metadata_error_for_topic(topic)
        if e:
            log.info("Error getting metadata for topic %r: %s (will retry)",
                     topic, e)

    defer.returnValue(client)


@defer.inlineCallbacks
def consume(reactor, hosts='localhost:9092'):
    topic = b'example_topic'
    client = yield ready_client(reactor, hosts, topic)
    partitions = client.topic_partitions[topic]

    def process(consumer, message_list):
        """
        This function is called for every batch of messages received from
        Kafka. It may return a Deferred, but this implementation just logs the
        messages received.
        """
        for m in message_list:
            log.info("Got message %r", m)

    consumers = [Consumer(client, topic, partition, process)
                 for partition in partitions]

    def cb_closed(result):
        """
        Called when a consumer cleanly stops.
        """
        log.info("Consumer stopped")

    def eb_failed(failure):
        """
        Called when a consumer fails due to an uncaught exception in the
        processing callback or a network error on shutdown. In this case we
        simply log the error.
        """
        log.error("Consumer failed: %s", failure)

    def stop_consumers():
        log.info("\n")
        log.info("Time is up, stopping consumers...")
        d = defer.gatherResults([c.shutdown() for c in consumers])
        d.addCallback(lambda result: client.close())
        return d

    yield defer.gatherResults(
        [c.start(OFFSET_LATEST).addCallbacks(cb_closed, eb_failed) for c in consumers] +
        [task.deferLater(reactor, 60.0, stop_consumers)]
    )


def main():
    logging.basicConfig(
        format='%(name)s %(levelname)s %(message)s',
        level=logging.INFO,
    )
    task.react(consume, sys.argv[1:])
    log.info("All Done!")


if __name__ == "__main__":
    main()
