145 lines
5.3 KiB
Python
145 lines
5.3 KiB
Python
"""
|
|
|
|
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())
|