144 lines
5.3 KiB
Python
144 lines
5.3 KiB
Python
"""
|
|
|
|
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("MESSAGE:", json.to_string(message["value"]))
|
|
|
|
async def main():
|
|
await asyncio.gather(*[keep_producing(), keep_consuming()])
|
|
|
|
|
|
asyncio.run(main())
|