main.py
2.31 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
import json
import time
from aabd.base.patched_logging import init_logging, get_logger
init_logging()
logger = get_logger(__name__)
from threading import Thread
from .core.api import replay_match_event
from aabd.mq.kafka_client import KafkaMessageIterator, KafkaProducer
def get_message_iterator(kafka_config):
return KafkaMessageIterator(bootstrap_servers=kafka_config.get('servers', None),
group_id=kafka_config.get('group_id', None),
topic=kafka_config.get('topic', None),
sasl_plain_username=kafka_config.get('username', None),
sasl_plain_password=kafka_config.get('password', None),
value_deserializer="str")
def get_message_producer(kafka_config):
return KafkaProducer(bootstrap_servers=kafka_config.get('servers', None),
sasl_plain_username=kafka_config.get('username', None),
sasl_plain_password=kafka_config.get('password', None))
def get_callback_func(producer, topic):
def callback_func(task_id, call_data):
try:
st = time.time()
producer.send_message_async(topic, call_data, key=task_id, timeout=5)
logger.info(f"sent_message[{task_id}][{time.time() - st:.2f}s]: {call_data}")
except:
logger.exception(f"Error sending message: {call_data}")
return callback_func
def kafka_message_iterator_thread(message_iterator, config, callback_func):
with message_iterator:
for message in message_iterator:
try:
json_message = json.loads(message)
replay_match_event(json_message)
logger.info(message)
except:
logger.exception(f"Error processing message: {message}")
if __name__ == '__main__':
from config import settings
message_iter = get_message_iterator(settings.input_kafka)
kafka_producer = get_message_producer(settings.output_kafka)
try:
Thread(target=kafka_message_iterator_thread,
args=(message_iter, settings,
get_callback_func(kafka_producer, settings.output_kafka.get('topic', None)))).start()
except KeyboardInterrupt:
message_iter.running = False
kafka_producer.close()