6862430736

Committed 14 Nov 2023 10:37AM UTC coverage: 42.467%. First build

Build # 6862430736

Build Type

push

github

Committed by

feihoo87

Commit Message

add auto save state for dht

Run Details

0 of 4 new or added lines in 1 file covered. (0.0%)

7289 of 17164 relevant lines covered (42.47%)

2.54 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/waveforms/sys/net/cli.py

import asyncio
import pathlib
import pickle

import click

from .kad import Server


async def client_get(key, server, port, timeout=5):
    reader, writer = await asyncio.wait_for(asyncio.open_connection(
        server, port),
                                            timeout=timeout)
    writer.write(pickle.dumps(('get', key)))
    await asyncio.wait_for(writer.drain(), timeout=timeout)
    data = await asyncio.wait_for(reader.read(1600), timeout=timeout)
    value = pickle.loads(data)
    writer.close()
    await writer.wait_closed()
    return value


async def client_set(key, value, server, port, timeout=5):
    reader, writer = await asyncio.wait_for(asyncio.open_connection(
        server, port),
                                            timeout=timeout)
    writer.write(pickle.dumps(('set', key, value)))
    await asyncio.wait_for(writer.drain(), timeout=timeout)
    writer.close()
    await writer.wait_closed()


async def client_bootstrap(address, server, port, timeout=5):
    reader, writer = await asyncio.wait_for(asyncio.open_connection(
        server, port),
                                            timeout=timeout)
    addr, p = address.split(':')
    writer.write(pickle.dumps(('bootstrap', (addr, int(p)))))
    await asyncio.wait_for(writer.drain(), timeout=timeout)
    writer.close()
    await writer.wait_closed()


@click.group()
def dht():
    pass


@dht.command()
@click.option('--interface',
              default='127.0.0.1',
              help='Interface to listen on')
@click.option('--port', default=8467, help='Port to listen on')
@click.option('--dht-only', default=False, is_flag=True, help='DHT only')
@click.option('--dht-interface',
              default='0.0.0.0',
              help='DHT interface to listen on')
@click.option('--dht-port', default=8468, help='DHT port to listen on')
@click.option('--bootstrap', default=None, help='Address of bootstrap node')
@click.option('--interval', default=300, help='Refresh interval')
@click.option('--state-file', default=None, help='State file')
def server(interface, port, dht_only, dht_interface, dht_port, bootstrap,
           interval, state_file):

    async def main(interface, port, dht_only, dht_interface, dht_port,
                   bootstrap, state_file):

        if state_file is None:
            state_file = pathlib.Path.home() / '.waveforms' / 'dht.pickle'
        else:
            state_file = pathlib.Path(state_file)

        if state_file.exists():
            node = await Server.load_state(state_file, dht_port, dht_interface,
                                           interval)
        else:
            node = Server()
            state_file.parent.mkdir(parents=True, exist_ok=True)
            await node.listen(dht_port, dht_interface, interval)
        if bootstrap:
            addr, p = bootstrap.split(':')
            await node.bootstrap([(addr, int(p))])

        loop = asyncio.get_running_loop()
        loop.call_later(min(interval / 2, 30), node.save_state_regularly,
                        state_file, interval)

        async def handle_cmds(reader, writer):
            data = await reader.read(1600)
            cmd, *args = pickle.loads(data)
            match cmd:
                case 'get':
                    value = await node.get(*args)
                    data = pickle.dumps(value)
                case 'set':
                    value = await node.set(*args)
                    data = pickle.dumps(value)
                case 'ping':
                    data = pickle.dumps(node.node.id)
                case 'bootstrap':
                    await node.bootstrap(args)
                    data = pickle.dumps(None)
                case _:
                    data = pickle.dumps(None)

            writer.write(data)
            await writer.drain()
            writer.close()
            await writer.wait_closed()

        if dht_only:
            try:
                while True:
                    await asyncio.sleep(1)
            except asyncio.CancelledError:
                pass
        else:
            server = await asyncio.start_server(handle_cmds, interface, port)

            async with server:
                try:
                    await server.serve_forever()
                except asyncio.CancelledError:
                    pass

    asyncio.run(
        main(interface, port, dht_only, dht_interface, dht_port, bootstrap,
             state_file))


@dht.command('get')
@click.argument('key')
@click.option('--server', default='127.0.0.1', help='Server to connect to')
@click.option('--port', default=8467, help='Port to connect to')
@click.option('--timeout', default=5, help='Timeout')
def get(key, server, port, timeout):

    async def client():
        value = await client_get(key, server, port, timeout)
        print(value)

    asyncio.run(client())


@dht.command('set')
@click.argument('key')
@click.argument('value')
@click.option('--server', default='127.0.0.1', help='Server to connect to')
@click.option('--port', default=8467, help='Port to connect to')
@click.option('--timeout', default=5, help='Timeout')
def set_(key, value, server, port, timeout):

    async def client():
        await client_set(key, value, server, port, timeout)

    asyncio.run(client())


@dht.command('bootstrap')
@click.argument('address')
@click.option('--server', default='127.0.0.1', help='Server to connect to')
@click.option('--port', default=8467, help='Port to connect to')
@click.option('--timeout', default=5, help='Timeout')
def bootstrap(address, server, port, timeout):

    async def client():
        await client_bootstrap(address, server, port, timeout)

    asyncio.run(client())

