"""

    AUTHOR:

        Khushal P Soonderji

    DATE:

        Thursday, 19th Sept., 2024

    OBJECTIVE:

        To provide a quick way to test out Kafka.

    REFERENCES:

        N/A

    DOWNLOADS:

        N/A

"""


#   *****************************************************************************************************************
#  *****                                                                                                          ****
# ***                                                   IMPORT                                                      ***
#  *****                                                                                                          ****
#   *****************************************************************************************************************


# To make sibling directories accessible for imports:
import sys
sys.path.append(".")
sys.path.append("..")

# System-level:
import os

# Utils:
from utils_v2.string import json
from utils_v2.date_time import date_time
from utils_v2.queue.async_kafka import ProducerKafka, ConsumerKafka, get_ssl_context

# For async activities:
import asyncio

# Common:
from shared import constants


#   *****************************************************************************************************************
#  *****                                                                                                          ****
# ***                                           MACROS / ONE-TIME INIT                                              ***
#  *****                                                                                                          ****
#   *****************************************************************************************************************


# --- Nothing Yet


#   *****************************************************************************************************************
#  *****                                                                                                          ****
# ***                                                VARIABLES                                                      ***
#  *****                                                                                                          ****
#   *****************************************************************************************************************


# --- Nothing Yet


#   *****************************************************************************************************************
#  *****                                                                                                          ****
# ***                                                  FUNCTIONS                                                    ***
#  *****                                                                                                          ****
#   *****************************************************************************************************************


# --- Nothing Yet


#   *****************************************************************************************************************
#  *****                                                                                                          ****
# ***                                               MAIN PROGRAM                                                    ***
#  *****                                                                                                          ****
#   *****************************************************************************************************************


if __name__ == "__main__":

    # Define the test params:
    TOPIC = "kft_file_upload"
    BOOTSTRAP_SERVERS = "del.ditscentre.in:9092"
    SSL_CONTEXT = get_ssl_context(
        ca_file = os.path.join(constants.CREDENTIALS_DIRECTORY, "kafka", "cert_authority.pem"),
        cert_file = os.path.join(constants.CREDENTIALS_DIRECTORY, "kafka", "fullchain.pem"),
        key_file = os.path.join(constants.CREDENTIALS_DIRECTORY, "kafka", "privkey.pem")
    )

    async def keep_producing():

        # Create the producer:
        my_producer = ProducerKafka(
            topic = TOPIC,
            bootstrap_servers = BOOTSTRAP_SERVERS,
            security_protocol = "SSL",
            ssl_context = SSL_CONTEXT
        )

        # Create a message for the producer to produce:
        producer_message = {
            "message": "Testing! Testing! 1, 2, 3...",
            "ts": date_time.get_current_utc_date_time(as_string = True)
        }

        # Keep sending the message in intervals:
        while True:
            success = await my_producer.produce(producer_message)
            print("PRODUCED:", success)
            await asyncio.sleep(1.0)

    async def keep_consuming():

        # Create the consumer:
        my_consumer = ConsumerKafka(
            topic = TOPIC,
            bootstrap_servers = BOOTSTRAP_SERVERS,
            security_protocol = "SSL",
            ssl_context = SSL_CONTEXT
        )

        # Keep checking for messages:
        while True:
            await asyncio.sleep(1.0)
            messages = await my_consumer.consume(count = 10, timeout = 1.0)
            for message in messages: print(json.to_string(message["value"]))

    async def main():
        await asyncio.gather(*[keep_producing(), keep_consuming()])


    asyncio.run(main())