idds/tests/split_messages.py

0001 #!/usr/bin/env python
0002
0003 from idds.common.constants import MessageStatus                        # noqa F401
0004 from idds.core.messages import retrieve_messages, add_messages         # noqa F401
0005
0006
0007 def clean_msg(msg):
0008     new_msg = {'msg_type': msg['msg_type'],
0009                'status': MessageStatus.New,
0010                'source': msg['source'],
0011                'destination': msg['destination'],
0012                'request_id': msg['request_id'],
0013                'workload_id': msg['workload_id'],
0014                'transform_id': msg['transform_id'],
0015                'num_contents': msg['num_contents'],
0016                'msg_content': msg['msg_content']}
0017     return new_msg
0018
0019
0020 for req_id in range(1073, 1084):
0021     msgs = retrieve_messages(request_id=req_id)
0022     number_contents = 0
0023     for msg in msgs:
0024         # print(json_dumps(msg['msg_content'], sort_keys=True, indent=4))
0025         # if msg['num_contents'] > 10000 and msg['status'] == MessageStatus.New:
0026         if msg['num_contents'] >= 10000:
0027             print(msg['request_id'], msg['msg_id'], msg['num_contents'])
0028             msg = clean_msg(msg)
0029             add_messages([msg], bulk_size=10000)
0030         # break