"""Sends a message to a topic."""

from confluent_kafka import SerializingProducer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer
from confluent_kafka.serialization import StringSerializer
from lambdacommon.common_config import logger
import config


def delivery_report(err, msg):
    """Log delivery results from callbacks for each message produced."""
    if err is not None:
        logger.exception('Message delivery failed: {}'.format(err))
    else:
        print('Message delivered to {} [{}]'.format(msg.topic(), msg.partition()))


class KafkaProducerWithAvroSchema:
    """Event producer for a Kafka topic."""

    producer = None
    topic = None

    def __init__(self, topic):
        """Initialize the producer for this topic."""
        self.topic = topic
        schema_registry_client = SchemaRegistryClient({'url': config.SCHEMA_REGISTRY_URL})
        value_schema = schema_registry_client.get_latest_version(subject_name=f'{topic}-value')
        avro_serializer = AvroSerializer(
            schema_registry_client=schema_registry_client,
            schema_str=value_schema.schema.schema_str)
        producer_conf = {
            'bootstrap.servers': config.KAFKA_BROKERS,
            'security.protocol': 'SSL',
            'key.serializer': StringSerializer('utf_8'),
            'value.serializer': avro_serializer}
        self.producer = SerializingProducer(producer_conf)

    def produce_message(self, data, key=None):
        """Send message to topic."""
        self.producer.poll(0)
        self.producer.produce(
            topic=self.topic,
            value=data,
            key=key,
            on_delivery=delivery_report)
        self.producer.flush()