155 lines
5.6 KiB
Python
155 lines
5.6 KiB
Python
"""
|
|
|
|
AUTHOR:
|
|
|
|
Khushal P Soonderji
|
|
|
|
DATE:
|
|
|
|
Saturday, 28th Dec., 2024
|
|
|
|
OBJECTIVE:
|
|
|
|
To provide a quick way to test out passthrough messages over 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.system import files
|
|
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
|
|
|
|
# For random choices:
|
|
import random
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** MACROS / ONE-TIME INIT ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** VARIABLES ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** FUNCTIONS ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** MAIN PROGRAM ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
# Define the test params:
|
|
TOPIC = "socket-io-bcast"
|
|
BOOTSTRAP_SERVERS = "del.ditscentre.in:9092"
|
|
cwd = files.get_cwd()
|
|
pdir = files.get_parent_directory(cwd, depth = 2)
|
|
SSL_CONTEXT = ssl_context = get_ssl_context(
|
|
ca_file = os.path.join(pdir, "creds", "kafka", "cert_authority.pem"),
|
|
cert_file = os.path.join(pdir, "creds", "kafka", "fullchain.pem"),
|
|
key_file = os.path.join(pdir, "creds", "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:
|
|
strategy_message = {
|
|
"to": "FC9O3N75Ax7B1v4NAAAD",
|
|
"event": "Strategy",
|
|
"namespace": "/finstitutions/trading",
|
|
"data": {
|
|
"message": "Your strategy ABC says buy XYZ.",
|
|
"playSound": True
|
|
}
|
|
}
|
|
|
|
corporate_action_message = {
|
|
"to": "FC9O3N75Ax7B1v4NAAAD",
|
|
"event": "Corporate Action",
|
|
"namespace": "/finstitutions/trading",
|
|
"data": {
|
|
"message": "Stock ABC has a corporate action tomorrow.",
|
|
"playSound": True,
|
|
"date": "29-Dec-2024",
|
|
"action": "Extraordinary Board Meeting"
|
|
}
|
|
}
|
|
|
|
# Keep sending the message in intervals:
|
|
while True:
|
|
producer_message = random.choice([
|
|
strategy_message,
|
|
corporate_action_message
|
|
])
|
|
success = await my_producer.produce(producer_message)
|
|
print("PRODUCED:", success)
|
|
await asyncio.sleep(1.0)
|
|
|
|
async def main():
|
|
await asyncio.gather(*[keep_producing()])
|
|
|
|
|
|
asyncio.run(main())
|