feat: synchronize peer-assisted quality profiles

This commit is contained in:
AAA committed 2026-08-19 02:55:42 +08:00
1 parent 855b023ae8
commit baa8bcab4a
10 files changed
+482 -93

No files matched your search

+14 -2
View File
@@ -1,3 +1,7 @@
import type { QualityProfileId } from "../../shared/protocol";
export type { QualityProfileId } from "../../shared/protocol";
export const QUALITY_PROFILES = {
"1080p60": {
label: "1080p 60",
@@ -20,9 +24,17 @@ export const QUALITY_PROFILES = {
frameRate: 30,
maxBitrate: 3_000_000,
},
} as const;
} as const satisfies Record<
QualityProfileId,
{
label: string;
width: number;
height: number;
frameRate: number;
maxBitrate: number;
}
>;
export type QualityProfileId = keyof typeof QUALITY_PROFILES;
export type QualityProfile = (typeof QUALITY_PROFILES)[QualityProfileId];
function captureConstraints(profile: QualityProfile): MediaTrackConstraints {
+20 -10
View File
@@ -10,10 +10,11 @@ import {
Users,
} from "lucide-react";
import { useEffect, useMemo, useRef, useState } from "react";
import type {
CreateRoomResponse,
IceConfig,
ServerMessage,
import {
DEFAULT_QUALITY_PROFILE_ID,
type CreateRoomResponse,
type IceConfig,
type ServerMessage,
} from "../../shared/protocol";
import { AppHeader } from "../components/AppHeader";
import {
@@ -109,7 +110,9 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
() => new URLSearchParams(window.location.search).get("relay") === "1",
[],
);
const [qualityId, setQualityId] = useState<QualityProfileId>("1080p60");
const [qualityId, setQualityId] = useState<QualityProfileId>(
DEFAULT_QUALITY_PROFILE_ID,
);
const [phase, setPhase] = useState<HostPhase>("idle");
const [signalStatus, setSignalStatus] =
useState<SignalConnectionState>("offline");
@@ -137,7 +140,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
const activeGenerationRef = useRef<number | null>(null);
const sourceSwitchRef = useRef<object | null>(null);
const qualityChangeRef = useRef<object | null>(null);
const qualityIdRef = useRef<QualityProfileId>("1080p60");
const qualityIdRef = useRef<QualityProfileId>(DEFAULT_QUALITY_PROFILE_ID);
const picturePausedRef = useRef(false);
const retiringStreamRef = useRef<MediaStream | null>(null);
@@ -302,6 +305,12 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
commitQuality(nextId);
setDetails(captureDetails(activeStream));
if (peerAssistedRef.current) {
signalRef.current?.send({
type: "set-quality-profile",
qualityProfileId: nextId,
});
}
const results = await Promise.all(
[...peersRef.current.values()].map((peer) => peer.updateProfile(profile)),
);
@@ -507,6 +516,10 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
message.mediaMode === "peer-assisted"
) {
peerAssistedRef.current = true;
signalRef.current?.send({
type: "set-quality-profile",
qualityProfileId: qualityIdRef.current,
});
reconcilePeerAssistedChildren(
message.mediaAssignment.childPeerIds,
generation,
@@ -812,10 +825,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
return {
peerId,
peer,
replaced: await peer.replaceStream(
captured,
QUALITY_PROFILES[qualityIdRef.current],
),
replaced: await peer.replaceStream(captured),
};
} catch {
return { peerId, peer, replaced: false };
+25 -7
View File
@@ -7,10 +7,11 @@ import {
VolumeX,
} from "lucide-react";
import { useEffect, useMemo, useRef, useState } from "react";
import type {
IceConfig,
MediaAssignment,
ServerMessage,
import {
DEFAULT_QUALITY_PROFILE_ID,
type IceConfig,
type MediaAssignment,
type ServerMessage,
} from "../../shared/protocol";
import { AppHeader } from "../components/AppHeader";
import {
@@ -22,7 +23,10 @@ import {
import { StatsGrid } from "../components/StatsGrid";
import { getStableClientId } from "../lib/session";
import { SignalingClient } from "../lib/signaling";
import { QUALITY_PROFILES } from "../media/quality";
import {
QUALITY_PROFILES,
type QualityProfileId,
} from "../media/quality";
import type {
PeerSnapshot,
SignalConnectionState,
@@ -65,6 +69,7 @@ export function ViewerPage({ roomId, onAuthorizationRequired }: ViewerPageProps)
let currentIceConfig: IceConfig | null = null;
let currentHostOnline = false;
let peerAssisted = false;
let currentQualityProfileId: QualityProfileId = DEFAULT_QUALITY_PROFILE_ID;
let currentAssignment: MediaAssignment = {
parentPeerId: null,
childPeerIds: [],
@@ -118,7 +123,7 @@ export function ViewerPage({ roomId, onAuthorizationRequired }: ViewerPageProps)
}
viewerRelay = new ViewerRelay(
currentIceConfig,
QUALITY_PROFILES["1080p60"],
QUALITY_PROFILES[currentQualityProfileId],
{
sendSignal: (targetPeerId, payload) =>
active &&
@@ -249,7 +254,11 @@ export function ViewerPage({ roomId, onAuthorizationRequired }: ViewerPageProps)
currentHostOnline = message.hostOnline;
setRelayAvailable(message.iceConfig.relayAvailable);
setHostOnline(message.hostOnline);
if (nextPeerAssisted) {
if (nextPeerAssisted && "qualityProfileId" in message) {
currentQualityProfileId = message.qualityProfileId;
void viewerRelay?.updateProfile(
QUALITY_PROFILES[currentQualityProfileId],
);
applyMediaAssignment(message.mediaAssignment);
}
if (
@@ -295,6 +304,15 @@ export function ViewerPage({ roomId, onAuthorizationRequired }: ViewerPageProps)
}
return;
}
if (message.type === "quality-profile") {
if (peerAssisted) {
currentQualityProfileId = message.qualityProfileId;
void viewerRelay?.updateProfile(
QUALITY_PROFILES[currentQualityProfileId],
);
}
return;
}
if (message.type === "signal") {
if (
peerAssisted &&
+80 -60
View File
@@ -33,13 +33,14 @@ export class HostPeer {
private statsTimer: number | null = null;
private disposed = false;
private negotiating = false;
private senderMutationTail: Promise<void> = Promise.resolve();
private snapshot: PeerSnapshot;
constructor(
readonly peerId: string,
iceConfig: IceConfig,
private stream: MediaStream,
private profile: QualityProfile,
private desiredProfile: QualityProfile,
private readonly events: HostPeerEvents,
private readonly forceRelay = false,
) {
@@ -74,7 +75,12 @@ export class HostPeer {
streams: [this.stream],
}).sender;
try {
await configureVideoSender(this.videoSender, this.profile);
await this.enqueueSenderMutation(async () => {
if (this.disposed || !this.videoSender) {
return;
}
await configureVideoSender(this.videoSender, this.desiredProfile);
});
} catch (error) {
console.warn("Browser rejected preferred sender parameters", error);
}
@@ -88,74 +94,79 @@ export class HostPeer {
return true;
}
async replaceStream(
nextStream: MediaStream,
profile: QualityProfile,
): Promise<boolean> {
async replaceStream(nextStream: MediaStream): Promise<boolean> {
const nextVideoTrack = nextStream.getVideoTracks()[0];
if (
this.disposed ||
!this.videoSender ||
!this.audioSender ||
!nextVideoTrack
) {
if (this.disposed || !nextVideoTrack) {
return false;
}
const nextAudioTrack = nextStream.getAudioTracks()[0] ?? null;
const previousVideoTrack = this.videoSender.track;
const previousAudioTrack = this.audioSender.track;
return this.enqueueSenderMutation(async () => {
const videoSender = this.videoSender;
const audioSender = this.audioSender;
if (this.disposed || !videoSender || !audioSender) {
return false;
}
try {
await this.videoSender.replaceTrack(nextVideoTrack);
await this.audioSender.replaceTrack(nextAudioTrack);
} catch (error) {
await Promise.allSettled([
this.videoSender.replaceTrack(previousVideoTrack),
this.audioSender.replaceTrack(previousAudioTrack),
]);
this.setError(error, "切换共享源失败");
return false;
}
const nextAudioTrack = nextStream.getAudioTracks()[0] ?? null;
const previousVideoTrack = videoSender.track;
const previousAudioTrack = audioSender.track;
if (this.disposed) {
return false;
}
this.stream = nextStream;
this.profile = profile;
this.statsAccumulator.bytes = null;
this.statsAccumulator.frames = null;
this.statsAccumulator.timestamp = null;
try {
await configureVideoSender(this.videoSender, this.profile);
} catch (error) {
console.warn("Browser rejected preferred sender parameters", error);
}
this.snapshot = { ...this.snapshot, error: null };
this.emit();
return true;
try {
await videoSender.replaceTrack(nextVideoTrack);
await audioSender.replaceTrack(nextAudioTrack);
} catch (error) {
await Promise.allSettled([
videoSender.replaceTrack(previousVideoTrack),
audioSender.replaceTrack(previousAudioTrack),
]);
this.setError(error, "切换共享源失败");
return false;
}
if (this.disposed) {
return false;
}
this.stream = nextStream;
this.statsAccumulator.bytes = null;
this.statsAccumulator.frames = null;
this.statsAccumulator.timestamp = null;
try {
await configureVideoSender(videoSender, this.desiredProfile);
} catch (error) {
console.warn("Browser rejected preferred sender parameters", error);
}
this.snapshot = { ...this.snapshot, error: null };
this.emit();
return true;
});
}
async updateProfile(profile: QualityProfile): Promise<boolean> {
if (this.disposed || !this.videoSender) {
return false;
}
try {
await configureVideoSender(this.videoSender, profile);
} catch (error) {
this.setError(error, "调整画质失败");
return false;
}
updateProfile(profile: QualityProfile): Promise<boolean> {
if (this.disposed) {
return false;
return Promise.resolve(false);
}
this.profile = profile;
this.statsAccumulator.bytes = null;
this.statsAccumulator.frames = null;
this.statsAccumulator.timestamp = null;
this.snapshot = { ...this.snapshot, error: null };
this.emit();
return true;
this.desiredProfile = profile;
return this.enqueueSenderMutation(async () => {
const videoSender = this.videoSender;
if (this.disposed || !videoSender) {
return false;
}
try {
await configureVideoSender(videoSender, this.desiredProfile);
} catch (error) {
this.setError(error, "调整画质失败");
return false;
}
if (this.disposed) {
return false;
}
this.statsAccumulator.bytes = null;
this.statsAccumulator.frames = null;
this.statsAccumulator.timestamp = null;
this.snapshot = { ...this.snapshot, error: null };
this.emit();
return true;
});
}
async acceptSignal(payload: SignalPayload): Promise<void> {
@@ -242,6 +253,15 @@ export class HostPeer {
this.connection.addEventListener("iceconnectionstatechange", () => this.emit());
}
private enqueueSenderMutation<T>(operation: () => Promise<T>): Promise<T> {
const result = this.senderMutationTail.then(operation);
this.senderMutationTail = result.then(
() => undefined,
() => undefined,
);
return result;
}
private async createOffer(restart: boolean): Promise<boolean> {
if (this.disposed) {
return false;
+27 -3
View File
@@ -18,7 +18,7 @@ export class ViewerRelay {
constructor(
private iceConfig: IceConfig,
private readonly profile: QualityProfile,
private desiredProfile: QualityProfile,
private readonly events: ViewerRelayEvents,
private readonly forceRelay = false,
) {}
@@ -61,6 +61,19 @@ export class ViewerRelay {
.catch(() => undefined);
}
updateProfile(profile: QualityProfile): Promise<boolean> {
if (this.disposed) {
return Promise.resolve(false);
}
this.desiredProfile = profile;
const result = this.syncQueue.then(() => this.syncProfile(profile));
this.syncQueue = result.then(
() => undefined,
() => undefined,
);
return result.catch(() => false);
}
async acceptSignal(
fromPeerId: string,
payload: SignalPayload,
@@ -134,7 +147,7 @@ export class ViewerRelay {
let replaced = false;
try {
replaced = await peer.replaceStream(stream, this.profile);
replaced = await peer.replaceStream(stream);
} catch {
// Rebuilding below is safer than leaving a stopped upstream track.
}
@@ -151,6 +164,17 @@ export class ViewerRelay {
this.startPeer(childPeerId, stream);
}
private async syncProfile(profile: QualityProfile): Promise<boolean> {
if (this.disposed || this.desiredProfile !== profile) {
return false;
}
const peer = this.peer;
if (!peer) {
return true;
}
return peer.updateProfile(this.desiredProfile);
}
private startPeer(
childPeerId: string,
stream: MediaStream,
@@ -170,7 +194,7 @@ export class ViewerRelay {
childPeerId,
this.iceConfig,
stream,
this.profile,
this.desiredProfile,
{
sendSignal: (targetPeerId, payload) =>
!this.disposed &&
+33
View File
@@ -5,9 +5,11 @@ import type { Duplex } from "node:stream";
import WebSocket, { WebSocketServer } from "ws";
import {
DEFAULT_QUALITY_PROFILE_ID,
MAX_SIGNAL_BYTES,
decodeClientMessage,
type ClientMessage,
type QualityProfileId,
type Role,
type ServerMessage,
} from "../shared/protocol.js";
@@ -65,6 +67,10 @@ export class SignalingServer {
private readonly socketsBySessionId = new Map<string, WebSocket>();
private readonly viewerGraceTimers = new Map<string, NodeJS.Timeout>();
private readonly connectionIdsByViewer = new Map<string, string>();
private readonly qualityProfileIdsByRoom = new Map<
string,
QualityProfileId
>();
private readonly peerRelayTopology = new PeerRelayTopology();
private readonly now: () => number;
private readonly authenticationTimeoutMs: number;
@@ -348,6 +354,9 @@ export class SignalingServer {
...authenticatedMessage,
mediaMode: "peer-assisted",
mediaAssignment,
qualityProfileId:
this.qualityProfileIdsByRoom.get(participant.roomId) ??
DEFAULT_QUALITY_PROFILE_ID,
});
} else {
this.send(socket, authenticatedMessage);
@@ -446,6 +455,28 @@ export class SignalingServer {
),
});
return;
case "set-quality-profile":
if (!this.options.peerAssistedMedia || authenticated.role !== "host") {
this.sendError(
socket,
"FORBIDDEN",
"Only a peer-assisted host may set the quality profile",
);
return;
}
this.qualityProfileIdsByRoom.set(
authenticated.roomId,
message.qualityProfileId,
);
for (const viewer of this.options.roomStore.getConnectedViewers(
authenticated.roomId,
)) {
this.sendToSession(viewer.sessionId, {
type: "quality-profile",
qualityProfileId: message.qualityProfileId,
});
}
return;
case "stop-sharing":
case "close-room":
if (authenticated.role !== "host") {
@@ -729,6 +760,7 @@ export class SignalingServer {
this.clearRoomGraceTimers(roomId);
this.clearRoomConnectionIds(roomId);
this.peerRelayTopology.deleteRoom(roomId);
this.qualityProfileIdsByRoom.delete(roomId);
for (const sessionId of abandoned.sessionIds) {
const socket = this.socketsBySessionId.get(sessionId);
if (!socket) {
@@ -744,6 +776,7 @@ export class SignalingServer {
this.clearRoomGraceTimers(expired.roomId);
this.clearRoomConnectionIds(expired.roomId);
this.peerRelayTopology.deleteRoom(expired.roomId);
this.qualityProfileIdsByRoom.delete(expired.roomId);
for (const sessionId of expired.sessionIds) {
const socket = this.socketsBySessionId.get(sessionId);
if (!socket) {
+21
View File
@@ -25,6 +25,14 @@ export const roomCodeSchema = z
export const roleSchema = z.enum(["host", "viewer"]);
export type Role = z.infer<typeof roleSchema>;
export const qualityProfileIdSchema = z.enum([
"1080p60",
"1080p30",
"720p30",
]);
export type QualityProfileId = z.infer<typeof qualityProfileIdSchema>;
export const DEFAULT_QUALITY_PROFILE_ID: QualityProfileId = "1080p60";
const iceServerSchema = z
.object({
urls: z.union([
@@ -125,6 +133,12 @@ export const clientMessageSchema = z.union([
})
.strict(),
z.object({ type: z.literal("refresh-ice") }).strict(),
z
.object({
type: z.literal("set-quality-profile"),
qualityProfileId: qualityProfileIdSchema,
})
.strict(),
z.object({ type: z.literal("stop-sharing") }).strict(),
// Kept as a compatibility alias while previously deployed clients age out.
z.object({ type: z.literal("close-room") }).strict(),
@@ -163,6 +177,7 @@ const authenticatedMessageSchema = z.union([
...authenticatedMessageShape,
mediaMode: z.literal("peer-assisted"),
mediaAssignment: mediaAssignmentSchema,
qualityProfileId: qualityProfileIdSchema,
})
.strict(),
]);
@@ -202,6 +217,12 @@ export const serverMessageSchema = z.union([
mediaAssignment: mediaAssignmentSchema,
})
.strict(),
z
.object({
type: z.literal("quality-profile"),
qualityProfileId: qualityProfileIdSchema,
})
.strict(),
z
.object({
type: z.literal("ice-config"),
+106 -11
View File
@@ -9,12 +9,22 @@ const statsCallbacks: Array<() => void> = [];
class FakeSender {
failNextReplace = false;
failNextSetParameters = false;
deferNextSetParameters = false;
readonly appliedMaxBitrates: Array<number | undefined> = [];
private releaseSetParameters: (() => void) | null = null;
readonly setParameters = vi.fn(
async (_parameters: RTCRtpSendParameters) => {
async (parameters: RTCRtpSendParameters) => {
if (this.failNextSetParameters) {
this.failNextSetParameters = false;
throw new Error("setParameters failed");
}
if (this.deferNextSetParameters) {
this.deferNextSetParameters = false;
await new Promise<void>((resolve) => {
this.releaseSetParameters = resolve;
});
}
this.appliedMaxBitrates.push(parameters.encodings[0]?.maxBitrate);
},
);
readonly replaceTrack = vi.fn(async (track: MediaStreamTrack | null) => {
@@ -30,6 +40,11 @@ class FakeSender {
getParameters(): RTCRtpSendParameters {
return { encodings: [{}] } as RTCRtpSendParameters;
}
releaseDeferredSetParameters(): void {
this.releaseSetParameters?.();
this.releaseSetParameters = null;
}
}
class FakePeerConnection {
@@ -246,10 +261,7 @@ describe("HostPeer source replacement", () => {
const nextVideo = createTrack("video", "next-video");
const nextAudio = createTrack("audio", "next-audio");
await expect(
peer.replaceStream(
createStream(nextVideo, nextAudio),
QUALITY_PROFILES["1080p60"],
),
peer.replaceStream(createStream(nextVideo, nextAudio)),
).resolves.toBe(true);
expect(connection.senders[0]?.track).toBe(nextVideo);
@@ -272,7 +284,6 @@ describe("HostPeer source replacement", () => {
await expect(
peer.replaceStream(
createStream(createTrack("video", "next-video"), nextAudio),
QUALITY_PROFILES["1080p30"],
),
).resolves.toBe(true);
expect(connection.senders[1]?.track).toBe(nextAudio);
@@ -291,7 +302,6 @@ describe("HostPeer source replacement", () => {
await expect(
peer.replaceStream(
createStream(createTrack("video", "next-video"), null),
QUALITY_PROFILES["720p30"],
),
).resolves.toBe(true);
@@ -320,6 +330,55 @@ describe("HostPeer source replacement", () => {
});
});
it("serializes initial sender configuration with a live profile update", async () => {
const peer = createPeer(
createStream(
createTrack("video", "video"),
createTrack("audio", "audio"),
),
);
const starting = peer.start();
const videoSender = FakePeerConnection.latest!.senders[0]!;
videoSender.deferNextSetParameters = true;
await vi.waitFor(() =>
expect(videoSender.setParameters).toHaveBeenCalledTimes(1),
);
const updating = peer.updateProfile(QUALITY_PROFILES["1080p60"]);
await Promise.resolve();
expect(videoSender.setParameters).toHaveBeenCalledTimes(1);
videoSender.releaseDeferredSetParameters();
await expect(starting).resolves.toBe(true);
await expect(updating).resolves.toBe(true);
expect(videoSender.appliedMaxBitrates).toEqual([3_000_000, 8_000_000]);
});
it("continues queued profile updates after initial configuration rejects", async () => {
const warn = vi.spyOn(console, "warn").mockImplementation(() => undefined);
try {
const peer = createPeer(
createStream(
createTrack("video", "video"),
createTrack("audio", "audio"),
),
);
const starting = peer.start();
const videoSender = FakePeerConnection.latest!.senders[0]!;
videoSender.failNextSetParameters = true;
const updating = peer.updateProfile(QUALITY_PROFILES["1080p60"]);
await expect(starting).resolves.toBe(true);
await expect(updating).resolves.toBe(true);
expect(videoSender.setParameters).toHaveBeenCalledTimes(2);
expect(videoSender.appliedMaxBitrates).toEqual([8_000_000]);
} finally {
warn.mockRestore();
}
});
it("can retry the selected quality after a sender update fails", async () => {
const peer = createPeer(
createStream(
@@ -359,10 +418,7 @@ describe("HostPeer source replacement", () => {
const nextVideo = createTrack("video", "next-video");
const nextAudio = createTrack("audio", "next-audio");
await expect(
peer.replaceStream(
createStream(nextVideo, nextAudio),
QUALITY_PROFILES["1080p60"],
),
peer.replaceStream(createStream(nextVideo, nextAudio)),
).resolves.toBe(false);
expect(videoSender.replaceTrack).toHaveBeenNthCalledWith(1, nextVideo);
@@ -630,4 +686,43 @@ describe("ViewerRelay downstream ownership", () => {
expect(connection.connectionState).toBe("closed");
await vi.waitFor(() => expect(targets.at(-1)).toBe("child-peer-two"));
});
it("applies the latest profile to the current and future child", async () => {
const relay = new ViewerRelay(
{ iceServers: [], expiresAt: null, relayAvailable: false },
QUALITY_PROFILES["1080p60"],
{ sendSignal: () => true },
);
relay.setChild("first-profile-child");
relay.setStream(createStream(createTrack("video", "profile-video"), null));
await vi.waitFor(() =>
expect(
FakePeerConnection.latest?.senders[0]?.setParameters,
).toHaveBeenCalled(),
);
const firstConnection = FakePeerConnection.latest!;
await expect(
relay.updateProfile(QUALITY_PROFILES["720p30"]),
).resolves.toBe(true);
expect(
firstConnection.senders[0]?.setParameters.mock.calls.at(-1)?.[0],
).toMatchObject({
encodings: [{ maxBitrate: 3_000_000, maxFramerate: 30 }],
});
relay.setChild("second-profile-child");
await vi.waitFor(() =>
expect(FakePeerConnection.latest).not.toBe(firstConnection),
);
const secondConnection = FakePeerConnection.latest!;
await vi.waitFor(() =>
expect(secondConnection.senders[0]?.setParameters).toHaveBeenCalled(),
);
expect(
secondConnection.senders[0]?.setParameters.mock.calls[0]?.[0],
).toMatchObject({
encodings: [{ maxBitrate: 3_000_000, maxFramerate: 30 }],
});
});
});
+55
View File
@@ -132,6 +132,28 @@ describe("client signaling protocol", () => {
true,
);
});
it("accepts only a bounded quality profile update", () => {
expect(
clientMessageSchema.safeParse({
type: "set-quality-profile",
qualityProfileId: "1080p30",
}).success,
).toBe(true);
expect(
clientMessageSchema.safeParse({
type: "set-quality-profile",
qualityProfileId: "1440p60",
}).success,
).toBe(false);
expect(
clientMessageSchema.safeParse({
type: "set-quality-profile",
qualityProfileId: "1080p30",
bitrate: 5_000_000,
}).success,
).toBe(false);
});
});
describe("server signaling protocol", () => {
@@ -187,6 +209,7 @@ describe("server signaling protocol", () => {
parentPeerId: null,
childPeerIds: ["viewer_12345678", "viewer_87654321"],
},
qualityProfileId: "1080p60",
};
expect(serverMessageSchema.safeParse(peerAssisted).success).toBe(true);
@@ -200,6 +223,14 @@ describe("server signaling protocol", () => {
serverMessageSchema.safeParse({
...authenticatedMessage(8),
mediaMode: "peer-assisted",
qualityProfileId: "1080p60",
}).success,
).toBe(false);
expect(
serverMessageSchema.safeParse({
...authenticatedMessage(8),
mediaMode: "peer-assisted",
mediaAssignment: peerAssisted.mediaAssignment,
}).success,
).toBe(false);
expect(
@@ -215,6 +246,30 @@ describe("server signaling protocol", () => {
},
}).success,
).toBe(false);
expect(
serverMessageSchema.safeParse({
...peerAssisted,
qualityProfileId: "1440p60",
}).success,
).toBe(false);
expect(
serverMessageSchema.safeParse({
...authenticatedMessage(8),
qualityProfileId: "1080p60",
}).success,
).toBe(false);
expect(
serverMessageSchema.safeParse({
type: "quality-profile",
qualityProfileId: "720p30",
}).success,
).toBe(true);
expect(
serverMessageSchema.safeParse({
type: "quality-profile",
qualityProfileId: "1440p60",
}).success,
).toBe(false);
});
it.each([0, 1.5, MAX_VIEWERS_PER_ROOM_LIMIT + 1])(
+101
View File
@@ -328,6 +328,94 @@ describe("WebSocket signaling", () => {
expect(authenticated.hostOnline).toBe(false);
});
it("synchronizes a bounded quality profile across a peer-assisted room", async () => {
const harness = await startHarness({ peerAssistedMedia: true });
const host = await openClient(harness.webSocketUrl);
const hostAuth = peerAssisted(
await authenticate(host, harness.room, "host", "quality-host"),
);
expect(hostAuth.qualityProfileId).toBe("1080p60");
const viewer = await openClient(harness.webSocketUrl);
const viewerAuth = peerAssisted(
await authenticate(viewer, harness.room, "viewer", "quality-viewer"),
);
expect(viewerAuth.qualityProfileId).toBe("1080p60");
await host.inbox.next("media-assignment");
host.socket.send(
JSON.stringify({
type: "set-quality-profile",
qualityProfileId: "1080p30",
}),
);
expect(await viewer.inbox.next("quality-profile")).toEqual({
type: "quality-profile",
qualityProfileId: "1080p30",
});
const lateViewer = await openClient(harness.webSocketUrl);
const lateViewerAuth = peerAssisted(
await authenticate(
lateViewer,
harness.room,
"viewer",
"quality-viewer-late",
),
);
expect(lateViewerAuth.qualityProfileId).toBe("1080p30");
await host.inbox.next("media-assignment");
viewer.socket.send(
JSON.stringify({
type: "set-quality-profile",
qualityProfileId: "720p30",
}),
);
expect((await viewer.inbox.next("error")).code).toBe("FORBIDDEN");
await lateViewer.inbox.expectNone(30);
});
it("keeps the P2P authenticated wire unchanged and forbids profile updates", async () => {
const harness = await startHarness();
const host = await openClient(harness.webSocketUrl);
const hostAuth = await authenticate(
host,
harness.room,
"host",
"p2p-quality-host",
);
expect("mediaMode" in hostAuth).toBe(false);
expect("qualityProfileId" in hostAuth).toBe(false);
const viewer = await openClient(harness.webSocketUrl);
const viewerAuth = await authenticate(
viewer,
harness.room,
"viewer",
"p2p-quality-viewer",
);
expect("mediaMode" in viewerAuth).toBe(false);
expect("qualityProfileId" in viewerAuth).toBe(false);
await host.inbox.next("peer-joined");
host.socket.send(
JSON.stringify({
type: "set-quality-profile",
qualityProfileId: "1080p30",
}),
);
expect((await host.inbox.next("error")).code).toBe("FORBIDDEN");
viewer.socket.send(
JSON.stringify({
type: "set-quality-profile",
qualityProfileId: "720p30",
}),
);
expect((await viewer.inbox.next("error")).code).toBe("FORBIDDEN");
await viewer.inbox.expectNone(30);
});
it("routes offer and answer only between the host and the targeted viewer", async () => {
const harness = await startHarness();
const secondRoom = harness.roomStore.createRoom();
@@ -804,6 +892,17 @@ describe("WebSocket signaling", () => {
);
await host.inbox.next("media-assignment");
host.socket.send(
JSON.stringify({
type: "set-quality-profile",
qualityProfileId: "720p30",
}),
);
expect(await viewer.inbox.next("quality-profile")).toEqual({
type: "quality-profile",
qualityProfileId: "720p30",
});
host.socket.send(
JSON.stringify({
type: "signal",
@@ -844,6 +943,7 @@ describe("WebSocket signaling", () => {
parentPeerId: hostAuth.peerId,
childPeerIds: [],
},
qualityProfileId: "720p30",
});
const resumedHost = await openClient(harness.webSocketUrl);
@@ -861,6 +961,7 @@ describe("WebSocket signaling", () => {
parentPeerId: null,
childPeerIds: [viewerAuth.peerId],
},
qualityProfileId: "720p30",
});
expect(await reconnectedViewer.inbox.next("host-status")).toMatchObject({
online: true,