forked from candidelabs/voltaire
-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
45 lines (36 loc) · 1.44 KB
/
main.py
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
import asyncio
import uvloop
from functools import partial
from signal import SIGINT, SIGTERM
from init import initialize
from rpc.rpc_http_server import run_rpc_http_server
from init import InitData
from bundler.execution_endpoint import ExecutionEndpoint
from utils.SignalHaltError import immediate_exit
async def main() -> None:
initData: InitData = initialize()
loop = asyncio.get_running_loop()
for signal_enum in [SIGINT, SIGTERM]:
exit_func = partial(immediate_exit, signal_enum=signal_enum, loop=loop)
loop.add_signal_handler(signal_enum, exit_func)
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
# loop.set_debug(True)
async with asyncio.TaskGroup() as task_group:
execution_endpoint: ExecutionEndpoint = ExecutionEndpoint(
initData.ethereum_node_url,
initData.bundler_pk,
initData.bundler_address,
initData.entrypoint,
initData.bundler_helper_byte_code,
initData.chain_id,
initData.is_unsafe,
initData.is_legacy_mode,
initData.is_send_raw_transaction_conditional,
initData.bundle_interval
)
task_group.create_task(execution_endpoint.start_execution_endpoint())
task_group.create_task(
run_rpc_http_server(host=initData.rpc_url, port=initData.rpc_port, is_debug=initData.is_debug)
)
if __name__ == "__main__":
asyncio.run(main())