diff --git a/src/core/player.ts b/src/core/player.ts index 8920670..a021db9 100644 --- a/src/core/player.ts +++ b/src/core/player.ts @@ -379,7 +379,15 @@ export class GuildPlayer extends EventEmitter { private async startScreenShare(video: NonNullable): Promise { const room = this.connection?.room; if (!room) return; - const publisher = new VideoPublisher({ width: video.width, height: video.height }); + const publisher = new VideoPublisher({ + width: video.width, + height: video.height, + fps: video.fps, + // Frames are released against the sound that has actually played, which is + // the only clock both sides share: the audio path runs through revoice's + // own ffmpeg and buffer, so it always trails the raw frames. + audioClock: () => this.media?.seconds ?? 0, + }); try { await publisher.start(room, video.stream, (reason) => { this.notify(`📺 Видео отключено: ${reason}.`); diff --git a/src/core/video.ts b/src/core/video.ts index 03923fd..08c0c13 100644 --- a/src/core/video.ts +++ b/src/core/video.ts @@ -1,143 +1,218 @@ -import { createRequire } from "node:module"; -import type { Readable } from "node:stream"; -import type { LocalVideoTrack, Room, VideoSource } from "@livekit/rtc-node"; -import { config } from "../config.js"; -import { logger } from "../logger.js"; - -// revoice.js pulls in the CommonJS build of @livekit/rtc-node, and the native -// FFI client is per-module-instance: importing the ESM build here would give us -// a second instance whose tracks the room (created by revoice) cannot publish. -// So we require the very same module object it uses. -const require = createRequire(import.meta.url); -const rtc = require("@livekit/rtc-node") as typeof import("@livekit/rtc-node"); - -const log = logger.child({ mod: "video" }); -/** Give up on the video track if the source never produced a picture. */ -const FIRST_FRAME_TIMEOUT_MS = 15_000; - -export interface VideoPublisherOptions { - width: number; - height: number; -} - -/** - * Publishes a raw I420 stream into a LiveKit room as a screen share. ffmpeg - * feeds frames at playback speed, so this class only has to slice the byte - * stream into frames and hand them over. - */ -export class VideoPublisher { - private readonly frameSize: number; - private source: VideoSource | null = null; - private track: LocalVideoTrack | null = null; - private trackSid: string | null = null; - private room: Room | null = null; - private stream: Readable | null = null; - private pending: Buffer[] = []; - private pendingBytes = 0; - private frames = 0; - private firstFrameTimer: NodeJS.Timeout | null = null; - - constructor(private readonly options: VideoPublisherOptions) { - // I420: one luma plane plus two half-resolution chroma planes. - this.frameSize = Math.floor((options.width * options.height * 3) / 2); - } - - get isPublishing(): boolean { - return this.trackSid !== null; - } - - /** Publishes the track and starts pumping frames. Resolves once published. */ - async start(room: Room, stream: Readable, onGiveUp?: (reason: string) => void): Promise { - const { width, height } = this.options; - const source = new rtc.VideoSource(width, height); - // As a camera the picture appears inside the bot's own tile; a screen share - // is rendered as a separate tile by every client. - const asScreen = config.VIDEO_SOURCE === "screen"; - const track = rtc.LocalVideoTrack.createVideoTrack(asScreen ? "screen" : "video", source); - - const options = new rtc.TrackPublishOptions(); - options.source = asScreen ? rtc.TrackSource.SOURCE_SCREENSHARE : rtc.TrackSource.SOURCE_CAMERA; - - const participant = room.localParticipant; - if (!participant) throw new Error("room has no local participant yet"); - const publication = await participant.publishTrack(track, options); - - this.room = room; - this.source = source; - this.track = track; - this.trackSid = publication.sid ?? null; - this.stream = stream; - - stream.on("data", (chunk: Buffer) => this.consume(chunk)); - stream.once("error", (err) => log.debug({ err }, "video stream error")); - - this.firstFrameTimer = setTimeout(() => { - if (this.frames > 0) return; - log.warn("no video frames arrived, dropping the screen share"); - onGiveUp?.("источник не отдал видео"); - void this.stop(); - }, FIRST_FRAME_TIMEOUT_MS); - this.firstFrameTimer.unref?.(); - - log.info( - { width, height, source: config.VIDEO_SOURCE, sid: publication.sid }, - "video track published", - ); - } - - private consume(chunk: Buffer): void { - if (!this.source) return; - this.pending.push(chunk); - this.pendingBytes += chunk.length; - if (this.pendingBytes < this.frameSize) return; - - let buffer = this.pending.length === 1 ? (this.pending[0] as Buffer) : Buffer.concat(this.pending, this.pendingBytes); - while (buffer.length >= this.frameSize) { - const frame = buffer.subarray(0, this.frameSize); - try { - this.source.captureFrame( - new rtc.VideoFrame(frame, this.options.width, this.options.height, rtc.VideoBufferType.I420), - ); - this.frames += 1; - } catch (err) { - log.debug({ err }, "captureFrame failed"); - } - buffer = buffer.subarray(this.frameSize); - } - // Keep only the tail of a partial frame for the next chunk. - this.pending = buffer.length > 0 ? [Buffer.from(buffer)] : []; - this.pendingBytes = buffer.length; - } - - async stop(): Promise { - if (this.firstFrameTimer) { - clearTimeout(this.firstFrameTimer); - this.firstFrameTimer = null; - } - this.stream?.removeAllListeners("data"); - this.stream = null; - this.pending = []; - this.pendingBytes = 0; - - const { room, trackSid } = this; - this.trackSid = null; - if (room?.localParticipant && trackSid) { - try { - await room.localParticipant.unpublishTrack(trackSid, true); - } catch (err) { - log.debug({ err }, "unpublishTrack failed"); - } - } - - try { - this.track?.close?.(); - await this.source?.close?.(); - } catch (err) { - log.debug({ err }, "closing video track failed"); - } - this.track = null; - this.source = null; - this.room = null; - this.frames = 0; - } -} +import { createRequire } from "node:module"; +import type { Readable } from "node:stream"; +import type { LocalVideoTrack, Room, VideoSource } from "@livekit/rtc-node"; +import { config } from "../config.js"; +import { logger } from "../logger.js"; + +// revoice.js pulls in the CommonJS build of @livekit/rtc-node, and the native +// FFI client is per-module-instance: importing the ESM build here would give us +// a second instance whose tracks the room (created by revoice) cannot publish. +// So we require the very same module object it uses. +const require = createRequire(import.meta.url); +const rtc = require("@livekit/rtc-node") as typeof import("@livekit/rtc-node"); + +const log = logger.child({ mod: "video" }); +/** Give up on the video track if the source never produced a picture. */ +const FIRST_FRAME_TIMEOUT_MS = 15_000; +/** How far behind the sound a frame may be before it is dropped instead of shown. */ +const MAX_LATE_SECONDS = 0.25; +/** + * Frames wait in memory until the sound catches up, so the queue is bounded by + * bytes rather than count — 720p frames are four times the size of 360p ones. + * Never by pausing the stream: one ffmpeg feeds both outputs, so blocking the + * video pipe would also stop the audio whose clock we are waiting for. + */ +const MAX_QUEUE_BYTES = 48 * 1024 * 1024; + +export interface VideoPublisherOptions { + width: number; + height: number; + fps: number; + /** + * Seconds of audio actually played out so far. Frames are released against + * this clock, which is what keeps picture and sound together. + */ + audioClock(): number; +} + +interface QueuedFrame { + /** Position in the stream, so a dropped frame does not shift the rest. */ + index: number; + data: Buffer; +} + +/** + * Publishes a raw I420 stream into a LiveKit room. + * + * Frames are not handed over as they arrive: the audio takes a longer road to + * the call (through revoice's own ffmpeg and its buffer), so a frame released on + * arrival runs ahead of the sound. Instead every frame carries its position in + * the stream and waits until the audio clock reaches it. + */ +export class VideoPublisher { + private readonly frameSize: number; + private source: VideoSource | null = null; + private track: LocalVideoTrack | null = null; + private trackSid: string | null = null; + private room: Room | null = null; + private stream: Readable | null = null; + private partial: Buffer[] = []; + private partialBytes = 0; + private queue: QueuedFrame[] = []; + private produced = 0; + private published = 0; + private dropped = 0; + private pump: NodeJS.Timeout | null = null; + private firstFrameTimer: NodeJS.Timeout | null = null; + + private readonly maxQueuedFrames: number; + + constructor(private readonly options: VideoPublisherOptions) { + // I420: one luma plane plus two half-resolution chroma planes. + this.frameSize = Math.floor((options.width * options.height * 3) / 2); + this.maxQueuedFrames = Math.max(12, Math.floor(MAX_QUEUE_BYTES / this.frameSize)); + } + + get isPublishing(): boolean { + return this.trackSid !== null; + } + + /** Publishes the track and starts pumping frames. Resolves once published. */ + async start(room: Room, stream: Readable, onGiveUp?: (reason: string) => void): Promise { + const { width, height } = this.options; + const source = new rtc.VideoSource(width, height); + // As a camera the picture appears inside the bot's own tile; a screen share + // is rendered as a separate tile by every client. + const asScreen = config.VIDEO_SOURCE === "screen"; + const track = rtc.LocalVideoTrack.createVideoTrack(asScreen ? "screen" : "video", source); + + const options = new rtc.TrackPublishOptions(); + options.source = asScreen ? rtc.TrackSource.SOURCE_SCREENSHARE : rtc.TrackSource.SOURCE_CAMERA; + + const participant = room.localParticipant; + if (!participant) throw new Error("room has no local participant yet"); + const publication = await participant.publishTrack(track, options); + + this.room = room; + this.source = source; + this.track = track; + this.trackSid = publication.sid ?? null; + this.stream = stream; + + stream.on("data", (chunk: Buffer) => this.consume(chunk)); + stream.once("error", (err) => log.debug({ err }, "video stream error")); + + // Checked often enough for smooth release at any sane frame rate. + this.pump = setInterval(() => this.release(), 5); + this.pump.unref?.(); + + this.firstFrameTimer = setTimeout(() => { + if (this.published > 0) return; + log.warn("no video frames arrived, dropping the video track"); + onGiveUp?.("источник не отдал видео"); + void this.stop(); + }, FIRST_FRAME_TIMEOUT_MS); + this.firstFrameTimer.unref?.(); + + log.info( + { width, height, source: config.VIDEO_SOURCE, sid: publication.sid }, + "video track published", + ); + } + + private consume(chunk: Buffer): void { + if (!this.source) return; + this.partial.push(chunk); + this.partialBytes += chunk.length; + if (this.partialBytes < this.frameSize) return; + + let buffer = + this.partial.length === 1 ? (this.partial[0] as Buffer) : Buffer.concat(this.partial, this.partialBytes); + while (buffer.length >= this.frameSize) { + this.queue.push({ index: this.produced, data: Buffer.from(buffer.subarray(0, this.frameSize)) }); + this.produced += 1; + buffer = buffer.subarray(this.frameSize); + } + // Keep only the tail of a partial frame for the next chunk. + this.partial = buffer.length > 0 ? [Buffer.from(buffer)] : []; + this.partialBytes = buffer.length; + + // The queue grows while the audio buffer fills; beyond that, something is + // wrong and holding more frames would only cost memory. + while (this.queue.length > this.maxQueuedFrames) { + this.queue.shift(); + this.dropped += 1; + } + this.release(); + } + + /** Releases every frame the sound has already caught up with. */ + private release(): void { + if (!this.source) return; + const { fps } = this.options; + const now = this.options.audioClock(); + const tolerance = 1 / (2 * fps); + + while (this.queue.length > 0) { + const frame = this.queue[0] as QueuedFrame; + const dueAt = frame.index / fps; + if (now + tolerance < dueAt) break; + + this.queue.shift(); + if (now > dueAt + MAX_LATE_SECONDS) { + this.dropped += 1; + continue; + } + try { + this.source.captureFrame( + new rtc.VideoFrame(frame.data, this.options.width, this.options.height, rtc.VideoBufferType.I420), + ); + this.published += 1; + } catch (err) { + log.debug({ err }, "captureFrame failed"); + } + } + } + + async stop(): Promise { + for (const timer of [this.firstFrameTimer, this.pump]) { + if (timer) clearTimeout(timer as NodeJS.Timeout); + } + if (this.pump) clearInterval(this.pump); + this.firstFrameTimer = null; + this.pump = null; + + this.stream?.removeAllListeners("data"); + this.stream = null; + this.partial = []; + this.partialBytes = 0; + this.queue = []; + + if (this.published > 0) { + log.info({ published: this.published, dropped: this.dropped }, "video track finished"); + } + + const { room, trackSid } = this; + this.trackSid = null; + if (room?.localParticipant && trackSid) { + try { + await room.localParticipant.unpublishTrack(trackSid, true); + } catch (err) { + log.debug({ err }, "unpublishTrack failed"); + } + } + + try { + this.track?.close?.(); + await this.source?.close?.(); + } catch (err) { + log.debug({ err }, "closing video track failed"); + } + this.track = null; + this.source = null; + this.room = null; + this.produced = 0; + this.published = 0; + this.dropped = 0; + } +} diff --git a/src/sources/index.ts b/src/sources/index.ts index bccbb28..4e79cec 100644 --- a/src/sources/index.ts +++ b/src/sources/index.ts @@ -169,8 +169,8 @@ export interface PlaybackInput { cleanup(): void; /** Resolves with a reason if the downloader died on its own, for reporting. */ failure?: Promise; - /** Raw I420 frames to publish as a screen share, when video is on. */ - video?: { stream: Readable; width: number; height: number }; + /** Raw I420 frames to publish alongside the sound, when video is on. */ + video?: { stream: Readable; width: number; height: number; fps: number }; } export interface PlaybackOptions { @@ -252,7 +252,12 @@ export async function openPlayback( input: pipeline.audio, inputOptions: pipeline.audioInputOptions, cleanup: () => pipeline.kill(), - video: { stream: pipeline.video, width: pipeline.width, height: pipeline.height }, + video: { + stream: pipeline.video, + width: pipeline.width, + height: pipeline.height, + fps: config.VIDEO_FPS, + }, }; }