-
Notifications
You must be signed in to change notification settings - Fork 4
/
Basic_aioquic_client.py
54 lines (42 loc) · 1.54 KB
/
Basic_aioquic_client.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
46
47
48
49
50
51
52
53
54
############ BASIC CLIENT ##################
import argparse
import asyncio
import ssl
import struct
import aioquic.asyncio
from aioquic.quic.configuration import QuicConfiguration
from aioquic.quic.events import StreamDataReceived
_I = struct.Struct('!I')
class QuicClient(aioquic.asyncio.QuicConnectionProtocol):
def quic_event_received(self, quic_event):
if isinstance(quic_event, StreamDataReceived):
if quic_event.end_stream:
self.time.set_result(self._loop.time() - self.start_time)
def transfer(self, size):
self._quic.send_stream_data(
stream_id=self._quic.get_next_available_stream_id(),
data=_I.pack(size),
end_stream=True,
)
self.transmit()
self.start_time = self._loop.time()
self.time = self._loop.create_future()
return self.time
async def main(host='172.16.2.1', port=9999, size=1048576):
async with aioquic.asyncio.connect(
host, port,
configuration=QuicConfiguration(verify_mode=ssl.CERT_NONE),
create_protocol=QuicClient,
) as client:
print('Transfer speed: {:.3f} Mbps'.format(
size * 8 / 1000 / 1000 / await client.transfer(size))
)
if __name__ == '__main__':
parser = argparse.ArgumentParser()
parser.add_argument('size', nargs='?', type=int, default=64, help='Transfer size (MiB)')
args = parser.parse_args()
asyncio.run(
main(
size=args.size * 1048576,
)
)