main.py
2.77 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
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
from aabd.base.patched_logging import init_logging, get_logger
init_logging()
logger = get_logger(__name__)
import json
import time
try:
from config import settings
except:
from .config import settings
pass
try:
from .core.api import FootballReplayMatch
except:
from core.api import FootballReplayMatch
pass
from threading import Thread
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):
frm = FootballReplayMatch(config)
with message_iterator:
for message in message_iterator:
try:
json_message = json.loads(message)
result = frm.replay_match_event(json_message)
task_id = json_message.get("id")
callback_func(task_id, json.dumps(result, ensure_ascii=False))
logger.info(f"Processed task_id={task_id}, result={result}")
except Exception:
logger.exception(f"Error processing message: {message}")
def main():
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()
if __name__ == '__main__':
main()