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
|
"""Start a frozen meshbay-node as a connection process and carry bytes through it.
With `connection_processes` above 1 the node starts copies of itself as
`meshbay-node.exe webrtc-worker` (transport/webrtc/workers.py). The daemon's own
smoke test never gets there: the processes start once the node has signed in to
a hub, and that test has none. This is the node's side of it, unchanged, with a
frozen executable as the process and an aiortc peer here as the browser.
Run by build-node-runtime.ps1 with the build venv's interpreter, which has the
node's own modules. It takes any meshbay-node.exe, an installed one included
(`resources/node-runtime` under the application's directory).
"""
import asyncio
import sys
import time
from aiortc import RTCPeerConnection, RTCSessionDescription
from meshbay_node.transport.webrtc import workers
CHUNK = bytes(48_000)
CHUNKS = 200
async def until(test, what: str, seconds: float = 30.0) -> None:
deadline = time.monotonic() + seconds
while not test():
if time.monotonic() > deadline:
raise SystemExit(f"FAILED {what}: not within {seconds:.0f}s")
await asyncio.sleep(0.02)
async def carry(exe: str) -> str:
workers._argv = lambda: [exe, "webrtc-worker"]
pool = workers.WorkerPool(2)
browser = RTCPeerConnection()
try:
await pool.ensure()
if pool.processes() != 2:
raise SystemExit(f"FAILED {pool.processes()} of 2 connection processes started")
channel = browser.createDataChannel("mnp")
got: list[bytes] = []
channel.on("message", got.append)
await browser.setLocalDescription(await browser.createOffer())
peer = await pool.open(browser.localDescription.sdp)
ours: list = []
peer.on("datachannel", ours.append)
await browser.setRemoteDescription(
RTCSessionDescription(sdp=await peer.negotiate(), type="answer"))
await until(lambda: channel.readyState == "open" and ours, "the data channel opens")
heard: list[bytes] = []
ours[0].on("message", heard.append)
channel.send(b"from the browser")
await until(lambda: heard == [b"from the browser"], "a message reaches the node")
for _ in range(CHUNKS):
ours[0].send(CHUNK)
while ours[0].bufferedAmount > 2_000_000:
await asyncio.sleep(0.005)
await until(lambda: sum(map(len, got)) == CHUNKS * len(CHUNK),
"what the node sent arrives whole")
await until(lambda: ours[0].bufferedAmount == 0,
"the node's count of what is buffered comes back to zero")
finally:
started = [w.process for w in pool._workers]
await browser.close()
await pool.close()
# A process still there would be one an upgrade finds holding
# meshbay-node.exe. Its exit code says nothing: the node ends a process
# whose socket has closed, and on Windows that reads 1 however it left.
if any(p.returncode is None for p in started):
raise SystemExit("FAILED a connection process outlived the node's stop")
return f"{CHUNKS * len(CHUNK) // 1_000_000} MB through a connection process"
def main() -> None:
if len(sys.argv) < 2:
raise SystemExit(__doc__)
for exe in sys.argv[1:]:
print(f"OK {exe}: {asyncio.run(carry(exe))}")
if __name__ == "__main__":
main()
|