61 lines
1.5 KiB
Python
61 lines
1.5 KiB
Python
import time
|
|
|
|
import redis
|
|
|
|
from settings import settings
|
|
from v2 import *
|
|
from v2.event_attr import EventAttr
|
|
from v2.ipregion import IpSearch, Ip2Region
|
|
from multiprocessing import Lock
|
|
|
|
ipsearch = IpSearch(Ip2Region, "ip2region.db")
|
|
lock = Lock()
|
|
|
|
from v2.log import logger
|
|
|
|
rdb = redis.Redis(**settings.REDIS_CONF)
|
|
event_attr = EventAttr(rdb)
|
|
|
|
|
|
def run():
|
|
db_client = CK(**settings.CK_CONFIG)
|
|
sketch = Sketch(db_client)
|
|
handler_event = HandlerEvent(db_client, settings.GAME, ipsearch)
|
|
handler_user = HandlerUser(db_client, settings.GAME)
|
|
transmitter = Transmitter(db_client, settings.GAME, sketch, logger, lock, event_attr)
|
|
transmitter.add_source(handler_event, 10000, 10)
|
|
transmitter.add_source(handler_user, 1000, 10)
|
|
last_ts = int(time.time())
|
|
consumer, kafka_client = create_consumer(-1)
|
|
|
|
for topic, msg in consumer():
|
|
# print(msg)
|
|
type_ = msg['#type']
|
|
if msg['#event_name'] != 'pay':
|
|
continue
|
|
print(msg)
|
|
|
|
del msg['#type']
|
|
ts = int(time.time())
|
|
|
|
if 'user' in type_:
|
|
# continue
|
|
obj = getattr(handler_user, type_)
|
|
handler_user.receive_data.append(UserAct(obj, msg))
|
|
if len(handler_user.receive_data) >= 1000 or last_ts + 60 < ts:
|
|
last_ts = ts
|
|
handler_user.execute()
|
|
|
|
elif 'track' in type_:
|
|
# continue
|
|
obj = getattr(handler_event, type_)
|
|
obj(msg)
|
|
else:
|
|
continue
|
|
|
|
transmitter.run()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
run()
|