1	import asyncio	×
2	import pathlib	×
3	import pickle	×
4
5	import click	×
6
7	from .kad import Server	×
8
9
10	async def client_get(key, server, port, timeout=5):	×
11	reader, writer = await asyncio.wait_for(asyncio.open_connection(	×
12	server, port),
13	timeout=timeout)
14	writer.write(pickle.dumps(('get', key)))	×
15	await asyncio.wait_for(writer.drain(), timeout=timeout)	×
16	data = await asyncio.wait_for(reader.read(1600), timeout=timeout)	×
17	value = pickle.loads(data)	×
18	writer.close()	×
19	await writer.wait_closed()	×
20	return value	×
21
22
23	async def client_set(key, value, server, port, timeout=5):	×
24	reader, writer = await asyncio.wait_for(asyncio.open_connection(	×
25	server, port),
26	timeout=timeout)
27	writer.write(pickle.dumps(('set', key, value)))	×
28	await asyncio.wait_for(writer.drain(), timeout=timeout)	×
29	writer.close()	×
30	await writer.wait_closed()	×
31
32
33	async def client_bootstrap(address, server, port, timeout=5):	×
34	reader, writer = await asyncio.wait_for(asyncio.open_connection(	×
35	server, port),
36	timeout=timeout)
37	addr, p = address.split(':')	×
38	writer.write(pickle.dumps(('bootstrap', (addr, int(p)))))	×
39	await asyncio.wait_for(writer.drain(), timeout=timeout)	×
40	writer.close()	×
41	await writer.wait_closed()	×
42
43
44	@click.group()	×
45	def dht():	×
46	pass	×
47
48
49	@dht.command()	×
50	@click.option('--interface',	×
51	default='127.0.0.1',
52	help='Interface to listen on')
53	@click.option('--port', default=8467, help='Port to listen on')	×
54	@click.option('--dht-only', default=False, is_flag=True, help='DHT only')	×
55	@click.option('--dht-interface',	×
56	default='0.0.0.0',
57	help='DHT interface to listen on')
58	@click.option('--dht-port', default=8468, help='DHT port to listen on')	×
59	@click.option('--bootstrap', default=None, help='Address of bootstrap node')	×
60	@click.option('--interval', default=300, help='Refresh interval')	×
61	@click.option('--state-file', default=None, help='State file')	×
62	def server(interface, port, dht_only, dht_interface, dht_port, bootstrap,	×
63	interval, state_file):
64
65	async def main(interface, port, dht_only, dht_interface, dht_port,	×
66	bootstrap, state_file):
67
68	if state_file is None:	×
69	state_file = pathlib.Path.home() / '.waveforms' / 'dht.pickle'	×
70	else:
71	state_file = pathlib.Path(state_file)	×
72
73	if state_file.exists():	×
NEW 74	node = await Server.load_state(state_file, dht_port, dht_interface,	×
75	interval)
76	else:
77	node = Server()	×
78	state_file.parent.mkdir(parents=True, exist_ok=True)	×
79	await node.listen(dht_port, dht_interface, interval)	×
NEW 80	if bootstrap:	×
NEW 81	addr, p = bootstrap.split(':')	×
NEW 82	await node.bootstrap([(addr, int(p))])	×
83
84	loop = asyncio.get_running_loop()	×
85	loop.call_later(min(interval / 2, 30), node.save_state_regularly,	×
86	state_file, interval)
87
88	async def handle_cmds(reader, writer):	×
89	data = await reader.read(1600)	×
90	cmd, *args = pickle.loads(data)	×
91	match cmd:	×
92	case 'get':	×
93	value = await node.get(*args)	×
94	data = pickle.dumps(value)	×
95	case 'set':	×
96	value = await node.set(*args)	×
97	data = pickle.dumps(value)	×
98	case 'ping':	×
99	data = pickle.dumps(node.node.id)	×
100	case 'bootstrap':	×
101	await node.bootstrap(args)	×
102	data = pickle.dumps(None)	×
103	case _:	×
104	data = pickle.dumps(None)	×
105
106	writer.write(data)	×
107	await writer.drain()	×
108	writer.close()	×
109	await writer.wait_closed()	×
110
111	if dht_only:	×
112	try:	×
113	while True:	×
114	await asyncio.sleep(1)	×
115	except asyncio.CancelledError:	×
116	pass	×
117	else:
118	server = await asyncio.start_server(handle_cmds, interface, port)	×
119
120	async with server:	×
121	try:	×
122	await server.serve_forever()	×
123	except asyncio.CancelledError:	×
124	pass	×
125
126	asyncio.run(	×
127	main(interface, port, dht_only, dht_interface, dht_port, bootstrap,
128	state_file))
129
130
131	@dht.command('get')	×
132	@click.argument('key')	×
133	@click.option('--server', default='127.0.0.1', help='Server to connect to')	×
134	@click.option('--port', default=8467, help='Port to connect to')	×
135	@click.option('--timeout', default=5, help='Timeout')	×
136	def get(key, server, port, timeout):	×
137
138	async def client():	×
139	value = await client_get(key, server, port, timeout)	×
140	print(value)	×
141
142	asyncio.run(client())	×
143
144
145	@dht.command('set')	×
146	@click.argument('key')	×
147	@click.argument('value')	×
148	@click.option('--server', default='127.0.0.1', help='Server to connect to')	×
149	@click.option('--port', default=8467, help='Port to connect to')	×
150	@click.option('--timeout', default=5, help='Timeout')	×
151	def set_(key, value, server, port, timeout):	×
152
153	async def client():	×
154	await client_set(key, value, server, port, timeout)	×
155
156	asyncio.run(client())	×
157
158
159	@dht.command('bootstrap')	×
160	@click.argument('address')	×
161	@click.option('--server', default='127.0.0.1', help='Server to connect to')	×
162	@click.option('--port', default=8467, help='Port to connect to')	×
163	@click.option('--timeout', default=5, help='Timeout')	×
164	def bootstrap(address, server, port, timeout):	×
165
166	async def client():	×
167	await client_bootstrap(address, server, port, timeout)	×
168
169	asyncio.run(client())	×

feihoo87 / waveforms / 6862430736

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous