-
Notifications
You must be signed in to change notification settings - Fork 1
/
electrumz_rpc
162 lines (143 loc) · 5.17 KB
/
electrumz_rpc
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
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
'''Script to send RPC commands to a running ElectrumZ server.'''
import argparse
import asyncio
import json
import sys
from os import environ
from aiorpcx import timeout_after, connect_rs
import electrumz.lib.text as text
simple_commands = {
'getinfo': 'Print a summary of server state',
'groups': 'Print current session groups',
'peers': 'Print information about peer servers for the same coin',
'sessions': 'Print information about client sessions',
'stop': 'Shut down the server cleanly',
}
session_commands = {
'disconnect': 'Disconnect sessions',
'log': 'Control logging of sessions',
}
other_commands = {
'add_peer': (
'add a peer to the peers list',
[], {
'type': str,
'dest': 'real_name',
'help': 'e.g. "a.domain.name s995 t"',
},
),
'daemon_url': (
"replace the daemon's URL at run-time, and forecefully rotate "
" to the first URL in the list",
[], {
'type': str,
'nargs': '?',
'default': '',
'dest': 'daemon_url',
'help': 'see documentation of DAEMON_URL envvar',
},
),
'query': (
'query the UTXO and history databases',
['-l', '--limit'], {
'type': int,
'default': 1000,
'help': 'UTXO and history output limit',
}, ['items'], {
'nargs': '+',
'type': str,
'help': 'hex scripts, or addresses, to query',
},
),
'reorg': (
'simulate a chain reorganization',
[], {
'type': int,
'dest': 'count',
'default': 3,
'help': 'number of blocks to back up'
},
),
'debug_memusage_list_all_objects': (
'Print a table of types of most common types in memory',
['--limit'], {
'type': int,
'default': 50,
'help': 'max number of types to return',
},
),
'debug_memusage_get_random_backref_chain': (
'Return a dotfile as text containing the backref chain for a randomly selected object of type objtype',
[], {
'type': str,
'dest': 'objtype',
'help': 'e.g. "_asyncio.Task"',
},
),
}
def main():
'''Send the RPC command to the server and print the result.'''
main_parser = argparse.ArgumentParser(
'electrumz_rpc',
description='Send electrumz an RPC command'
)
main_parser.add_argument('-p', '--port', metavar='port_num', type=int,
help='RPC port number')
main_parser.add_argument('--timeout', type=int, default=30,
help='timeout for command in seconds')
subparsers = main_parser.add_subparsers(help='sub-command help',
dest='command')
for command, help in simple_commands.items():
parser = subparsers.add_parser(command, help=help)
for command, help in session_commands.items():
parser = subparsers.add_parser(command, help=help)
parser.add_argument('session_ids', nargs='+', type=str,
help='list of session ids')
for command, data in other_commands.items():
parser_help, *arguments = data
parser = subparsers.add_parser(command, help=parser_help)
for n in range(0, len(arguments), 2):
args, kwargs = arguments[n: n+2]
parser.add_argument(*args, **kwargs)
args = main_parser.parse_args()
args = vars(args)
port = args.pop('port')
if port is None:
port = int(environ.get('RPC_PORT', 8000))
method = args.pop('command')
timeout = args.pop('timeout')
# aiorpcX makes this so easy...
async def send_request():
try:
async with timeout_after(timeout):
async with connect_rs('localhost', port) as session:
session.transport._framer.max_size = 0
session.sent_request_timeout = timeout
result = await session.send_request(method, args)
if method in ('query', ):
for line in result:
print(line)
elif method in (
'debug_memusage_list_all_objects',
'debug_memusage_get_random_backref_chain',
):
for line in result.split('\n'):
print(line)
elif method in ('groups', 'peers', 'sessions'):
lines_func = getattr(text, f'{method}_lines')
for line in lines_func(result):
print(line)
else:
print(json.dumps(result, indent=4, sort_keys=True))
return 0
except OSError:
print('cannot connect - is ElectrumZ catching up, not running, or '
f'is {port} the wrong RPC port?')
return 1
except Exception as e:
print(f'error making request: {e}')
return 1
code = asyncio.run(send_request())
sys.exit(code)
if __name__ == '__main__':
main()