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
163
164
165
166
167
168
169
170
171
|
"""
The desktop cast relay's stream, run for real: plain Node, no dependencies.
Three faults, each of which a receiver showed as a cast that hung or froze:
* **The header is everything before the first moof.** The page hands the relay
its first decrypted chunk, which is a slice of the stream: the ftyp alone,
the header exactly, or the header and the start of a fragment, depending on
when the node read ffmpeg's output. Taken as the header, two of the three
broke the stream, so the same film cast one try in three.
* **A receiver reading at playback speed loses nothing.** The page runs ahead
of the television by its whole read-ahead; dropped past 8 MB, each lost
fragment froze the picture for its few seconds.
* **A stop the page does not wait for cannot clear the next start.** The old
stop cleared its state after the port closed, which was after the new
start, and the receiver was handed `:null/stream.mp4?t=null`.
"""
import json
import shutil
import subprocess
from pathlib import Path
import pytest
PACKAGES = Path(__file__).resolve().parents[2]
RELAY = PACKAGES / "meshbay-client" / "src" / "cast-relay.js"
SCRIPT = r"""
const http = require('node:http');
const CastRelay = require(process.argv[2]);
const box = (type, size) => {
const b = Buffer.alloc(size, type.charCodeAt(0));
b.writeUInt32BE(size, 0); b.write(type, 4, 'latin1');
return b;
};
const ftyp = box('ftyp', 28);
const moov = box('moov', 1272);
const header = Buffer.concat([ftyp, moov]);
const fragment = (n, size) => Buffer.concat([box('moof', 100 + n), box('mdat', size)]);
// Read a whole response, or only what it holds after `ms`.
const read = (url, ms) => new Promise((resolve, reject) => {
const parts = [];
const req = http.get(url, { agent: false }, (res) => {
res.on('data', (c) => parts.push(c));
res.on('end', () => resolve(Buffer.concat(parts)));
if (ms) setTimeout(() => { req.destroy(); resolve(Buffer.concat(parts)); }, ms);
});
req.on('error', (e) => (ms ? null : reject(e)));
});
const sleep = (ms) => new Promise((ok) => setTimeout(ok, ms));
// The page's order: start with its first chunk, then push every chunk,
// that one included.
async function cast(relay, chunks, { early } = {}) {
const started = await relay.start({ initSegment: chunks[0] });
const body = early ? read(started.url) : null;
if (early) await sleep(50);
for (const c of chunks) relay.pushSegment(c);
const served = early ? null : read(started.url);
await sleep(50);
relay.finish();
return (await (body || served)).toString('hex');
}
(async () => {
const out = {};
const frags = [fragment(1, 300000), fragment(2, 800000), fragment(3, 250000)];
const stream = Buffer.concat([header, ...frags]);
out.expected = stream.toString('hex');
const relay = new CastRelay();
// The three first chunks measured on a real film.
out.ftypAlone = await cast(relay, [ftyp, moov, ...frags]);
out.headerExactly = await cast(relay, [header, ...frags]);
const cut = header.length + 4000;
out.headerAndFilm = await cast(relay, [stream.subarray(0, cut), stream.subarray(cut)]);
// A receiver connected before the first moof waits for the whole header.
out.earlyReceiver = await cast(relay, [ftyp, moov, ...frags], { early: true });
await relay.stop();
// A receiver that reads nothing while 40 MB of film is pushed, then reads.
const started = await relay.start({ initSegment: header });
let slowRes = null;
const parts = [];
const done = new Promise((resolve) => {
http.get(started.url, { agent: false }, (res) => {
slowRes = res; res.pause();
res.on('data', (c) => parts.push(c));
res.on('end', resolve);
}).on('error', () => {});
});
await sleep(100);
out.headersBeforeTheHeader = slowRes !== null;
relay.pushSegment(header);
const big = [];
for (let i = 0; i < 8; i++) big.push(fragment(i, 5 * 1024 * 1024));
for (const f of big) relay.pushSegment(f);
relay.finish();
slowRes.resume();
await done;
out.slowReceiverGotAll = Buffer.concat(parts).equals(Buffer.concat([header, ...big]));
out.dropped = relay._dropped;
// stop() not awaited, then start(), while a receiver still holds the old
// stream: the new relay keeps its own token, port and header.
const holder = read(relay.url, 300);
await sleep(50);
relay.stop();
const next = await relay.start({ initSegment: header });
await holder;
await sleep(100);
out.afterRestart = { url: relay.url, returned: next.url, active: relay.active };
relay.pushSegment(header);
relay.pushSegment(frags[0]);
const after = read(relay.url);
await sleep(50);
relay.finish();
out.afterRestartBody = (await after).toString('hex');
out.afterRestartExpected = Buffer.concat([header, frags[0]]).toString('hex');
await relay.stop();
out.trimmed = CastRelay.headerOnly(stream.subarray(0, cut)).toString('hex');
out.header = header.toString('hex');
console.log(JSON.stringify(out));
})().catch((e) => { console.error(e); process.exit(1); });
"""
@pytest.fixture(scope="module")
def run(tmp_path_factory):
if shutil.which("node") is None or not RELAY.exists():
pytest.skip("node or the desktop client sources are not available")
script = tmp_path_factory.mktemp("relay") / "stream.cjs"
script.write_text(SCRIPT, encoding="utf-8")
proc = subprocess.run(["node", str(script), str(RELAY)],
capture_output=True, text=True, encoding="utf-8", timeout=60)
assert proc.returncode == 0, proc.stderr
return json.loads(proc.stdout)
@pytest.mark.parametrize("first_chunk", ["ftypAlone", "headerExactly", "headerAndFilm"])
def test_the_header_is_everything_before_the_first_moof(run, first_chunk):
assert run[first_chunk] == run["expected"]
def test_a_receiver_that_connects_first_waits_for_the_whole_header(run):
assert run["earlyReceiver"] == run["expected"]
def test_a_waiting_receiver_is_answered_at_once(run):
assert run["headersBeforeTheHeader"] is True
def test_the_pages_first_chunk_is_trimmed_to_its_header(run):
assert run["trimmed"] == run["header"]
def test_a_receiver_far_behind_loses_no_fragment(run):
assert run["slowReceiverGotAll"] is True
assert run["dropped"] == 0
def test_a_stop_not_waited_for_leaves_the_next_start_alone(run):
r = run["afterRestart"]
assert r["active"] is True
assert r["url"] == r["returned"]
assert ":null" not in r["url"] and "t=null" not in r["url"]
assert run["afterRestartBody"] == run["afterRestartExpected"]
|