diff --git a/background/finstitutions/trading/json.py b/background/finstitutions/trading/json.py deleted file mode 100644 index 2f51391..0000000 --- a/background/finstitutions/trading/json.py +++ /dev/null @@ -1,87 +0,0 @@ -""" - - AUTHOR: - - Khushal P Soonderji - - DATE: - - Monday, 30th Dec. 2024 - - OBJECTIVE: - - To provide an easy way to work with '.json' data and files. - - REFERENCES: - - N/A - - DOWNLOADS: - - N/A - -""" - - -# ***************************************************************************************************************** -# ***** **** -# *** IMPORT *** -# ***** **** -# ***************************************************************************************************************** - - -# To make sibling directories accessible for imports: -import sys -sys.path.append(".") -sys.path.append("..") - -# System-level activities: -import io - -# To work with the JSON standard: -import json - -# To work with files: -from utils_v2.system import files - - -# ***************************************************************************************************************** -# ***** **** -# *** MACROS / ONE-TIME INIT *** -# ***** **** -# ***************************************************************************************************************** - - -# --- Nothing Yet - - -# ***************************************************************************************************************** -# ***** **** -# *** VARIABLES *** -# ***** **** -# ***************************************************************************************************************** - - -# --- Nothing Yet - - -# ***************************************************************************************************************** -# ***** **** -# *** FUNCTIONS *** -# ***** **** -# ***************************************************************************************************************** - - -# --- Nothing Yet - - -# ***************************************************************************************************************** -# ***** **** -# *** MAIN PROGRAM *** -# ***** **** -# ***************************************************************************************************************** - - -if __name__ == "__main__": - - pass diff --git a/background/finstitutions/trading/to_kafka.py b/background/finstitutions/trading/to_kafka.py index ef2a688..84dd741 100644 --- a/background/finstitutions/trading/to_kafka.py +++ b/background/finstitutions/trading/to_kafka.py @@ -43,7 +43,7 @@ import os # My utils: from utils_v2.string import json from utils_v2.system import files -from utils_v2.queue.kafka import ProducerKafka, ConsumerKafka, create_config +from utils_v2.queue.kafka import ProducerKafka, ConsumerKafka # To make HTTP calls: import httpx @@ -77,6 +77,9 @@ from kiteconnect import KiteConnect, KiteTicker # ***************************************************************************************************************** +# For user management: +accounts = {} + # For Zerodha and related to ticks: INSTRUMENT_TOKENS = [] INSTRUMENT_LOOKUP = {} @@ -85,32 +88,31 @@ INSTRUMENT_LOOKUP = {} cwd = files.get_cwd() parent_dir = cwd kafka_consumer = ConsumerKafka( - topic = "tickers", - config = create_config( + topic = "tick-listners", + config = ConsumerKafka.create_config( bootstrap_servers = "del.ditscentre.in:9092", - # buffer_memory = 3_35_54_432, + group_id = "123", security_protocol = "SSL", - # ca_file = r"../../creds/kafka/cert_authority.pem", - # cert_file = r"../../creds/kafka/fullchain.pem", - # key_file = r"../../creds/kafka/privkey.pem" - ca_file = os.path.join(parent_dir, "creds", "kafka", "cert_authority.pem"), - cert_file = os.path.join(parent_dir, "creds", "kafka", "fullchain.pem"), - key_file = os.path.join(parent_dir, "creds", "kafka", "privkey.pem") + ca_file = r"../../../creds/kafka/cert_authority.pem", + cert_file = r"../../../creds/kafka/fullchain.pem", + key_file = r"../../../creds/kafka/privkey.pem" + # ca_file = os.path.join(parent_dir, "creds", "kafka", "cert_authority.pem"), + # cert_file = os.path.join(parent_dir, "creds", "kafka", "fullchain.pem"), + # key_file = os.path.join(parent_dir, "creds", "kafka", "privkey.pem") ), debug = False ) kafka_producer = ProducerKafka( topic = "tickers", - config = create_config( + config = ProducerKafka.create_config( bootstrap_servers = "del.ditscentre.in:9092", - # buffer_memory = 3_35_54_432, security_protocol = "SSL", - # ca_file = r"../../creds/kafka/cert_authority.pem", - # cert_file = r"../../creds/kafka/fullchain.pem", - # key_file = r"../../creds/kafka/privkey.pem" - ca_file = os.path.join(parent_dir, "creds", "kafka", "cert_authority.pem"), - cert_file = os.path.join(parent_dir, "creds", "kafka", "fullchain.pem"), - key_file = os.path.join(parent_dir, "creds", "kafka", "privkey.pem") + ca_file = r"../../../creds/kafka/cert_authority.pem", + cert_file = r"../../../creds/kafka/fullchain.pem", + key_file = r"../../../creds/kafka/privkey.pem" + # ca_file = os.path.join(parent_dir, "creds", "kafka", "cert_authority.pem"), + # cert_file = os.path.join(parent_dir, "creds", "kafka", "fullchain.pem"), + # key_file = os.path.join(parent_dir, "creds", "kafka", "privkey.pem") ), debug = False ) @@ -145,8 +147,6 @@ def to_kafka(tick: TradingTick) -> bool: success = False summary = tick.summary - print(json.to_string(summary)) - print(json.to_string(tick.model_dump(), default=str)) summary["messageType"] = "ticks" success = kafka_producer.produce(value = summary) if not success: @@ -187,14 +187,14 @@ def on_ticks(ws, ticks): # --------------------------------------------------------------------------------------------------------------------- -def main(): +def init(): # Global vars: global INSTRUMENT_TOKENS global INSTRUMENT_LOOKUP # Load Zerodha credentials: - creds = json.from_file(r"../../creds/zerodha/api.json") + creds = json.from_file(r"../../../creds/zerodha/api.json") # creds = json.from_file(os.path.join(parent_dir, "creds", "zerodha", "api.json")) api_key = creds["apiKey"] access_token = creds["accessToken"] @@ -250,6 +250,36 @@ def main(): kite_ws.connect(threaded = True) +# --------------------------------------------------------------------------------------------------------------------- + + +def main(): + + while True: + + messages = kafka_consumer.consume(count = 1, timeout = 5.0) + for message in messages: + + token_key = message["tokenKey"] + account = None + + # Create the websocket: + if accounts[token_key] not in list(accounts.keys()): + pass + + account = accounts[token_key] + + # Subscribe/unsubscribe: + ws = account["ws"] + if message["action"] == "subscribe": + ws.subscribe(message["brokerTokens"]) + ws.set_mode(ws.MODE_FULL, message["brokerTokens"]) + pass + elif message["action"] == "unsubscribe": + ws.unsubscribe(message["brokerTokens"]) + pass + + # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** @@ -259,5 +289,5 @@ def main(): if __name__ == "__main__": + init() main() - while True: time.sleep(3_600.00) diff --git a/models/core/auth_token.py b/models/core/auth_token.py index 067f252..a61ae3d 100644 --- a/models/core/auth_token.py +++ b/models/core/auth_token.py @@ -173,10 +173,16 @@ class CoreAuthTokenModel(BaseModel): default = "pending" ) - syncFreq: Literal[60, 300, 900, 1500] = Field( + syncFreq: Literal[None, 60, 300, 900, 1500] = Field( description = "the no. of seconds after which to poll for updates from the client (if applicable)", frozen = False, - default = 300 + default = None + ) + + lastSyncTs: AwareDatetime | None = Field( + description = "the time at which this client's updates were last polled", + frozen = False, + default = None ) # ┏┓ ┏• diff --git a/models/finstitutions/trading/ticks.py b/models/finstitutions/trading/ticks.py index 162924d..4805d74 100644 --- a/models/finstitutions/trading/ticks.py +++ b/models/finstitutions/trading/ticks.py @@ -375,7 +375,9 @@ class TradingTick(BaseModel): "qty": self.qty, "chg": self.chg, "pChg": self.pChg, - "totVol": self.totVol + "totVol": self.totVol, + "tradeTs": self.tradeTs.timestamp(), + "tradeTz": self.tradeTz } @staticmethod diff --git a/playground/socketio/to_kafka.py b/playground/socketio/to_kafka.py index cc267c3..6abfda2 100644 --- a/playground/socketio/to_kafka.py +++ b/playground/socketio/to_kafka.py @@ -43,7 +43,7 @@ import os # My utils: from utils_v2.string import json from utils_v2.system import files -from utils_v2.queue.kafka import ProducerKafka, create_config +from utils_v2.queue.kafka.controllers.kafka import ProducerKafka # To make HTTP calls: import httpx @@ -86,7 +86,7 @@ cwd = files.get_cwd() parent_dir = cwd kafka_producer = ProducerKafka( topic = "tickers", - config = create_config( + config = ProducerKafka.create_config( bootstrap_servers = "del.ditscentre.in:9092", # buffer_memory = 3_35_54_432, security_protocol = "SSL", @@ -130,9 +130,10 @@ def to_kafka(tick: TradingTick) -> bool: success = False summary = tick.summary - print(json.to_string(summary)) - print(json.to_string(tick.model_dump(), default=str)) + # print(json.to_string(summary, default=str)) + # print(json.to_string(tick.model_dump(), default=str)) summary["messageType"] = "ticks" + summary = json.from_string(json.to_string(summary, default=str)) success = kafka_producer.produce(value = summary) if not success: print("ERROR ON TICK NO.:", tick_count) @@ -187,8 +188,16 @@ def main(): # Get the instruments of interest: response = httpx.post(url = r"https://api.thecaoffice.com/markets/watchlist/distincts") instruments_of_interest = response.json()["data"]["rs0"] - print("TOTAL INSTR. OF INTEREST:", len(instruments_of_interest)) - symbols_of_interest = [i["symbol"] for i in instruments_of_interest] + symbols_of_interest = [] + broker_tokens_of_interest = [] + for i in instruments_of_interest: + symbol = i["symbol"] + broker_token = i["broker_token"] + if broker_token is not None and i["source"] == "zerodha": + symbols_of_interest.append(symbol) + broker_tokens_of_interest.append(broker_token) + print("TOTAL INSTR. OF INTEREST:", f"{len(broker_tokens_of_interest)}/{len(instruments_of_interest)}") + print(broker_tokens_of_interest) # Create an instance of Zerodha's Kite connection: kite = KiteConnect(api_key = api_key) @@ -196,29 +205,31 @@ def main(): # Get the entire list of instruments: instruments = [] - # instruments += kite.instruments(exchange = "NSE") - # instruments += kite.instruments(exchange = "NFO") - # instruments += kite.instruments(exchange = "BSE") - # instruments += kite.instruments(exchange = "BFO") + instruments += kite.instruments(exchange = "NSE") + instruments += kite.instruments(exchange = "NFO") + instruments += kite.instruments(exchange = "BSE") + instruments += kite.instruments(exchange = "BFO") instruments += kite.instruments(exchange = "MCX") - # instruments += kite.instruments(exchange = "CDS") - # instruments += kite.instruments(exchange = "BCD") + instruments += kite.instruments(exchange = "CDS") + instruments += kite.instruments(exchange = "BCD") - # # Pick the instruments of interest: + # Pick the instruments of interest: # instruments = [TradingSymbol.from_zerodha_kite(i) for i in instruments[:1000]] - # instruments = [ - # TradingSymbol.from_zerodha_kite(i) for i in instruments - # if i["tradingsymbol"] in symbols_of_interest or i["name"] in symbols_of_interest - # ] instruments = [ TradingSymbol.from_zerodha_kite(i) for i in instruments - if i["instrument_token"] in [109760007] + if str(i["instrument_token"]) in broker_tokens_of_interest ] + # instruments = [ + # TradingSymbol.from_zerodha_kite(i) for i in instruments + # if i["instrument_token"] in [109760007] + # ] + print("SELECTED INSTRUMENTS:", len(instruments)) # Create the lookup: for i in instruments: INSTRUMENT_TOKENS.append(i.brokerToken) INSTRUMENT_LOOKUP[i.brokerToken] = i.model_dump() + print("LOOK-UP READY!") # Start the websocket with Zerodha: kite_ws = KiteTicker( @@ -232,6 +243,7 @@ def main(): # If you choose to go threaded, you will need to work purely with callbacks. # You will need to have an infinite loop in the main thread. + print("STARTING WS...") kite_ws.connect(threaded = True) diff --git a/utils_v2/queue/kafka/__init__.py b/utils_v2/queue/kafka/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/utils_v2/queue/kafka/controllers/__init__.py b/utils_v2/queue/kafka/controllers/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/utils_v2/queue/async_kafka.py b/utils_v2/queue/kafka/controllers/async_kafka.py similarity index 97% rename from utils_v2/queue/async_kafka.py rename to utils_v2/queue/kafka/controllers/async_kafka.py index f0b874a..1a87e8f 100644 --- a/utils_v2/queue/async_kafka.py +++ b/utils_v2/queue/kafka/controllers/async_kafka.py @@ -43,6 +43,9 @@ from aiokafka import AIOKafkaConsumer from utils_v2.string import json from utils_v2.serialization.json_serializer import JSONSerializer +# Data models: +from utils_v2.queue.kafka.models.message import ConsumedKafkaMessage + # For debugging: from icecream import IceCreamDebugger @@ -333,12 +336,10 @@ class ConsumerKafka: if results: for topic_partition, records in results.items(): for record in records: - record_dict = record.__dict__ - record_dict["value"] = self.__serializer.deserialize( - data = record_dict["value"], - encoding = encoding - ) - messages.append(record_dict) + messages.append(ConsumedKafkaMessage.from_aiokafka( + message = record, + deserializer = lambda x: self.__serializer.deserialize(x, encoding = encoding) + )) # Debugging print if something went wrong: except Exception as exception: self.__printer(exception) @@ -484,7 +485,7 @@ if __name__ == "__main__": async def consumer_test(): consumer = ConsumerKafka( - topic = "kft_file_upload", + topic = "tick-listners", # group_id = "assessImg", group_id = "updateMedia", bootstrap_servers = "wtt.ditscentre.in:9092", @@ -497,13 +498,13 @@ if __name__ == "__main__": while True: messages = await consumer.consume(count = 1) - if len(messages) > 0: print("MESSAGE:", json.to_string(messages[0], default = str)) + for message in messages: print(message) await asyncio.sleep(1.0) async def producer_test(): producer = ProducerKafka( - topic = "tickers", + topic = "tick-listners", bootstrap_servers = "del.ditscentre.in:9092", security_protocol = "SSL", ssl_context = ssl_ctx @@ -519,4 +520,4 @@ if __name__ == "__main__": await producer.close() - asyncio.run(producer_test()) + asyncio.run(consumer_test()) diff --git a/utils_v2/queue/kafka.py b/utils_v2/queue/kafka/controllers/kafka.py similarity index 94% rename from utils_v2/queue/kafka.py rename to utils_v2/queue/kafka/controllers/kafka.py index dc8bc64..23c3906 100644 --- a/utils_v2/queue/kafka.py +++ b/utils_v2/queue/kafka/controllers/kafka.py @@ -44,6 +44,9 @@ from confluent_kafka import Consumer from utils_v2.string import json from utils_v2.serialization.json_serializer import JSONSerializer +# Data models: +from utils_v2.queue.kafka.models.message import ConsumedKafkaMessage + # For debugging: from icecream import IceCreamDebugger @@ -406,18 +409,21 @@ class ConsumerKafka: message = self.__consumer.poll(timeout = timeout) if message is None or message.error(): return None - else: - message_value = message.value() - if message_value is not None: message_value = self.__serializer.deserialize(message_value, encoding = encoding) - return { - "topic": message.topic(), - "partition": message.partition(), - "offset": message.offset(), - "key": message.key().decode("utf-8") if message.key() else None, - "value": message_value, - "timestamp": message.timestamp()[1], - "headers": message.headers() - } + else: return ConsumedKafkaMessage.from_confluent_kafka( + message = message, + deserializer = lambda x: self.__serializer.deserialize(x, encoding = encoding) + ) + # message_value = message.value() + # if message_value is not None: message_value = self.__serializer.deserialize(message_value, encoding = encoding) + # return { + # "topic": message.topic(), + # "partition": message.partition(), + # "offset": message.offset(), + # "key": message.key().decode("utf-8") if message.key() else None, + # "value": message_value, + # "timestamp": message.timestamp()[1], + # "headers": message.headers() + # } def consume(self, count = 1, timeout = 0.05, encoding = "utf-8"): @@ -514,20 +520,20 @@ if __name__ == "__main__": # Create and connect the producer: consumer = ConsumerKafka( - topic = "tickers", + topic = "tick-listners", config = ConsumerKafka.create_config( bootstrap_servers = "del.ditscentre.in:9092", group_id = "test-group", security_protocol = "SSL", - ca_file = r"../../creds/kafka/cert_authority.pem", - cert_file = r"../../creds/kafka/fullchain.pem", - key_file = r"../../creds/kafka/privkey.pem", + ca_file = r"../../../../creds/kafka/cert_authority.pem", + cert_file = r"../../../../creds/kafka/fullchain.pem", + key_file = r"../../../../creds/kafka/privkey.pem", ) ) consumer.connect() print("CONSUMER READY!") - for _ in range(5): + for _ in range(500): messages = consumer.consume(count = 3, timeout = 2.5) print(json.to_string(messages, default = str)) diff --git a/utils_v2/queue/kafka/models/__init__.py b/utils_v2/queue/kafka/models/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/utils_v2/queue/kafka/models/message.py b/utils_v2/queue/kafka/models/message.py new file mode 100644 index 0000000..465f0f8 --- /dev/null +++ b/utils_v2/queue/kafka/models/message.py @@ -0,0 +1,257 @@ +""" + + AUTHOR: + + Khushal P Soonderji + + DATE: + + Monday, 30th Dec., 2024. + + OBJECTIVE: + + To provide a standardized structure for Kafka messages. + + REFERENCES: + + N/A + + DOWNLOADS: + + N/A + +""" + + +# ***************************************************************************************************************** +# ***** **** +# *** IMPORT *** +# ***** **** +# ***************************************************************************************************************** + + +# To make sibling directories accessible for imports: +import sys +sys.path.append(".") +sys.path.append("..") + +# For making data behaviour_models: +from pydantic import BaseModel, Field, field_validator, model_validator, AwareDatetime +from typing import Optional, Literal, Union, Dict, List, Any + +# Related to Google: +from google.auth.transport.requests import Request +from google.oauth2.credentials import Credentials + +# My utils: +from utils_v2.string import json +from utils_v2.string import regex +from utils_v2.date_time import date_time + +# To work with date and time: +import datetime +import dateparser + +# To make API calls: +import httpx + + +# ***************************************************************************************************************** +# ***** **** +# *** MACROS / ONE-TIME INIT *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** VARIABLES *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** FUNCTIONS *** +# ***** **** +# ***************************************************************************************************************** + + +class ConsumedKafkaMessage(BaseModel): + + topic: str = Field( + description = "the topic on which this message was received", + frozen = True + ) + + partition: int = Field( + description = "the partition in which this message was received", + frozen = True + ) + + offset: int = Field( + description = "the message's no. in the partition", + frozen = True + ) + + headers: List[Any] = Field( + description = "the headers received with the message", + frozen = True + ) + + key: Any = Field( + description = "the key with which this message is associated; important for partition management", + frozen = True + ) + + value: Any = Field( + description = "the actual payload of the message", + frozen = True + ) + + ts: AwareDatetime | None = Field( + description = "the time at which this message was sent to the queue", + frozen = True + ) + + tsType: Literal[ + "createTime", # ...... The time at which the producer produced the message. + "logAppendTime", # ... The time at which the message was received by the broker. + None # ............... Unknown. + ] = Field( + description = "to understand the source of the timestamp", + frozen = True + ) + + # ┏┓ ┏• + # ┃ ┏┓┏┓╋┓┏┓ + # ┗┛┗┛┛┗┛┗┗┫ + # ┛ + + class Config: + extra = "ignore" + populate_by_name = True + + # ┓┏ ┓• ┓ • + # ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓ + # ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗ + + @field_validator("ts", mode = "before") + def parse_dates(cls, value): + if value is None: return None + if not isinstance(value, datetime.datetime): + parsed = date_time.parse_date_time( + value, + date_formats = ["%Y-%m-%d %H:%M:%S"], + timezone = None + ) + value = parsed if isinstance(parsed, datetime.datetime) else dateparser.parse(value) + if isinstance(value, datetime.datetime): value = date_time.to_timezone(value, date_time.TIMEZONE_UTC) + return value + + # ┏┓ • + # ┃┃┏┓┏┓┏┓┏┓┏┓╋┓┏┓┏ + # ┣┛┛ ┗┛┣┛┗ ┛ ┗┗┗ ┛ + # ┛ + + pass + + # ┏┓ ┏┓ + # ┃ ┓┏┏╋┏┓┏┳┓ ┣ ┓┏┏┓┏┏ + # ┗┛┗┻┛┗┗┛┛┗┗ ┻ ┗┻┛┗┗┛ + + @staticmethod + def from_aiokafka( + message, + deserializer = None + ): + + """ + To populate this model directly from the output of the 'aiokafka' library. + :param message: The raw message from the library. + :param deserializer: The function to use to deserialize the contents of the message. + :return: The standardized Kafka consumed message. + """ + + # Extract the key and value: + key = message.key + value = message.value + if deserializer: + if key: key = deserializer(key) + if value: value = deserializer(value) + + # Build and return the model: + return ConsumedKafkaMessage( + topic = message.topic, + partition = message.partition, + offset = message.offset, + headers = message.headers or [], + key = key, + value = value, + ts = message.timestamp / 1000.0 if message.timestamp else None, + tsType = { + 0: "createTime", + 1: "logAppendTime" + }.get(message.timestamp_type) + ) + + @staticmethod + def from_confluent_kafka( + message, + deserializer = None + ): + + """ + To populate this model directly from the output of the 'aiokafka' library. + :param message: The raw message from the library. + :param deserializer: The function to use to deserialize the contents of the message. + :return: The standardized Kafka consumed message. + """ + + # If the message was null or an error: + if message is None or message.error(): return None + + # Figure out the timestamp: + raw_ts = message.timestamp() + ts_type = raw_ts[0] if raw_ts else None + ts = raw_ts[1] / 1_000.0 if raw_ts else None + + # Extract the key and value: + key = message.key() + value = message.value() + if deserializer: + if key: key = deserializer(key) + if value: value = deserializer(value) + + # Build and return the model: + return ConsumedKafkaMessage( + topic = message.topic(), + partition = message.partition(), + offset = message.offset(), + headers = message.headers() or [], + key = key, + value = value, + ts = ts, + tsType = { + 1: "createTime", + 2: "logAppendTime" + }.get(ts_type) + ) + + +# ***************************************************************************************************************** +# ***** **** +# *** MAIN PROGRAM *** +# ***** **** +# ***************************************************************************************************************** + + +if __name__ == "__main__": + + pass diff --git a/utils_v2/trading/zerodha_kite/controllers/async_zerodha_kite.py b/utils_v2/trading/zerodha_kite/controllers/async_zerodha_kite.py index e794ac7..61704b2 100644 --- a/utils_v2/trading/zerodha_kite/controllers/async_zerodha_kite.py +++ b/utils_v2/trading/zerodha_kite/controllers/async_zerodha_kite.py @@ -349,7 +349,6 @@ class AsyncZerodhaKite: client_json = await client_response.get_json() self.__access_token = client_json["data"]["access_token"] - self.__ print(client_response.to_markdown()) print("CLIENT RESPONSE;", json.to_string(await client_response.get_json())) @@ -381,6 +380,6 @@ if __name__ == "__main__": # Login flow: print("LOGIN URL:", my_kite.login_url) my_kite.set_request_token(input("Request Token: ")) - await my_kite.get_access_token() + await my_kite.generate_session() asyncio.run(main()) diff --git a/wsio/finstitutions/trading/main.py b/wsio/finstitutions/trading/main.py index 8126f13..e87263d 100644 --- a/wsio/finstitutions/trading/main.py +++ b/wsio/finstitutions/trading/main.py @@ -43,7 +43,8 @@ import os from utils_v2.string import json from utils_v2.system import files from utils_v2.database.async_mongo_v2 import AsyncMongo -from utils_v2.queue.async_kafka import ConsumerKafka, get_ssl_context +from utils_v2.queue.kafka.controllers.async_kafka import ConsumerKafka, get_ssl_context +from utils_v2.cache.async_redis_cache_v2 import AsyncRedisCache # To make HTTP calls: import httpx @@ -98,6 +99,9 @@ EVENT_DISCONNECT = "disconnect" EVENT_ECHO = "echo" EVENT_TICKS = "ticks" +# Redis: +redis_cache = None + # ***************************************************************************************************************** # ***** **** @@ -137,12 +141,19 @@ async def handle_connect(sid, environ) -> bool: async with lock: connected_clients[sid] = { "user": None, + "redisKey": f"io_{environ.get('HTTP_X_SESSION_TOKEN')}", "rooms": [] } # Allow/reject requests: printer(sid) print("ENVIRON:", json.to_string(environ, default = str)) + while redis_cache is None: await asyncio.sleep(0.5) + result = await redis_cache.set( + key = connected_clients[sid]["redisKey"], + value = {"server": SERVER_HOSTNAME, "socket_id": sid} + ) + print("CACHED:", result) return True @@ -152,6 +163,8 @@ async def handle_connect(sid, environ) -> bool: @sio.on(event = EVENT_DISCONNECT, namespace = NAMESPACE_MODULE) async def handle_disconnect(sid, reason) -> None: printer(sid, reason) + result = await redis_cache.delete(key = connected_clients[sid]["redisKey"]) + print("UN-CACHED:", result) # --------------------------------------------------------------------------------------------------------------------- @@ -313,7 +326,7 @@ async def ticks_from_kafka( # Each message must be treated as an array of tick updates (list of dicts). # In case the producer is sending each individual tick as a separate message, # we normalize it to be a list: - tasks = [send_ticks(t["value"] if isinstance(t["value"], list) else [t["value"]]) for t in ticks] + tasks = [send_ticks(t.value if isinstance(t.value, list) else [t.value]) for t in ticks] results = await asyncio.gather(*tasks) printer(len(ticks)) @@ -327,6 +340,11 @@ async def init(): if os.environ["DEBUG"].lower() == "true": printer.enable() printer("Initializing.") + global redis_cache + redis_cache = AsyncRedisCache( + connection_string = r"redis://:dc4da94197c843ab6a730113c2b801d9@192.168.2.251/0" + ) + # Start consuming ticks in the background: cwd = files.get_cwd() parent_dir = cwd