"""

    AUTHOR:

        Khushal P Soonderji

    DATE:

        Tuesday, 24th Dec., 2024

    OBJECTIVE:

        To provide a quick way to empty the contents of a Kafka topic to make space in the buffer.

    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

# For random values:
import random

# 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_consuming():

        # Generate a random id:
        random_id = ""
        for _ in range(8): random_id += random.choice([
            '0', '1', '2', '3',
            '4', '5', '6', '7',
            '8', '9', 'a', 'b',
            'c', 'd', 'e', 'f'
        ])

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

        # If the connection attempt fails:
        if not connected:
            print(f"id-{random_id}: CONNECTION FAILED :(")
            return

        # Keep checking for messages:
        print(f"id-{random_id}: Connected.")
        while True:
            await asyncio.sleep(1.0)
            messages = await my_consumer.consume(count = 100, timeout = 1.0)
            print(f"id-{random_id}: Flushed {len(messages)} message(s).")

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


    asyncio.run(main())