From b7069ff101e3f79898500c390c6bb886c12e57ad Mon Sep 17 00:00:00 2001 From: Manish Date: Mon, 25 May 2026 17:06:34 +0530 Subject: [PATCH] feat(logger): add configurable scoped logger and migrate console.* calls MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add lib/common/logger.ts — a zero-dependency, single-dispatcher logger: - LogLevel type: "*" | "trace" | "debug" | "info" | "warn" | "error" | "none" - Logger interface: optional level() + per-level methods; library calls level() before every emit so consumers own their own filtering - Default: built-in console logger at "error" level - setGlobalLogger(logger) / setGlobalLogger(undefined) to install or restore - createConsoleLogger(level) helper for consumers who just want a different verbosity on the built-in output - notifyLoggerLevelChanged() to re-sync workers when level changes at runtime - getLogger() — called once at module level with no args; derives scope automatically from the call stack (e.g. "transport/connection"), returns a frozen ScopedLogger that closes over scope and calls the single dispatcher - Worker propagation: getWorkerLogger() / getWorkletLogger() for Web Worker and AudioWorklet; setWorkerLogLevel() / setWorkletLogLevel() wired to ToWorker.logLevel and worklet message.logLevel; installWorkerLogReceiver() / installWorkletLogReceiver() on the main thread route FromWorker.log records into the global logger; Backend sends initial log level and re-syncs on every setGlobalLogger call via onLoggerLevelChange() - All output prefixed with "[MoQJS]" and "[]" as leading args so consumers can filter or format independently Migrate all console.* call sites in lib/** to scoped loggers: - error: caught failures, decoder/encoder errors, fatal stream errors - warn: dropped chunks/frames, missing elements, unsupported APIs - debug: lifecycle transitions, setup details, subscribe/publish state - trace: per-object/per-frame chatty logs, protocol payload dumps Re-export public logger API (setGlobalLogger, getGlobalLogger, createConsoleLogger, notifyLoggerLevelChanged, Logger, LogLevel) from all three entry points so consumers of any bundle variant can configure logging. --- lib/common/index.ts | 11 + lib/common/logger.ts | 375 ++++++++++++++++++++++++ lib/contribute/audio.ts | 8 +- lib/contribute/broadcast.ts | 31 +- lib/contribute/container.ts | 9 +- lib/contribute/track.ts | 9 +- lib/contribute/video.ts | 6 +- lib/media/catalog/index.ts | 9 +- lib/media/mp4/parser.ts | 5 +- lib/moq-publisher/index.ts | 52 ++-- lib/playback/audio.ts | 32 ++- lib/playback/backend.ts | 29 +- lib/playback/index.ts | 55 ++-- lib/playback/worker/audio.ts | 9 +- lib/playback/worker/index.ts | 16 +- lib/playback/worker/message.ts | 7 + lib/playback/worker/video.ts | 25 +- lib/playback/worklet/index.ts | 9 +- lib/playback/worklet/message.ts | 8 + lib/transport/client.ts | 13 +- lib/transport/connection.ts | 17 +- lib/transport/control/client_setup.ts | 61 ++-- lib/transport/objects.ts | 40 +-- lib/transport/publisher.ts | 41 ++- lib/transport/stream.ts | 41 ++- lib/transport/subgroup.ts | 392 +++++++++++++------------- lib/transport/subscriber.ts | 37 +-- lib/transport/utils.ts | 12 +- lib/video-moq/index.ts | 95 ++++--- 29 files changed, 1006 insertions(+), 448 deletions(-) create mode 100644 lib/common/logger.ts diff --git a/lib/common/index.ts b/lib/common/index.ts index 6d519d3..872d507 100644 --- a/lib/common/index.ts +++ b/lib/common/index.ts @@ -1 +1,12 @@ export { asError } from "./error" +export { + setGlobalLogger, + getGlobalLogger, + createConsoleLogger, + getLogger, + notifyLoggerLevelChanged, + onLoggerLevelChange, + installWorkerLogReceiver, + installWorkletLogReceiver, +} from "./logger" +export type { Logger, LogLevel, LogLevelName, ScopedLogger, WorkerLogRecord } from "./logger" diff --git a/lib/common/logger.ts b/lib/common/logger.ts new file mode 100644 index 0000000..65a2c4f --- /dev/null +++ b/lib/common/logger.ts @@ -0,0 +1,375 @@ +/** + * MoQJS Logger + * + * Design: + * - One global dispatcher lives in this module. + * - `getLogger()` is called once per module with no arguments. It reads the + * caller's file path from the call stack to derive a scope string + * (e.g. "transport/connection"), then returns a tiny frozen object whose + * five methods close over that scope string and call the single dispatcher. + * - Only one dispatcher function set exists in memory. Per-module objects are + * five arrow-function references + one string — negligible and unavoidable. + * + * Usage inside the library (top of every .ts file): + * import { getLogger } from "../common/logger" + * const log = getLogger() + * log.debug("starting control loop") + * + * Usage as a library consumer: + * import { setGlobalLogger, createConsoleLogger } from "@moq-js/player" + * setGlobalLogger(createConsoleLogger("debug")) + * setGlobalLogger({ level: () => "warn", warn: console.warn, error: console.error }) + * setGlobalLogger(undefined) // restore default (error-only) + * + * Every emit forwards: "[MoQJS]", "[]", ...args + */ + +// --------------------------------------------------------------------------- +// Public types +// --------------------------------------------------------------------------- + +/** + * Log level for controlling MoQJS output verbosity. + * + * `"*"` — enable everything (same as "trace") + * `"none"` — suppress all MoQJS output + */ +export type LogLevel = "*" | "trace" | "debug" | "info" | "warn" | "error" | "none" + +/** + * Logger interface implemented by consumers to receive MoQJS log messages. + * + * The library calls `level()` before every emit. If `level` is absent it + * defaults to `"error"` and emits one `console.warn` exactly once. + * + * All methods receive leading args: `"[MoQJS]"`, `"[]"`, ...original args. + * Individual level methods are optional — silently skipped when absent. + */ +export interface Logger { + level?(): LogLevel + trace?(...args: unknown[]): void + debug?(...args: unknown[]): void + info?(...args: unknown[]): void + warn?(...args: unknown[]): void + error?(...args: unknown[]): void +} + +/** The scoped logger returned to each module by `getLogger()`. */ +export interface ScopedLogger { + readonly scope: string + trace(...args: unknown[]): void + debug(...args: unknown[]): void + info(...args: unknown[]): void + warn(...args: unknown[]): void + error(...args: unknown[]): void +} + +/** Concrete level names that appear in log records (no meta-levels). */ +export type LogLevelName = "trace" | "debug" | "info" | "warn" | "error" + +/** A log record forwarded from a Web Worker or AudioWorklet to the main thread. */ +export interface WorkerLogRecord { + level: LogLevelName + args: unknown[] +} + +// --------------------------------------------------------------------------- +// Level ordering +// --------------------------------------------------------------------------- + +const LEVEL_ORDER: Record = { + "*": 0, + trace: 0, + debug: 1, + info: 2, + warn: 3, + error: 4, + none: 5, +} + +const METHOD_ORDER: Record = { + trace: 0, + debug: 1, + info: 2, + warn: 3, + error: 4, +} + +function _isEnabled(configured: LogLevel, emitting: LogLevelName): boolean { + return METHOD_ORDER[emitting] >= LEVEL_ORDER[configured] +} + +// --------------------------------------------------------------------------- +// Global logger state +// --------------------------------------------------------------------------- + +let _globalLogger: Logger = createConsoleLogger("error") +let _missingLevelWarned = false +let _levelListeners: Array<(level: LogLevel) => void> = [] + +/** Install a custom logger. Pass `undefined` to restore the default (error level). */ +export function setGlobalLogger(logger: Logger | undefined): void { + _missingLevelWarned = false + _globalLogger = logger ?? createConsoleLogger("error") + _fireLevelListeners() +} + +/** Returns the currently installed global logger. */ +export function getGlobalLogger(): Logger { + return _globalLogger +} + +/** + * Call this when your custom logger's `level()` return value changes at + * runtime. Notifies internal infrastructure (e.g. workers) to re-sync. + */ +export function notifyLoggerLevelChanged(): void { + _fireLevelListeners() +} + +/** @internal */ +export function onLoggerLevelChange(cb: (level: LogLevel) => void): () => void { + _levelListeners.push(cb) + return () => { + _levelListeners = _levelListeners.filter((l) => l !== cb) + } +} + +function _fireLevelListeners(): void { + const level = _resolveLevel() + for (const cb of _levelListeners) cb(level) +} + +function _resolveLevel(): LogLevel { + const logger = _globalLogger + if (typeof logger.level !== "function") { + if (!_missingLevelWarned) { + _missingLevelWarned = true + console.warn('[MoQJS] custom logger missing level(); defaulting to "error"') + } + return "error" + } + return logger.level() +} + +// --------------------------------------------------------------------------- +// Built-in console logger factory +// --------------------------------------------------------------------------- + +/** + * Creates a Logger backed by the browser/Node console. + * @example + * setGlobalLogger(createConsoleLogger("debug")) + * setGlobalLogger(createConsoleLogger("none")) // silence all + */ +export function createConsoleLogger(level: LogLevel = "error"): Logger { + return { + level: () => level, + trace: (...args: unknown[]) => { + console.debug(...args) + }, + debug: (...args: unknown[]) => { + console.debug(...args) + }, + info: (...args: unknown[]) => { + console.info(...args) + }, + warn: (...args: unknown[]) => { + console.warn(...args) + }, + error: (...args: unknown[]) => { + console.error(...args) + }, + } +} + +// --------------------------------------------------------------------------- +// Core dispatcher — the only place that touches _globalLogger per emit +// --------------------------------------------------------------------------- + +function _dispatch(level: LogLevelName, scope: string, args: unknown[]): void { + const logger = _globalLogger + if (!_isEnabled(_resolveLevel(), level)) return + const method = logger[level] + if (typeof method === "function") { + method.call(logger, "[MoQJS]", `[${scope}]`, ...args) + } +} + +// --------------------------------------------------------------------------- +// Scope derivation — called once per module at import time +// --------------------------------------------------------------------------- + +/** + * Derives a scope string from a file URL or path. + * "file:///…/lib/transport/connection.ts" → "transport/connection" + * "/abs/path/lib/playback/index.ts" → "playback" + * Strips "index" suffix so directory modules look clean. + */ +function _scopeFromUrl(url: string): string { + const path = url + .replace(/^file:\/\//, "") + .split("?")[0] + .split("#")[0] + + const libIdx = path.lastIndexOf("/lib/") + const relative = libIdx >= 0 ? path.slice(libIdx + 5) : path + + const noExt = relative.replace(/\.[tj]s$/, "") + const noIndex = noExt.replace(/\/index$/, "") + + return noIndex || relative +} + +/** + * Reads the call stack to find the URL of the caller's module. + * Skips frames belonging to this file. + */ +function _callerUrl(): string { + const stack = new Error().stack ?? "" + const lines = stack.split("\n") + for (const line of lines) { + const match = line.match(/\(?([^\s()]+\.[tj]s)[^)]*\)?$/) + if (!match) continue + const url = match[1] + if (url.includes("common/logger")) continue + return url + } + return "" +} + +// --------------------------------------------------------------------------- +// getLogger — the only public API modules need to call +// --------------------------------------------------------------------------- + +/** + * Returns a `ScopedLogger` for the calling module. Call once at module level: + * + * const log = getLogger() + * + * Scope is derived automatically from the caller's file path. + */ +export function getLogger(explicitScope?: string): ScopedLogger { + const scope = explicitScope ?? _scopeFromUrl(_callerUrl()) + + return Object.freeze({ + scope, + trace: (...args: unknown[]) => _dispatch("trace", scope, args), + debug: (...args: unknown[]) => _dispatch("debug", scope, args), + info: (...args: unknown[]) => _dispatch("info", scope, args), + warn: (...args: unknown[]) => _dispatch("warn", scope, args), + error: (...args: unknown[]) => _dispatch("error", scope, args), + }) +} + +// --------------------------------------------------------------------------- +// Worker-side logger +// --------------------------------------------------------------------------- + +let _workerLogLevel: LogLevel = "error" + +/** Called by the worker's message handler when a logLevel message arrives. */ +export function setWorkerLogLevel(level: LogLevel): void { + _workerLogLevel = level +} + +/** + * Returns a ScopedLogger for use inside a Web Worker. Scope is derived from + * the caller's file path automatically. Log records are forwarded to the main + * thread via `postMessage`. + * + * const log = getWorkerLogger() + */ +export function getWorkerLogger(explicitScope?: string): ScopedLogger { + const scope = explicitScope ?? _scopeFromUrl(_callerUrl()) + + function _workerDispatch(level: LogLevelName, args: unknown[]): void { + if (!_isEnabled(_workerLogLevel, level)) return + const record: WorkerLogRecord = { level, args: ["[MoQJS]", `[${scope}]`, ...args] } + ;(self as unknown as { postMessage(msg: unknown): void }).postMessage({ log: record }) + } + + return Object.freeze({ + scope, + trace: (...args: unknown[]) => _workerDispatch("trace", args), + debug: (...args: unknown[]) => _workerDispatch("debug", args), + info: (...args: unknown[]) => _workerDispatch("info", args), + warn: (...args: unknown[]) => _workerDispatch("warn", args), + error: (...args: unknown[]) => _workerDispatch("error", args), + }) +} + +// --------------------------------------------------------------------------- +// Worklet-side logger +// --------------------------------------------------------------------------- + +let _workletLogLevel: LogLevel = "error" + +/** Called by the worklet's message handler when a logLevel message arrives. */ +export function setWorkletLogLevel(level: LogLevel): void { + _workletLogLevel = level +} + +/** + * Returns a ScopedLogger for use inside an AudioWorklet processor. Scope is + * derived automatically. Log records are forwarded through the worklet port. + * + * const log = getWorkletLogger(this.port) + */ +export function getWorkletLogger(port: MessagePort, explicitScope?: string): ScopedLogger { + const scope = explicitScope ?? _scopeFromUrl(_callerUrl()) + + function _workletDispatch(level: LogLevelName, args: unknown[]): void { + if (!_isEnabled(_workletLogLevel, level)) return + const record: WorkerLogRecord = { level, args: ["[MoQJS]", `[${scope}]`, ...args] } + port.postMessage({ log: record }) + } + + return Object.freeze({ + scope, + trace: (...args: unknown[]) => _workletDispatch("trace", args), + debug: (...args: unknown[]) => _workletDispatch("debug", args), + info: (...args: unknown[]) => _workletDispatch("info", args), + warn: (...args: unknown[]) => _workletDispatch("warn", args), + error: (...args: unknown[]) => _workletDispatch("error", args), + }) +} + +// --------------------------------------------------------------------------- +// Main-thread receivers +// --------------------------------------------------------------------------- + +/** + * Listens for `{ log: WorkerLogRecord }` messages from a Worker and routes + * them into the global logger. Returns a dispose function. + * @internal + */ +export function installWorkerLogReceiver(worker: Worker): () => void { + const handler = (e: MessageEvent) => { + const msg = e.data as { log?: WorkerLogRecord } + if (msg?.log) _dispatchRecord(msg.log) + } + worker.addEventListener("message", handler) + return () => worker.removeEventListener("message", handler) +} + +/** + * Listens for `{ log: WorkerLogRecord }` messages from an AudioWorklet port + * and routes them into the global logger. Returns a dispose function. + * @internal + */ +export function installWorkletLogReceiver(port: MessagePort): () => void { + const handler = (e: MessageEvent) => { + const msg = e.data as { log?: WorkerLogRecord } + if (msg?.log) _dispatchRecord(msg.log) + } + port.addEventListener("message", handler) + return () => port.removeEventListener("message", handler) +} + +function _dispatchRecord({ level, args }: WorkerLogRecord): void { + const logger = _globalLogger + if (!_isEnabled(_resolveLevel(), level)) return + const method = logger[level] + if (typeof method === "function") method.call(logger, ...args) +} diff --git a/lib/contribute/audio.ts b/lib/contribute/audio.ts index 11bfe43..9333790 100644 --- a/lib/contribute/audio.ts +++ b/lib/contribute/audio.ts @@ -1,3 +1,7 @@ +import { getLogger } from "../common/logger" + +const log = getLogger() + const SUPPORTED = [ // TODO support AAC // "mp4a" @@ -34,7 +38,7 @@ export class Encoder { this.#encoder.configure(this.#encoderConfig) } catch (e) { - console.error("Failed to configure AudioEncoder:", e) + log.error("failed to configure AudioEncoder", e) throw e } } @@ -43,7 +47,7 @@ export class Encoder { try { this.#encoder.encode(frame) } catch (e) { - console.error("Failed to encode audio frame:", e) + log.error("failed to encode audio frame", e) throw e } frame.close() diff --git a/lib/contribute/broadcast.ts b/lib/contribute/broadcast.ts index 43f140c..7a81bb2 100644 --- a/lib/contribute/broadcast.ts +++ b/lib/contribute/broadcast.ts @@ -6,6 +6,9 @@ import * as Catalog from "../media/catalog" import { isAudioTrackSettings, isVideoTrackSettings } from "../common/settings" import { sleep } from "../transport/utils" +import { getLogger } from "../common/logger" + +const log = getLogger() export interface BroadcastConfig { namespace: string[] @@ -46,16 +49,16 @@ export class Broadcast { const settings = media.getSettings() if (media.kind === "audio") { - const audioContext = new AudioContext(); + const audioContext = new AudioContext() audioContext.createMediaStreamSource(new MediaStream([media])) const sampleRate = audioContext.sampleRate Object.assign(settings, { sampleRate, }) - audioContext.close() + void audioContext.close() } - console.log("track settings", settings, media, mediaTracks) + log.debug("track settings", settings, media, mediaTracks) if (isVideoTrackSettings(settings)) { if (!config.video) { @@ -118,17 +121,17 @@ export class Broadcast { } async #run() { - console.log("[Broadcast] #run loop started") + log.debug("run loop started") await this.connection.publish_namespace(this.namespace) - for (; ;) { + for (;;) { const subscriber = await this.connection.subscribed() if (!subscriber) break // Run an async task to serve each subscription. this.#serveSubscribe(subscriber).catch((e) => { const err = asError(e) - console.warn("failed to serve subscribe", err) + log.warn("failed to serve subscribe", err) }) } } @@ -136,7 +139,7 @@ export class Broadcast { async #serveSubscribe(subscriber: SubscribeRecv) { try { const [base, ext] = splitExt(subscriber.track) - console.log("serving subscribe", subscriber.track, subscriber.namespace, base, ext) + log.debug("serving subscribe", subscriber.track, subscriber.namespace, base, ext) if (ext === "catalog") { await this.#serveCatalog(subscriber, base) } else if (ext === "mp4") { @@ -147,7 +150,7 @@ export class Broadcast { throw new Error(`unknown subscription: ${subscriber.track}`) } } catch (e) { - console.error("failed to serve subscribe", e) + log.error("failed to serve subscribe", e) const err = asError(e) // TODO(itzmanish): should check if the error is not found and send appropriate error code await subscriber.close({ code: 0n, reason: `failed to process subscribe: ${err.message}` }) @@ -164,7 +167,7 @@ export class Broadcast { const bytes = Catalog.encode(this.catalog) await subscriber.ack() - await sleep(500); + await sleep(500) const stream = await subscriber.subgroup({ group: 0, subgroup: 0 }) await stream.write({ object_id: 0, object_payload: bytes }) @@ -176,7 +179,7 @@ export class Broadcast { if (!track) throw new Error(`no track with name ${subscriber.track}`) await subscriber.ack() - await sleep(500); + await sleep(500) const init = await track.init() @@ -193,18 +196,18 @@ export class Broadcast { await subscriber.ack() // NOTE(itzmanish): hack to make sure subscribe ok reaches before the segement object - await sleep(500); + await sleep(500) const segments = track.segments().getReader() - for (; ;) { + for (;;) { const { value: segment, done } = await segments.read() if (done) break // Serve the segment and log any errors that occur. this.#serveSegment(subscriber, segment).catch((e) => { const err = asError(e) - console.warn("failed to serve segment", err) + log.warn("failed to serve segment", err) }) } } @@ -221,7 +224,7 @@ export class Broadcast { // Pipe the segment to the stream. const chunks = segment.chunks().getReader() - for (; ;) { + for (;;) { const { value, done } = await chunks.read() if (done) break diff --git a/lib/contribute/container.ts b/lib/contribute/container.ts index b82d4aa..bd419b5 100644 --- a/lib/contribute/container.ts +++ b/lib/contribute/container.ts @@ -1,5 +1,8 @@ import * as MP4 from "../media/mp4" import { Chunk } from "./chunk" +import { getLogger } from "../common/logger" + +const log = getLogger() type DecoderConfig = AudioDecoderConfig | VideoDecoderConfig type EncodedChunk = EncodedAudioChunk | EncodedVideoChunk @@ -20,13 +23,13 @@ export class Container { transform: (frame, controller) => { try { if (isDecoderConfig(frame)) { - console.log("Container received decoder config:", frame) + log.debug("received decoder config", frame) return this.#init(frame, controller) } else { return this.#enqueue(frame, controller) } } catch (e) { - console.error("Container failed to process frame:", e) + log.error("failed to process frame", e) throw e } }, @@ -125,7 +128,7 @@ export class Container { // Moof and mdat atoms are written in pairs. // TODO remove the moof/mdat from the Box to reclaim memory once everything works - for (; ;) { + for (;;) { const moof = this.#mp4.moofs.shift() const mdat = this.#mp4.mdats.shift() diff --git a/lib/contribute/track.ts b/lib/contribute/track.ts index e184ea7..4a65ac5 100644 --- a/lib/contribute/track.ts +++ b/lib/contribute/track.ts @@ -6,6 +6,9 @@ import { BroadcastConfig } from "./broadcast" import * as Audio from "./audio" import * as Video from "./video" +import { getLogger } from "../common/logger" + +const log = getLogger() export class Track { name: string @@ -50,7 +53,7 @@ export class Track { .pipeThrough(container.encode) .pipeTo(segments) .catch((err) => { - console.error("Audio pipeline error:", err) + log.error("audio pipeline error", err) throw err }) } @@ -108,7 +111,7 @@ export class Track { if ((writer.desiredSize || 0) > 0) { await writer.write(chunk) } else { - console.warn("dropping chunk", writer.desiredSize) + log.warn("dropping chunk", writer.desiredSize) } writer.releaseLock() @@ -141,7 +144,7 @@ export class Track { return new ReadableStream({ pull: async (controller) => { - for (; ;) { + for (;;) { let index = pos - this.#offset if (index < 0) index = 0 diff --git a/lib/contribute/video.ts b/lib/contribute/video.ts index 3e49fab..5844dbf 100644 --- a/lib/contribute/video.ts +++ b/lib/contribute/video.ts @@ -1,3 +1,7 @@ +import { getLogger } from "../common/logger" + +const log = getLogger() + const SUPPORTED = [ "avc1", // H.264 "hev1", // HEVC (aka h.265) @@ -63,7 +67,7 @@ export class Encoder { }, }) - console.log("Encoder configured", this.#encoderConfig) + log.debug("encoder configured", this.#encoderConfig) this.#encoder.configure(this.#encoderConfig) } diff --git a/lib/media/catalog/index.ts b/lib/media/catalog/index.ts index 5839a35..4b7950d 100644 --- a/lib/media/catalog/index.ts +++ b/lib/media/catalog/index.ts @@ -1,6 +1,9 @@ import { Connection } from "../../transport" import { asError } from "../../common/error" import { debug } from "../../transport/utils" +import { getLogger } from "../../common/logger" + +const log = getLogger() export interface CommonTrackFields { namespace?: string[] @@ -54,11 +57,11 @@ export async function fetch(connection: Connection, namespace: string[]) { const segment = await subscribe.data() if (!segment) throw new Error("no catalog data") - console.log("catalog segment", segment) + log.debug("catalog segment", segment) const chunk = await segment.read() if (!chunk) throw new Error("no catalog chunk") - console.log("catalog chunk", chunk) + log.debug("catalog chunk", chunk) await segment.close() await subscribe.close() // we done @@ -68,7 +71,7 @@ export async function fetch(connection: Connection, namespace: string[]) { throw new Error("invalid catalog chunk") } } catch (e) { - console.error("Catalog fetch error: ", e) + log.error("catalog fetch error", e) const err = asError(e) // // Close the subscription after we're done. diff --git a/lib/media/mp4/parser.ts b/lib/media/mp4/parser.ts index 7cc78d0..f6e4157 100644 --- a/lib/media/mp4/parser.ts +++ b/lib/media/mp4/parser.ts @@ -1,4 +1,7 @@ import * as MP4 from "./index" +import { getLogger } from "../../common/logger" + +const log = getLogger() export interface Frame { track: MP4.Track // The track this frame belongs to @@ -16,7 +19,7 @@ export class Parser { constructor(init: Uint8Array) { this.#mp4.onError = (err) => { - console.error("MP4 error", err) + log.error("MP4 error", err) } this.#mp4.onReady = (info: MP4.Info) => { diff --git a/lib/moq-publisher/index.ts b/lib/moq-publisher/index.ts index f976f1d..f64bb04 100644 --- a/lib/moq-publisher/index.ts +++ b/lib/moq-publisher/index.ts @@ -1,5 +1,8 @@ import STYLE_SHEET from "./publisher-moq.css" import { PublisherApi, PublisherOptions } from "../publish" +import { getLogger } from "../common/logger" + +const log = getLogger() export class PublisherMoq extends HTMLElement { private shadow: ShadowRoot @@ -14,6 +17,15 @@ export class PublisherMoq extends HTMLElement { private isPublishing = false private namespace = "" + // Arrow properties so they can be passed directly to addEventListener/removeEventListener + // without losing `this` context and without triggering no-misused-promises. + private readonly onDeviceChange = () => { + this.populateDeviceLists().catch((e) => log.error("populateDeviceLists failed", e)) + } + private readonly onConnectClick = () => { + this.handleClick().catch((e) => log.error("handleClick failed", e)) + } + constructor() { super() this.shadow = this.attachShadow({ mode: "open" }) @@ -52,27 +64,23 @@ export class PublisherMoq extends HTMLElement { ) this.shadow.appendChild(container) - // Bindings - this.handleDeviceChange = this.handleDeviceChange.bind(this) - this.handleClick = this.handleClick.bind(this) - // Listeners - navigator.mediaDevices.addEventListener("devicechange", this.handleDeviceChange) - this.cameraSelect.addEventListener("change", () => this.startPreview()) - this.microphoneSelect.addEventListener("change", () => this.startPreview()) - this.connectButton.addEventListener("click", this.handleClick) + navigator.mediaDevices.addEventListener("devicechange", this.onDeviceChange) + this.cameraSelect.addEventListener("change", () => { + this.startPreview().catch((e) => log.error("startPreview failed", e)) + }) + this.microphoneSelect.addEventListener("change", () => { + this.startPreview().catch((e) => log.error("startPreview failed", e)) + }) + this.connectButton.addEventListener("click", this.onConnectClick) } connectedCallback() { - this.populateDeviceLists() + this.populateDeviceLists().catch((e) => log.error("populateDeviceLists failed", e)) } disconnectedCallback() { - navigator.mediaDevices.removeEventListener("devicechange", this.handleDeviceChange) - } - - private async handleDeviceChange() { - await this.populateDeviceLists() + navigator.mediaDevices.removeEventListener("devicechange", this.onDeviceChange) } private async populateDeviceLists() { @@ -121,7 +129,7 @@ export class PublisherMoq extends HTMLElement { private async handleClick() { if (!this.isPublishing) { if (!this.mediaStream) { - console.warn("No media stream available") + log.warn("No media stream available") return } @@ -130,7 +138,7 @@ export class PublisherMoq extends HTMLElement { const audioTrack = this.mediaStream.getAudioTracks()[0] const settings = audioTrack.getSettings() - const sampleRate = settings.sampleRate ?? (await new AudioContext()).sampleRate + const sampleRate = settings.sampleRate ?? new AudioContext().sampleRate const numberOfChannels = settings.channelCount ?? 2 // H.264 requires even dimensions - round down to nearest even number @@ -156,7 +164,7 @@ export class PublisherMoq extends HTMLElement { audio: audioConfig, } - console.log("Publisher Options", opts) + log.debug("Publisher Options", opts) this.publisher = new PublisherApi(opts) @@ -175,13 +183,13 @@ export class PublisherMoq extends HTMLElement { } this.playbackUrlTextarea.style.display = "block" } catch (err) { - console.error("Publish failed:", err) + log.error("Publish failed:", err) } } else { try { await this.publisher!.stop() } catch (err) { - console.error("Stop failed:", err) + log.error("Stop failed:", err) } finally { this.isPublishing = false this.connectButton.textContent = "Connect" @@ -194,4 +202,8 @@ export class PublisherMoq extends HTMLElement { } customElements.define("publisher-moq", PublisherMoq) -export default PublisherMoq \ No newline at end of file +export default PublisherMoq + +// Re-export logger API so consumers of @moq-js/player/moq-publisher can configure logging. +export { setGlobalLogger, getGlobalLogger, createConsoleLogger, notifyLoggerLevelChanged } from "../common/logger" +export type { Logger, LogLevel } from "../common/logger" diff --git a/lib/playback/audio.ts b/lib/playback/audio.ts index 85817b1..ccf1aaf 100644 --- a/lib/playback/audio.ts +++ b/lib/playback/audio.ts @@ -1,6 +1,9 @@ import * as Message from "./worker/message" import registerMyAudioWorklet from "audio-worklet:./worklet/index.ts" +import { getLogger, getGlobalLogger, installWorkletLogReceiver, onLoggerLevelChange } from "../common/logger" + +const log = getLogger() // NOTE: This must be on the main thread export class Audio { @@ -8,6 +11,9 @@ export class Audio { worklet: Promise volumeNode: GainNode + // Dispose function for the logger level change listener + #disposeLoggerListener?: () => void + constructor(config: Message.ConfigAudio) { this.context = new AudioContext({ latencyHint: "interactive", @@ -28,17 +34,30 @@ export class Audio { // Create the worklet const worklet = new AudioWorkletNode(this.context, "renderer") - worklet.port.addEventListener("message", this.on.bind(this)) worklet.onprocessorerror = (e: Event) => { - console.error("Audio worklet error:", e) + log.error("audio worklet processor error", e) } + // Install log receiver so worklet log records appear in the global logger. + worklet.port.start() + installWorkletLogReceiver(worklet.port) + + // Keep the worklet's cached log level in sync whenever the global logger changes. + this.#disposeLoggerListener = onLoggerLevelChange((level) => { + worklet.port.postMessage({ logLevel: level }) + }) + // Connect the worklet to the volume node and then to the speakers worklet.connect(this.volumeNode) this.volumeNode.connect(this.context.destination) worklet.port.postMessage({ config }) + // Send the initial log level to the worklet. + const logger = getGlobalLogger() + const initialLevel = typeof logger.level === "function" ? logger.level() : "error" + worklet.port.postMessage({ logLevel: initialLevel }) + return worklet } @@ -53,4 +72,13 @@ export class Audio { public getVolume(): number { return this.volumeNode.gain.value } + + async close() { + this.#disposeLoggerListener?.() + const worklet = await this.worklet.catch(() => undefined) + if (worklet) { + worklet.port.close() + worklet.disconnect() + } + } } diff --git a/lib/playback/backend.ts b/lib/playback/backend.ts index 148ee8b..1a5b554 100644 --- a/lib/playback/backend.ts +++ b/lib/playback/backend.ts @@ -7,6 +7,7 @@ import MediaWorker from "web-worker:./worker/index.ts" import { RingShared } from "../common/ring" import { Root, isAudioTrack } from "../media/catalog" import { SubgroupHeader } from "../transport/subgroup" +import { getGlobalLogger, installWorkerLogReceiver, onLoggerLevelChange } from "../common/logger" export interface PlayerConfig { canvas: OffscreenCanvas @@ -26,12 +27,23 @@ export default class Backend { #eventTarget: EventTarget + // Dispose function for the logger level change listener + #disposeLoggerListener: () => void + constructor(config: PlayerConfig, eventTarget: EventTarget) { // TODO does this block the main thread? If so, make this async this.#worker = new MediaWorker() - this.#worker.addEventListener("message", this.on.bind(this)) + this.#worker.addEventListener("message", this.#on.bind(this)) this.#eventTarget = eventTarget + // Install log receiver so worker log records appear in the global logger. + installWorkerLogReceiver(this.#worker) + + // Keep the worker's cached log level in sync whenever the global logger changes. + this.#disposeLoggerListener = onLoggerLevelChange((level) => { + this.send({ logLevel: level }) + }) + let sampleRate: number | undefined let channels: number | undefined @@ -67,6 +79,11 @@ export default class Backend { } this.send({ config: msg }, msg.video.canvas) + + // Send the initial log level to the worker. + const logger = getGlobalLogger() + const initialLevel = typeof logger.level === "function" ? logger.level() : "error" + this.send({ logLevel: initialLevel }) } pause() { @@ -102,19 +119,21 @@ export default class Backend { } async close() { + this.#disposeLoggerListener() this.#worker.terminate() await this.#audio?.context.close() } // Enforce we're sending valid types to the worker private send(msg: Message.ToWorker, ...transfer: Transferable[]) { - //console.log("sent message from main to worker", msg) this.#worker.postMessage(msg, transfer) } - private on(e: MessageEvent) { - const msg = e.data - if (msg === "waitingforkeyframe") { + #on(e: MessageEvent) { + // log records are already handled by installWorkerLogReceiver listener. + if ((e.data as { log?: unknown })?.log) return + // The video worker posts the raw string "waitingforkeyframe". + if (e.data === "waitingforkeyframe") { this.#eventTarget.dispatchEvent(new Event("waitingforkeyframe")) } } diff --git a/lib/playback/index.ts b/lib/playback/index.ts index d468b96..7712132 100644 --- a/lib/playback/index.ts +++ b/lib/playback/index.ts @@ -8,6 +8,13 @@ import Backend from "./backend" import { Client } from "../transport/client" import { SubgroupReader } from "../transport/subgroup" +import { getLogger } from "../common/logger" + +// Re-export logger API so consumers of @moq-js/player/simple-player can configure logging. +export { setGlobalLogger, getGlobalLogger, createConsoleLogger, notifyLoggerLevelChanged } from "../common/logger" +export type { Logger, LogLevel } from "../common/logger" + +const log = getLogger() export type Range = Message.Range export type Timeline = Message.Timeline @@ -66,7 +73,7 @@ export default class Player extends EventTarget { this.#running = abort.catch(this.#close) this.#run().catch((err) => { - console.error("Error in #run():", err) + log.error("error in run", err) super.dispatchEvent(new CustomEvent("error", { detail: err })) this.#abort(err) }) @@ -77,7 +84,7 @@ export default class Player extends EventTarget { const connection = await client.connect() const catalog = await Catalog.fetch(connection, [config.namespace]) - console.log("catalog", catalog) + log.debug("catalog", catalog) const canvas = config.canvas.transferControlToOffscreen() @@ -98,8 +105,8 @@ export default class Player extends EventTarget { } }) - console.log("inits", inits) - console.log("tracks", tracks) + log.debug("inits", inits) + log.debug("tracks", tracks) // Call #runInit on each unique init track // TODO do this in parallel with #runTrack to remove a round trip @@ -113,14 +120,14 @@ export default class Player extends EventTarget { } async #runInit(namespace: string, name: string) { - console.log("running #runInit", namespace, name) + log.debug("running runInit", namespace, name) const sub = await this.#connection.subscribe([namespace], name) try { - console.log("waiting for init data") + log.debug("waiting for init data") const init = await Promise.race([sub.data(), this.#running]) if (!init) throw new Error("no init data") - console.log("got init data") + log.debug("got init data") // We don't care what type of reader we get, we just want the payload. const chunk = await init.read() if (!chunk) throw new Error("no init chunk") @@ -153,9 +160,9 @@ export default class Player extends EventTarget { const sub = await this.#connection.subscribe(track.namespace, track.name) try { - console.log("starting segment data loop") - for (; ;) { - console.log("waiting for segment data") + log.debug("starting segment data loop") + for (;;) { + log.trace("waiting for segment data") const segment = await Promise.race([sub.data(), this.#running]) if (!segment) continue @@ -188,9 +195,9 @@ export default class Player extends EventTarget { } } catch (error) { if (error instanceof Error && error.message.includes("cancelled")) { - console.log("Cancelled subscription to track: ", track.name) + log.debug("cancelled subscription to track", track.name) } else { - console.error("Error in #runTrack:", error) + log.error("error in runTrack", error) super.dispatchEvent(new CustomEvent("error", { detail: error })) } } finally { @@ -200,7 +207,7 @@ export default class Player extends EventTarget { #runTrack(track: Catalog.Track) { if (this.#trackTasks.has(track.name)) { - console.warn(`Already exist a runTrack task for the track: ${track.name}`) + log.warn(`runTrack task already exists for track: ${track.name}`) return } @@ -209,7 +216,7 @@ export default class Player extends EventTarget { this.#trackTasks.set(track.name, task) task.catch((err) => { - console.error(`Error to subscribe to track ${track.name}`, err) + log.error(`error subscribing to track ${track.name}`, err) super.dispatchEvent(new CustomEvent("error", { detail: err })) }).finally(() => { this.#trackTasks.delete(track.name) @@ -230,7 +237,7 @@ export default class Player extends EventTarget { if (this.#tracknum >= 0 && this.#tracknum < this.#catalog.tracks.length) { return this.#catalog.tracks[this.#tracknum] } else { - console.warn("Invalid track number:", this.#tracknum) + log.warn("invalid track number", this.#tracknum) return null } } @@ -266,10 +273,10 @@ export default class Player extends EventTarget { return } if (currentTrack) { - console.log(`Unsubscribing from track: ${currentTrack.name} and Subscribing to track: ${trackname}`) + log.debug(`unsubscribing from ${currentTrack.name}, subscribing to ${trackname}`) await this.unsubscribeFromTrack(currentTrack.name) } else { - console.log(`Subscribing to track: ${trackname}`) + log.debug(`subscribing to track ${trackname}`) } this.#tracknum = this.#catalog.tracks.findIndex((track) => track.name === trackname) @@ -279,11 +286,11 @@ export default class Player extends EventTarget { async mute(isMuted: boolean) { this.#muted = isMuted if (isMuted) { - console.log("Unsubscribing from audio track: ", this.#audioTrackName) + log.debug("unsubscribing from audio track", this.#audioTrackName) await this.unsubscribeFromTrack(this.#audioTrackName) await this.#backend.mute() } else { - console.log("Subscribing to audio track: ", this.#audioTrackName) + log.debug("subscribing to audio track", this.#audioTrackName) this.subscribeFromTrackName(this.#audioTrackName) await this.#backend.unmute() } @@ -291,7 +298,7 @@ export default class Player extends EventTarget { } async unsubscribeFromTrack(trackname: string) { - console.log(`Unsubscribing from track: ${trackname}`) + log.debug(`unsubscribing from track ${trackname}`) super.dispatchEvent(new CustomEvent("unsubscribestared", { detail: { track: trackname } })) await this.#connection.unsubscribe(trackname) const task = this.#trackTasks.get(trackname) @@ -302,14 +309,14 @@ export default class Player extends EventTarget { } subscribeFromTrackName(trackname: string) { - console.log(`Subscribing to track: ${trackname}`) + log.debug(`subscribing to track ${trackname}`) const track = this.#tracksByName.get(trackname) if (track) { super.dispatchEvent(new CustomEvent("subscribestared", { detail: { track: trackname } })) this.#runTrack(track) super.dispatchEvent(new CustomEvent("subscribedone", { detail: { track: trackname } })) } else { - console.warn(`Track ${trackname} not in #tracksByName`) + log.warn(`track ${trackname} not found`) } } @@ -331,7 +338,7 @@ export default class Player extends EventTarget { try { await this.#running } catch (e) { - console.error("Error in Player.closed():", e) + log.error("error in Player.closed", e) return asError(e) } } @@ -375,7 +382,7 @@ export default class Player extends EventTarget { const audioPromise = this.unsubscribeFromTrack(this.#audioTrackName) const videoPromise = this.unsubscribeFromTrack(this.#videoTrackName) super.dispatchEvent(new CustomEvent("pause", { detail: { track: this.#videoTrackName } })) - console.log("dispatchEvent pause") + log.debug("dispatching pause event") this.#backend.pause() await Promise.all([mutePromise, audioPromise, videoPromise]) diff --git a/lib/playback/worker/audio.ts b/lib/playback/worker/audio.ts index de57e36..6c089d8 100644 --- a/lib/playback/worker/audio.ts +++ b/lib/playback/worker/audio.ts @@ -2,6 +2,9 @@ import * as Message from "./message" import { Ring } from "../../common/ring" import { Component, Frame } from "./timeline" import * as MP4 from "../../media/mp4" +import { getWorkerLogger } from "../../common/logger" + +const log = getWorkerLogger() // This is run in a worker. export class Renderer { @@ -20,7 +23,7 @@ export class Renderer { transform: this.#transform.bind(this), }) - this.#run().catch(console.error) + this.#run().catch((e) => log.error("run failed", e)) } #start(controller: TransformStreamDefaultController) { @@ -28,7 +31,7 @@ export class Renderer { output: (frame: AudioData) => { controller.enqueue(frame) }, - error: console.warn, + error: (e) => log.warn("audio decoder error", e), }) } @@ -66,7 +69,7 @@ export class Renderer { const written = this.#ring.write(frame) if (written < frame.numberOfFrames) { - console.warn(`droppped ${frame.numberOfFrames - written} audio samples`) + log.warn(`dropped ${frame.numberOfFrames - written} audio samples`) } } } diff --git a/lib/playback/worker/index.ts b/lib/playback/worker/index.ts index 4a9d556..d8aff6b 100644 --- a/lib/playback/worker/index.ts +++ b/lib/playback/worker/index.ts @@ -9,6 +9,9 @@ import { asError } from "../../common/error" import { Deferred } from "../../common/async" import { SubgroupReader } from "../../transport/subgroup" import { ReadableStreamBuffer } from "../../transport/buffer" +import { getWorkerLogger, setWorkerLogLevel } from "../../common/logger" + +const log = getWorkerLogger() class Worker { // Timeline receives samples, buffering them and choosing the timestamp to render. @@ -25,13 +28,16 @@ class Worker { const msg = e.data as Message.ToWorker // console.log("message: ", msg) - if (msg.config) { + if (msg.logLevel !== undefined) { + // Update the module-level worker log level so all worker loggers re-read it. + setWorkerLogLevel(msg.logLevel) + } else if (msg.config) { this.#onConfig(msg.config) } else if (msg.init) { - // TODO buffer the init segmnet so we don't hold the stream open. + // TODO buffer the init segment so we don't hold the stream open. this.#onInit(msg.init) } else if (msg.segment) { - this.#onSegment(msg.segment).catch(console.warn) + this.#onSegment(msg.segment).catch((e) => log.warn("onSegment failed", e)) } else if (msg.play === false) { this.#onPause(msg.play) } else if (msg.play === true) { @@ -87,7 +93,7 @@ class Worker { segments.releaseLock() // Read each chunk, decoding the MP4 frames and adding them to the queue. - for (; ;) { + for (;;) { const chunk = await reader.read() if (!chunk) { break @@ -127,7 +133,7 @@ self.addEventListener("message", (msg) => { worker.on(msg) } catch (e) { const err = asError(e) - console.warn("worker error:", err) + log.warn("worker error:", err) } }) diff --git a/lib/playback/worker/message.ts b/lib/playback/worker/message.ts index c0c1868..71d3fa3 100644 --- a/lib/playback/worker/message.ts +++ b/lib/playback/worker/message.ts @@ -1,5 +1,6 @@ import { SubgroupHeader } from "../../transport/subgroup" import { RingShared } from "../../common/ring" +import type { LogLevel, WorkerLogRecord } from "../../common/logger" export interface Config { audio?: ConfigAudio @@ -78,6 +79,9 @@ export interface ToWorker { segment?: Segment play?: boolean + // Sent to update the worker's cached log level when setGlobalLogger is called. + logLevel?: LogLevel + /* // Sent to control playback play?: Play @@ -89,6 +93,9 @@ export interface ToWorker { export interface FromWorker { // Sent back to the main thread regularly to update the UI timeline?: Timeline + + // Log record forwarded from the worker to the main thread. + log?: WorkerLogRecord } /* diff --git a/lib/playback/worker/video.ts b/lib/playback/worker/video.ts index a8691d6..d35f895 100644 --- a/lib/playback/worker/video.ts +++ b/lib/playback/worker/video.ts @@ -1,6 +1,9 @@ import { Frame, Component } from "./timeline" import * as MP4 from "../../media/mp4" import * as Message from "./message" +import { getWorkerLogger } from "../../common/logger" + +const log = getWorkerLogger() interface DecoderConfig { codec: string @@ -40,13 +43,13 @@ export class Renderer { transform: this.#transform.bind(this), }) - this.#run().catch(console.error) + this.#run().catch((e) => log.error("run failed", e)) } pause() { this.#paused = true this.#decoder.flush().catch((err) => { - console.error(err) + log.error("flush failed on pause", err) }) this.#waitingForKeyframe = true } @@ -57,7 +60,7 @@ export class Renderer { async #run() { const reader = this.#timeline.frames.pipeThrough(this.#queue).getReader() - for (; ;) { + for (;;) { const { value: frame, done } = await reader.read() if (this.#paused) continue if (done) break @@ -80,13 +83,13 @@ export class Renderer { output: (frame: VideoFrame) => { controller.enqueue(frame) }, - error: console.error, + error: (e) => log.error("video decoder error", e), }) } #transform(frame: Frame) { if (this.#decoder.state === "closed" || this.#paused) { - console.warn("Decoder is closed or paused. Skipping frame.") + log.warn("decoder is closed or paused, skipping frame") return } @@ -129,9 +132,9 @@ export class Renderer { try { this.#decoder.configure(this.#decoderConfig) - console.log(`[VideoWorker] Decoder configured successfully. New state: ${this.#decoder.state}`) + log.debug("decoder configured", { codec: track.codec, state: this.#decoder.state }) } catch (e) { - console.error("[VideoWorker] FAILED to configure decoder:", e) + log.error("failed to configure decoder", e) return // Stop processing if configure fails } if (!frame.sample.is_sync) { @@ -141,10 +144,10 @@ export class Renderer { } } - //At the start of decode , VideoDecoder seems to expect a key frame after configure() or flush() + //At the start of decode, VideoDecoder seems to expect a key frame after configure() or flush() if (this.#decoder.state == "configured") { if (this.#waitingForKeyframe && !frame.sample.is_sync) { - console.warn("Skipping non-keyframe until a keyframe is found.") + log.warn("skipping non-keyframe until a keyframe is found") if (!this.#hasSentWaitingForKeyFrameEvent) { self.postMessage("waitingforkeyframe") this.#hasSentWaitingForKeyFrameEvent = true @@ -164,11 +167,11 @@ export class Renderer { timestamp: frame.sample.dts / frame.track.timescale, }) - console.log(`[VideoWorker] Decoding chunk, type: ${chunk.type}, size: ${chunk.byteLength}`) + log.trace("decoding chunk", { type: chunk.type, size: chunk.byteLength }) try { this.#decoder.decode(chunk) } catch (e) { - console.error("[VideoWorker] FAILED to decode chunk:", e) + log.error("failed to decode chunk", e) } } } diff --git a/lib/playback/worklet/index.ts b/lib/playback/worklet/index.ts index afe485d..1de9580 100644 --- a/lib/playback/worklet/index.ts +++ b/lib/playback/worklet/index.ts @@ -1,22 +1,27 @@ // TODO add support for @/ to avoid relative imports import { Ring } from "../../common/ring" import * as Message from "./message" +import { getWorkletLogger, setWorkletLogLevel } from "../../common/logger" class Renderer extends AudioWorkletProcessor { ring?: Ring base: number + #log!: ReturnType constructor() { // The super constructor call is required. super() this.base = 0 + this.#log = getWorkletLogger(this.port) this.port.onmessage = this.onMessage.bind(this) } onMessage(e: MessageEvent) { const msg = e.data as Message.From - if (msg.config) { + if (msg.logLevel !== undefined) { + setWorkletLogLevel(msg.logLevel) + } else if (msg.config) { this.onConfig(msg.config) } } @@ -39,7 +44,7 @@ class Renderer extends AudioWorkletProcessor { if (this.ring.size() == this.ring.capacity) { // This is a hack to clear any latency in the ring buffer. // The proper solution is to play back slightly faster? - console.warn("resyncing ring buffer") + this.#log.warn("resyncing ring buffer") this.ring.clear() return true } diff --git a/lib/playback/worklet/message.ts b/lib/playback/worklet/message.ts index e55ebdd..ca20382 100644 --- a/lib/playback/worklet/message.ts +++ b/lib/playback/worklet/message.ts @@ -1,7 +1,10 @@ import { RingShared } from "../../common/ring" +import type { LogLevel, WorkerLogRecord } from "../../common/logger" export interface From { config?: Config + // Sent to update the worklet's cached log level. + logLevel?: LogLevel } export interface Config { @@ -10,3 +13,8 @@ export interface Config { ring: RingShared } + +// Log record forwarded from the worklet to the main thread via port.postMessage. +export interface WorkletLogMessage { + log: WorkerLogRecord +} diff --git a/lib/transport/client.ts b/lib/transport/client.ts index b377841..19c50f7 100644 --- a/lib/transport/client.ts +++ b/lib/transport/client.ts @@ -1,9 +1,12 @@ import * as Control from "./control" -import * as Stream from './stream' +import * as Stream from "./stream" import { Objects } from "./objects" import { Connection } from "./connection" import { ClientSetup, ControlMessageType, ServerSetup } from "./control" import { ImmutableBytesBuffer, ReadableWritableStreamBuffer } from "./buffer" +import { getLogger } from "../common/logger" + +const log = getLogger() export interface ClientConfig { url: string @@ -21,7 +24,7 @@ export class Client { this.config = config this.#fingerprint = this.#fetchFingerprint(config.fingerprint).catch((e) => { - console.warn("failed to fetch fingerprint: ", e) + log.warn("failed to fetch fingerprint", e) return undefined }) } @@ -80,7 +83,8 @@ export class Client { async readServerSetup(buffer: ReadableWritableStreamBuffer): Promise { const type: ControlMessageType = await buffer.getNumberVarInt() - if (type !== ControlMessageType.ServerSetup) throw new Error(`server SETUP type must be ${ControlMessageType.ServerSetup}, got ${type}`) + if (type !== ControlMessageType.ServerSetup) + throw new Error(`server SETUP type must be ${ControlMessageType.ServerSetup}, got ${type}`) const advertisedLength = await buffer.getU16() const bufferLen = buffer.byteLength @@ -97,7 +101,8 @@ export class Client { async readClientSetup(buffer: ReadableWritableStreamBuffer): Promise { const type: ControlMessageType = await buffer.getNumberVarInt() - if (type !== ControlMessageType.ClientSetup) throw new Error(`client SETUP type must be ${ControlMessageType.ClientSetup}, got ${type}`) + if (type !== ControlMessageType.ClientSetup) + throw new Error(`client SETUP type must be ${ControlMessageType.ClientSetup}, got ${type}`) const advertisedLength = await buffer.getU16() const bufferLen = buffer.byteLength diff --git a/lib/transport/connection.ts b/lib/transport/connection.ts index 2c0f74f..5f13d4a 100644 --- a/lib/transport/connection.ts +++ b/lib/transport/connection.ts @@ -2,10 +2,13 @@ import * as Control from "./control" import { Objects } from "./objects" import { asError } from "../common/error" import { ControlStream } from "./stream" +import { getLogger } from "../common/logger" import { Publisher } from "./publisher" import { Subscriber } from "./subscriber" +const log = getLogger() + export class Connection { // The established WebTransport session. #quic: WebTransport @@ -67,29 +70,29 @@ export class Connection { async #runControl() { // Receive messages until the connection is closed. try { - console.log("starting control loop") - for (; ;) { + log.debug("starting control loop") + for (;;) { const msg = await this.#controlStream.recv() await this.#recv(msg) } } catch (e) { - console.error("Error in control stream:", e) + log.error("control stream error", e) throw e } } async #runObjects() { try { - console.log("starting object loop") - for (; ;) { + log.debug("starting object loop") + for (;;) { const obj = await this.#objects.recv() - console.log("object loop got obj", obj) + log.trace("object loop got obj", obj) if (!obj) break await this.#subscriber.recvObject(obj) } } catch (e) { - console.error("Error in object stream:", e) + log.error("object stream error", e) throw e } } diff --git a/lib/transport/control/client_setup.ts b/lib/transport/control/client_setup.ts index 24c190c..a8baff1 100644 --- a/lib/transport/control/client_setup.ts +++ b/lib/transport/control/client_setup.ts @@ -1,38 +1,41 @@ import { ControlMessageType, Version } from "." import { KeyValuePairs, Parameters } from "../base_data" import { ImmutableBytesBuffer, MutableBytesBuffer } from "../buffer" +import { getLogger } from "../../common/logger" + +const log = getLogger() export interface ClientSetup { - versions: Version[] - params: Parameters + versions: Version[] + params: Parameters } export namespace ClientSetup { - export function serialize(v: ClientSetup): Uint8Array { - const mainBuf = new MutableBytesBuffer(new Uint8Array()) - mainBuf.putVarInt(ControlMessageType.ClientSetup) - const payloadBuf = new MutableBytesBuffer(new Uint8Array()) - payloadBuf.putVarInt(v.versions.length) - v.versions.forEach((version) => { - payloadBuf.putVarInt(version) - }) - payloadBuf.putBytes(Parameters.serialize(v.params)) - mainBuf.putU16(payloadBuf.byteLength) - mainBuf.putBytes(payloadBuf.Uint8Array) - console.log("client setup: payload len:", payloadBuf.length, "msg len:", mainBuf.length) - return mainBuf.Uint8Array - } + export function serialize(v: ClientSetup): Uint8Array { + const mainBuf = new MutableBytesBuffer(new Uint8Array()) + mainBuf.putVarInt(ControlMessageType.ClientSetup) + const payloadBuf = new MutableBytesBuffer(new Uint8Array()) + payloadBuf.putVarInt(v.versions.length) + v.versions.forEach((version) => { + payloadBuf.putVarInt(version) + }) + payloadBuf.putBytes(Parameters.serialize(v.params)) + mainBuf.putU16(payloadBuf.byteLength) + mainBuf.putBytes(payloadBuf.Uint8Array) + log.trace("client setup: payload len:", payloadBuf.length, "msg len:", mainBuf.length) + return mainBuf.Uint8Array + } - export function deserialize(reader: ImmutableBytesBuffer): ClientSetup { - const supportedVersionLen = reader.getNumberVarInt() - const versions: Version[] = [] - for (let i = 0; i < supportedVersionLen; i++) { - versions.push(reader.getNumberVarInt() as Version) - } - const params = Parameters.deserialize(reader) - return { - versions, - params - } - } -} \ No newline at end of file + export function deserialize(reader: ImmutableBytesBuffer): ClientSetup { + const supportedVersionLen = reader.getNumberVarInt() + const versions: Version[] = [] + for (let i = 0; i < supportedVersionLen; i++) { + versions.push(reader.getNumberVarInt() as Version) + } + const params = Parameters.deserialize(reader) + return { + versions, + params, + } + } +} diff --git a/lib/transport/objects.ts b/lib/transport/objects.ts index e93846a..4add121 100644 --- a/lib/transport/objects.ts +++ b/lib/transport/objects.ts @@ -1,7 +1,17 @@ -import { SubgroupHeader, SubgroupObject, SubgroupReader, SubgroupType, SubgroupWriter } from "./subgroup" +import { SubgroupHeader, SubgroupReader, SubgroupType, SubgroupWriter } from "./subgroup" import { KeyValuePairs } from "./base_data" import { debug } from "./utils" -import { ImmutableBytesBuffer, MutableBytesBuffer, ReadableStreamBuffer, Reader, WritableStreamBuffer, Writer } from "./buffer" +import { + ImmutableBytesBuffer, + MutableBytesBuffer, + ReadableStreamBuffer, + Reader, + WritableStreamBuffer, + Writer, +} from "./buffer" +import { getLogger } from "../common/logger" + +const log = getLogger() export enum ObjectForwardingPreference { Datagram = "Datagram", @@ -59,7 +69,6 @@ export function isDatagram(obj: ObjectDatagram | SubgroupHeader): boolean { return obj.type in ObjectDatagramType } - export enum ObjectDatagramType { Type0x0 = 0x0, Type0x1 = 0x1, @@ -180,7 +189,6 @@ export namespace ObjectDatagram { } export function deserialize(reader: ImmutableBytesBuffer): ObjectDatagram { - const type = reader.getNumberVarInt() const alias = reader.getVarInt() const group = reader.getNumberVarInt() @@ -225,7 +233,7 @@ export class Objects { } async send(h: ObjectDatagram | SubgroupHeader): Promise { - const is_datagram = isDatagram(h); + const is_datagram = isDatagram(h) if (is_datagram) { // Datagram mode @@ -246,12 +254,11 @@ export class Objects { } async recv(): Promise { - console.log("Objects.recv waiting for streams") + log.trace("waiting for incoming streams") const streams = this.quic.incomingUnidirectionalStreams.getReader() - console.log("Objects.recv got streams", streams) const { value, done } = await streams.read() - console.log("Objects.recv got value, done", value, done) + log.trace("got stream", { done }) streams.releaseLock() if (done) return @@ -262,7 +269,7 @@ export class Objects { // Try to parse as SubgroupType try { const subgroupType = SubgroupType.try_from(type) - console.log("Objects.recv got type", subgroupType) + log.trace("parsed stream type", subgroupType) const track_alias = await r.getVarInt() const group_id = await r.getNumberVarInt() @@ -287,12 +294,12 @@ export class Objects { publisher_priority, } - console.log("Objects.recv got subgroup header", h) + log.trace("parsed subgroup header", h) return new SubgroupReader(h, r) } catch (e) { // Not a subgroup type, might be datagram or other type - console.log("transport/objects.ts: unknown stream type: ", type) + log.warn("unknown stream type", type) throw new Error(`unknown stream type: ${type}`) } } @@ -303,9 +310,7 @@ export class TrackWriter { // For compatibility with reader interface public header = { track_alias: 0n } - constructor( - public stream: Writer, - ) { } + constructor(public stream: Writer) {} async write(c: ObjectDatagram) { return this.stream.write(ObjectDatagram.serialize(c)) @@ -316,15 +321,11 @@ export class TrackWriter { } } - export class TrackReader { // Header with track_alias for routing public header: { track_alias: bigint } - constructor( - stream: Reader, - track_alias: bigint = 0n, - ) { + constructor(stream: Reader, track_alias: bigint = 0n) { this.stream = stream this.header = { track_alias } } @@ -372,4 +373,3 @@ export class TrackReader { await this.stream.close() } } - diff --git a/lib/transport/publisher.ts b/lib/transport/publisher.ts index 32a6c33..9d1eb46 100644 --- a/lib/transport/publisher.ts +++ b/lib/transport/publisher.ts @@ -3,6 +3,9 @@ import { ControlStream } from "./stream" import { Queue, Watch } from "../common/async" import { Objects, TrackWriter, ObjectDatagramType } from "./objects" import { SubgroupType, SubgroupWriter } from "./subgroup" +import { getLogger } from "../common/logger" + +const log = getLogger() export class Publisher { // Used to send control messages @@ -54,20 +57,20 @@ export class Publisher { } async recv(msg: Control.MessageWithType) { - const { type, message } = msg; + const { type, message } = msg switch (type) { case Control.ControlMessageType.Subscribe: await this.recvSubscribe(message) - break; + break case Control.ControlMessageType.Unsubscribe: this.recvUnsubscribe(message) - break; + break case Control.ControlMessageType.PublishNamespaceOk: this.recvPublishNamespaceOk(message) - break; + break case Control.ControlMessageType.PublishNamespaceError: this.recvPublishNamespaceError(message) - break; + break default: throw new Error(`unknown control message`) // impossible } @@ -84,7 +87,7 @@ export class Publisher { } publishNamespaceSend.onOk() - console.log("published namespace:", namespace) + log.debug("published namespace", namespace) } recvPublishNamespaceError(msg: Control.PublishNamespaceError) { @@ -95,7 +98,7 @@ export class Publisher { const publishNamespaceSend = this.#publishedNamespaces.get(namespace) if (!publishNamespaceSend) { // TODO debug this - console.warn(`publish namespace error for unknown announce: ${namespace}`) + log.warn(`publish namespace error for unknown announce: ${namespace}`) return } @@ -119,7 +122,7 @@ export class Publisher { id: msg.id, code: 0n, reason: e.message, - } + }, }) throw e } @@ -149,7 +152,7 @@ export class PublishNamespaceSend { } async ok() { - for (; ;) { + for (;;) { const [state, next] = this.#state.value() if (state === "ack") return if (state instanceof Error) throw state @@ -160,7 +163,7 @@ export class PublishNamespaceSend { } async active() { - for (; ;) { + for (;;) { const [state, next] = this.#state.value() if (state instanceof Error) throw state if (!next) return @@ -222,7 +225,7 @@ export class SubscribeRecv { if (this.#state !== "init") return this.#state = "ack" - console.log("got subscribe req:", this.#id, "track:", this.#trackAlias, "sending subscribe ok") + log.debug("sending subscribe ok", { id: this.#id, trackAlias: this.#trackAlias }) // NOTE(itzmanish): revisit this // Send the control message. @@ -235,12 +238,20 @@ export class SubscribeRecv { track_alias: this.#trackAlias, content_exists: 0, params: new Map(), - } + }, }) } // Close the subscription with an error. - async close({ code = 0n, reason = "", unsubscribe = true }: { code?: bigint; reason?: string; unsubscribe?: boolean }) { + async close({ + code = 0n, + reason = "", + unsubscribe = true, + }: { + code?: bigint + reason?: string + unsubscribe?: boolean + }) { if (this.#state === "closed") return const acked = this.#state === "ack" this.#state = "closed" @@ -248,13 +259,13 @@ export class SubscribeRecv { if (!acked) { return this.#control.send({ type: Control.ControlMessageType.SubscribeError, - message: { id: this.#id, code, reason } + message: { id: this.#id, code, reason }, }) } if (unsubscribe) { return this.#control.send({ type: Control.ControlMessageType.Unsubscribe, - message: { id: this.#id } + message: { id: this.#id }, }) } } diff --git a/lib/transport/stream.ts b/lib/transport/stream.ts index 27a2086..3549a49 100644 --- a/lib/transport/stream.ts +++ b/lib/transport/stream.ts @@ -1,17 +1,32 @@ - import { - ControlMessageType, FetchError, - MessageWithType, Publish, - PublishDone, PublishError, PublishNamespace, - PublishNamespaceDone, PublishNamespaceError, - PublishNamespaceOk, PublishOk, Unsubscribe, - Fetch, FetchOk, FetchCancel, - Subscribe, SubscribeOk, SubscribeError, - SubscribeUpdate, SubscribeNamespace, - SubscribeNamespaceOk, SubscribeNamespaceError, + ControlMessageType, + FetchError, + MessageWithType, + Publish, + PublishDone, + PublishError, + PublishNamespace, + PublishNamespaceDone, + PublishNamespaceError, + PublishNamespaceOk, + PublishOk, + Unsubscribe, + Fetch, + FetchOk, + FetchCancel, + Subscribe, + SubscribeOk, + SubscribeError, + SubscribeUpdate, + SubscribeNamespace, + SubscribeNamespaceOk, + SubscribeNamespaceError, } from "./control" import { debug } from "./utils" import { ImmutableBytesBuffer, ReadableWritableStreamBuffer, Reader, Writer } from "./buffer" +import { getLogger } from "../common/logger" + +const log = getLogger() export class ControlStream { private decoder: Decoder @@ -83,8 +98,8 @@ export class Decoder { const t = await this.messageType() const advertisedLength = await this.r.getU16() if (advertisedLength > this.r.byteLength) { - console.error( - `message: ${ControlMessageType.toString(t)} length mismatch: advertised ${advertisedLength} > ${this.r.byteLength} received`, + log.warn( + `message length mismatch: ${ControlMessageType.toString(t)} advertised ${advertisedLength} > ${this.r.byteLength} received`, ) // NOTE(itzmanish): should we have a timeout and retry few times even if timeout is reached? await this.r.waitForBytes(advertisedLength) @@ -219,7 +234,6 @@ export class Decoder { } return res - } } @@ -282,4 +296,3 @@ export class Encoder { await this.w.write(payload) } } - diff --git a/lib/transport/subgroup.ts b/lib/transport/subgroup.ts index 4c34026..84b8aa7 100644 --- a/lib/transport/subgroup.ts +++ b/lib/transport/subgroup.ts @@ -1,221 +1,223 @@ import { ImmutableBytesBuffer, MutableBytesBuffer, Reader, Writer } from "./buffer" import { KeyValuePairs } from "./base_data" import { Status } from "./objects" +import { getLogger } from "../common/logger" + +const log = getLogger() export interface SubgroupHeader { - type: SubgroupType - track_alias: bigint - group_id: number - subgroup_id?: number - publisher_priority: number + type: SubgroupType + track_alias: bigint + group_id: number + subgroup_id?: number + publisher_priority: number } export namespace SubgroupHeader { - export function serialize(header: SubgroupHeader): Uint8Array { - const buf = new MutableBytesBuffer(new Uint8Array()) - buf.putBytes(SubgroupType.serialize(header.type)) - buf.putVarInt(header.track_alias) - buf.putVarInt(header.group_id) - if (SubgroupType.hasExplicitSubgroupId(header.type) && header.subgroup_id !== undefined) { - buf.putVarInt(header.subgroup_id) - } - buf.putU8(header.publisher_priority) - return buf.Uint8Array - } + export function serialize(header: SubgroupHeader): Uint8Array { + const buf = new MutableBytesBuffer(new Uint8Array()) + buf.putBytes(SubgroupType.serialize(header.type)) + buf.putVarInt(header.track_alias) + buf.putVarInt(header.group_id) + if (SubgroupType.hasExplicitSubgroupId(header.type) && header.subgroup_id !== undefined) { + buf.putVarInt(header.subgroup_id) + } + buf.putU8(header.publisher_priority) + return buf.Uint8Array + } } export interface SubgroupObject { - object_id: number - extension_headers?: KeyValuePairs - status?: Status // only if payload is null - object_payload?: Uint8Array + object_id: number + extension_headers?: KeyValuePairs + status?: Status // only if payload is null + object_payload?: Uint8Array } export namespace SubgroupObject { - export function serialize(obj: SubgroupObject): Uint8Array { - const buf = new MutableBytesBuffer(new Uint8Array()) - buf.putVarInt(obj.object_id) - - if (obj.extension_headers) { - buf.putBytes(KeyValuePairs.serialize(obj.extension_headers)) - } - buf.putVarInt(obj.object_payload?.length ?? 0) - if (!obj.object_payload) { - buf.putVarInt(obj.status!) - } else { - buf.putBytes(obj.object_payload) - } - return buf.Uint8Array - } + export function serialize(obj: SubgroupObject): Uint8Array { + const buf = new MutableBytesBuffer(new Uint8Array()) + buf.putVarInt(obj.object_id) + + if (obj.extension_headers) { + buf.putBytes(KeyValuePairs.serialize(obj.extension_headers)) + } + buf.putVarInt(obj.object_payload?.length ?? 0) + if (!obj.object_payload) { + buf.putVarInt(obj.status!) + } else { + buf.putBytes(obj.object_payload) + } + return buf.Uint8Array + } } export enum SubgroupType { - Type0x10 = 0x10, - Type0x11 = 0x11, - Type0x12 = 0x12, - Type0x13 = 0x13, - Type0x14 = 0x14, - Type0x15 = 0x15, - Type0x18 = 0x18, - Type0x19 = 0x19, - Type0x1A = 0x1A, - Type0x1B = 0x1B, - Type0x1C = 0x1C, - Type0x1D = 0x1D, + Type0x10 = 0x10, + Type0x11 = 0x11, + Type0x12 = 0x12, + Type0x13 = 0x13, + Type0x14 = 0x14, + Type0x15 = 0x15, + Type0x18 = 0x18, + Type0x19 = 0x19, + Type0x1A = 0x1a, + Type0x1B = 0x1b, + Type0x1C = 0x1c, + Type0x1D = 0x1d, } export namespace SubgroupType { - export function serialize(type: SubgroupType): Uint8Array { - const w = new MutableBytesBuffer(new Uint8Array()) - w.putVarInt(type) - return w.Uint8Array - } - export function deserialize(reader: ImmutableBytesBuffer): SubgroupType { - return try_from(reader.getNumberVarInt()) - } - - // may throw if invalid value if provided - export function try_from(value: number | bigint): SubgroupType { - const v = typeof value === "bigint" ? Number(value) : value - - switch (v) { - case SubgroupType.Type0x10: - case SubgroupType.Type0x11: - case SubgroupType.Type0x12: - case SubgroupType.Type0x13: - case SubgroupType.Type0x14: - case SubgroupType.Type0x15: - case SubgroupType.Type0x18: - case SubgroupType.Type0x19: - case SubgroupType.Type0x1A: - case SubgroupType.Type0x1B: - case SubgroupType.Type0x1C: - case SubgroupType.Type0x1D: - return v as SubgroupType - default: - throw new Error(`invalid subgroup type: ${v}`) - } - } - - export function isSubgroupIdPresent(type: SubgroupType) { - switch (type) { - case SubgroupType.Type0x14: - case SubgroupType.Type0x15: - case SubgroupType.Type0x1C: - case SubgroupType.Type0x1D: - return true - default: - return false - } - } - - export function hasExplicitSubgroupId(type: SubgroupType) { - return isSubgroupIdPresent(type) - } - - export function isSubgroupIdZero(type: SubgroupType) { - switch (type) { - case SubgroupType.Type0x10: - case SubgroupType.Type0x11: - case SubgroupType.Type0x18: - case SubgroupType.Type0x19: - return true - default: - return false - } - } - - export function isSubgroupFirstObjectId(type: SubgroupType) { - return !(hasExplicitSubgroupId(type) || isSubgroupIdZero(type)) - } - - export function isExtensionPresent(type: SubgroupType) { - switch (type) { - case SubgroupType.Type0x11: - case SubgroupType.Type0x13: - case SubgroupType.Type0x15: - case SubgroupType.Type0x19: - case SubgroupType.Type0x1B: - case SubgroupType.Type0x1D: - return true - default: - return false - } - } - - export function contains_end_of_group(type: SubgroupType) { - switch (type) { - case SubgroupType.Type0x18: - case SubgroupType.Type0x19: - case SubgroupType.Type0x1A: - case SubgroupType.Type0x1B: - case SubgroupType.Type0x1C: - case SubgroupType.Type0x1D: - return true - default: - return false - } - } + export function serialize(type: SubgroupType): Uint8Array { + const w = new MutableBytesBuffer(new Uint8Array()) + w.putVarInt(type) + return w.Uint8Array + } + export function deserialize(reader: ImmutableBytesBuffer): SubgroupType { + return try_from(reader.getNumberVarInt()) + } + + // may throw if invalid value if provided + export function try_from(value: number | bigint): SubgroupType { + const v = typeof value === "bigint" ? Number(value) : value + + switch (v) { + case SubgroupType.Type0x10: + case SubgroupType.Type0x11: + case SubgroupType.Type0x12: + case SubgroupType.Type0x13: + case SubgroupType.Type0x14: + case SubgroupType.Type0x15: + case SubgroupType.Type0x18: + case SubgroupType.Type0x19: + case SubgroupType.Type0x1A: + case SubgroupType.Type0x1B: + case SubgroupType.Type0x1C: + case SubgroupType.Type0x1D: + return v as SubgroupType + default: + throw new Error(`invalid subgroup type: ${v}`) + } + } + + export function isSubgroupIdPresent(type: SubgroupType) { + switch (type) { + case SubgroupType.Type0x14: + case SubgroupType.Type0x15: + case SubgroupType.Type0x1C: + case SubgroupType.Type0x1D: + return true + default: + return false + } + } + + export function hasExplicitSubgroupId(type: SubgroupType) { + return isSubgroupIdPresent(type) + } + + export function isSubgroupIdZero(type: SubgroupType) { + switch (type) { + case SubgroupType.Type0x10: + case SubgroupType.Type0x11: + case SubgroupType.Type0x18: + case SubgroupType.Type0x19: + return true + default: + return false + } + } + + export function isSubgroupFirstObjectId(type: SubgroupType) { + return !(hasExplicitSubgroupId(type) || isSubgroupIdZero(type)) + } + + export function isExtensionPresent(type: SubgroupType) { + switch (type) { + case SubgroupType.Type0x11: + case SubgroupType.Type0x13: + case SubgroupType.Type0x15: + case SubgroupType.Type0x19: + case SubgroupType.Type0x1B: + case SubgroupType.Type0x1D: + return true + default: + return false + } + } + + export function contains_end_of_group(type: SubgroupType) { + switch (type) { + case SubgroupType.Type0x18: + case SubgroupType.Type0x19: + case SubgroupType.Type0x1A: + case SubgroupType.Type0x1B: + case SubgroupType.Type0x1C: + case SubgroupType.Type0x1D: + return true + default: + return false + } + } } - export class SubgroupWriter { - constructor( - public header: SubgroupHeader, - public stream: Writer, - ) { } - - async write(c: SubgroupObject) { - return this.stream.write(SubgroupObject.serialize(c)) - } - - async close() { - return this.stream.close() - } + constructor( + public header: SubgroupHeader, + public stream: Writer, + ) {} + + async write(c: SubgroupObject) { + return this.stream.write(SubgroupObject.serialize(c)) + } + + async close() { + return this.stream.close() + } } export class SubgroupReader { - constructor( - public header: SubgroupHeader, - public stream: Reader, - ) { } - - async read(): Promise { - if (await this.stream.done()) { - return - } - - const object_id = await this.stream.getNumberVarInt() - - let extHeaders: KeyValuePairs | undefined - if (SubgroupType.isExtensionPresent(this.header.type)) { - extHeaders = await KeyValuePairs.deserialize_with_reader(this.stream) - } - - console.log("subgroup header", object_id, extHeaders, this.stream) - - let obj_payload_len = await this.stream.getNumberVarInt() - - let object_payload: Uint8Array | undefined - let status: Status | undefined - - console.log("subgroup read", object_id, obj_payload_len) - - if (obj_payload_len == 0) { - status = Status.try_from(await this.stream.getNumberVarInt()) - } else { - object_payload = await this.stream.read(obj_payload_len) - } - - console.log("read success??", object_id, status, extHeaders, object_payload) - return { - object_id, - status, - extension_headers: extHeaders, - object_payload, - } - } - - async close() { - await this.stream.close() - } + constructor( + public header: SubgroupHeader, + public stream: Reader, + ) {} + + async read(): Promise { + if (await this.stream.done()) { + return + } + + const object_id = await this.stream.getNumberVarInt() + + let extHeaders: KeyValuePairs | undefined + if (SubgroupType.isExtensionPresent(this.header.type)) { + extHeaders = await KeyValuePairs.deserialize_with_reader(this.stream) + } + + log.trace("subgroup header", object_id, extHeaders) + + let obj_payload_len = await this.stream.getNumberVarInt() + + let object_payload: Uint8Array | undefined + let status: Status | undefined + + log.trace("subgroup read", object_id, obj_payload_len) + + if (obj_payload_len == 0) { + status = Status.try_from(await this.stream.getNumberVarInt()) + } else { + object_payload = await this.stream.read(obj_payload_len) + } + + log.trace("subgroup read success", object_id, status) + return { + object_id, + status, + extension_headers: extHeaders, + object_payload, + } + } + + async close() { + await this.stream.close() + } } diff --git a/lib/transport/subscriber.ts b/lib/transport/subscriber.ts index ac1d8ab..1f38ccf 100644 --- a/lib/transport/subscriber.ts +++ b/lib/transport/subscriber.ts @@ -5,6 +5,9 @@ import type { TrackReader } from "./objects" import { debug } from "./utils" import { ControlStream } from "./stream" import { SubgroupReader } from "./subgroup" +import { getLogger } from "../common/logger" + +const log = getLogger() export interface TrackInfo { track_alias: bigint @@ -39,7 +42,7 @@ export class Subscriber { } async recv(msg: Control.MessageWithType) { - const { type, message } = msg; + const { type, message } = msg switch (type) { case Control.ControlMessageType.PublishNamespace: await this.recvPublishNamespace(message) @@ -68,7 +71,7 @@ export class Subscriber { await this.#control.send({ type: Control.ControlMessageType.PublishNamespaceOk, - message: { id: msg.id } + message: { id: msg.id }, }) const publishNamespace = new PublishNamespaceRecv(this.#control, msg.namespace, msg.id) @@ -89,7 +92,7 @@ export class Subscriber { message: { id, namespace, - } + }, } await this.#control.send(msg) } @@ -113,7 +116,7 @@ export class Subscriber { filter_type: Control.FilterType.NextGroupStart, forward: 1, // always forward params: new Map(), - } + }, } await this.#control.send(subscription_req) @@ -126,17 +129,17 @@ export class Subscriber { if (this.#trackToIDMap.has(track)) { const trackID = this.#trackToIDMap.get(track) if (trackID === undefined) { - console.warn(`Exception track ${track} not found in trackToIDMap.`) + log.warn(`track ${track} not found in trackToIDMap`) return } try { await this.#control.send({ type: Control.ControlMessageType.Unsubscribe, message: { id: trackID } }) this.#trackToIDMap.delete(track) } catch (error) { - console.error(`Failed to unsubscribe from track ${track}:`, error) + log.error(`failed to unsubscribe from track ${track}`, error) } } else { - console.warn(`During unsubscribe request initiation attempt track ${track} not found in trackToIDMap.`) + log.warn(`unsubscribe attempted but track ${track} not found in trackToIDMap`) } } @@ -156,7 +159,7 @@ export class Subscriber { callback(msg.id) } - console.log("subscribe ok", msg) + log.debug("subscribe ok", msg) subscribe.onOk(msg.track_alias) } @@ -179,23 +182,23 @@ export class Subscriber { } async recvObject(reader: TrackReader | SubgroupReader) { - console.log("got object on recvObject", reader) + log.trace("recvObject", reader) // Get track alias from reader header const track_alias = reader.header.track_alias // Map track alias back to subscription ID const subscriptionId = this.#aliasToSubscriptionMap.get(track_alias) - console.log("got subscriptionId", subscriptionId) + log.trace("resolved subscriptionId", subscriptionId) const callback = async (id: bigint) => { const subscribe = this.#subscribe.get(id) if (!subscribe) { throw new Error(`data for unknown subscription: ${id}`) } - console.log("doing subscribe on data", reader) + log.trace("dispatching data to subscription", reader) return subscribe.onData(reader) } if (subscriptionId === undefined) { - console.warn(`Exception track alias ${track_alias} not found in aliasToSubscriptionMap.`) + log.warn(`track alias ${track_alias} not found in aliasToSubscriptionMap`) this.#pendingTrack.set(track_alias, callback) return } @@ -227,7 +230,7 @@ export class PublishNamespaceRecv { // Send the control message. return this.#control.send({ type: Control.ControlMessageType.PublishNamespaceOk, - message: { id: this.#id } + message: { id: this.#id }, }) } @@ -237,7 +240,7 @@ export class PublishNamespaceRecv { return this.#control.send({ type: Control.ControlMessageType.PublishNamespaceError, - message: { id: this.#id, code, reason } + message: { id: this.#id, code, reason }, }) } } @@ -270,11 +273,11 @@ export class SubscribeSend { } onOk(trackAlias: bigint) { - console.log("setting track alias", trackAlias) + log.debug("setting track alias", trackAlias) this.#trackAlias = trackAlias } - // FIXME(itzmanish): implement correctly + // FIXME(itzmanish): implement correctly async onDone(code: bigint, streamCount: bigint, reason: string) { throw new Error(`TODO onDone`) } @@ -293,7 +296,7 @@ export class SubscribeSend { } async onData(reader: TrackReader | SubgroupReader) { - console.log("subscribe send onData", reader) + log.trace("onData", reader) if (!this.#data.closed()) await this.#data.push(reader) } diff --git a/lib/transport/utils.ts b/lib/transport/utils.ts index f658e06..bb75de9 100644 --- a/lib/transport/utils.ts +++ b/lib/transport/utils.ts @@ -1,7 +1,11 @@ -export function debug(...msg: any[]) { - console.log("itzmanish:", ...msg) +import { getLogger } from "../common/logger" + +const log = getLogger() + +export function debug(...msg: unknown[]) { + log.trace(...msg) } export async function sleep(ms: number) { - return new Promise((resolve) => setTimeout(resolve, ms)) -} \ No newline at end of file + return new Promise((resolve) => setTimeout(resolve, ms)) +} diff --git a/lib/video-moq/index.ts b/lib/video-moq/index.ts index e16df1d..61c4056 100644 --- a/lib/video-moq/index.ts +++ b/lib/video-moq/index.ts @@ -1,6 +1,9 @@ import Player from "../playback/index" import { FULLSCREEN_BUTTON, PICTURE_IN_PICTURE_BUTTON, VOLUME_CONTROL } from "./control-buttons" import { ENTER_PIP_SVG, EXIT_PIP_SVG, PAUSE_SVG, PLAY_SVG } from "./icons" +import { getLogger } from "../common/logger" + +const log = getLogger() /** * This stylesheet is self contained within the shadow root @@ -62,12 +65,12 @@ export class VideoMoq extends HTMLElement { set muted(mute: boolean) { if (mute) { this.mute().catch((err) => { - console.error("Error muting:", err) + log.error("Error muting:", err) }) this.dispatchEvent(new Event("volumechange")) } else { this.unmute().catch((err) => { - console.error("Error unmuting:", err) + log.error("Error unmuting:", err) }) this.dispatchEvent(new Event("volumechange")) } @@ -80,11 +83,11 @@ export class VideoMoq extends HTMLElement { set fullscreen(fullscreen: boolean) { if (fullscreen) { this.requestFullscreen().catch((err) => { - console.error("Error entering fullscreen:", err) + log.error("Error entering fullscreen:", err) }) } else { this.exitFullscreen().catch((err) => { - console.error("Error exiting fullscreen:", err) + log.error("Error exiting fullscreen:", err) }) } } @@ -120,25 +123,25 @@ export class VideoMoq extends HTMLElement { // Bind event listeners to add and remove from lists. this.playPauseEventHandler = () => { this.togglePlayPause().catch((err) => { - console.error("Error toggling play/pause:", err) + log.error("Error toggling play/pause:", err) }) } this.toggleMuteEventHandler = () => { this.toggleMute().catch((err) => { - console.error("Error toggling mute:", err) + log.error("Error toggling mute:", err) }) } this.togglePictureInPictureEventHandler = () => { this.togglePictureInPicture().catch((err) => { - console.error("Error toggling picture-in-picture: ", err) + log.error("Error toggling picture-in-picture: ", err) }) } this.setVolume = (e: Event) => { this.handleVolumeChange(e as Event & { currentTarget: HTMLInputElement }).catch((err) => { - console.error("Error setting volume: ", err) + log.error("Error setting volume: ", err) }) } @@ -166,7 +169,7 @@ export class VideoMoq extends HTMLElement { * */ disconnectedCallback() { this.destroy().catch((error) => { - console.error("Error while destroying:", error) + log.error("Error while destroying:", error) }) } @@ -211,7 +214,7 @@ export class VideoMoq extends HTMLElement { private load() { this.destroy().catch((error) => { - console.error("Error while destroying:", error) + log.error("Error while destroying:", error) }) this.shadow.innerHTML = /*html*/ ` @@ -367,7 +370,7 @@ export class VideoMoq extends HTMLElement { await this.pause() } } catch (error) { - console.error("Error toggling play/pause:", error) + log.error("Error toggling play/pause:", error) } finally { if (this.#playButton) { this.#playButton.disabled = false @@ -378,20 +381,20 @@ export class VideoMoq extends HTMLElement { public play(): Promise { return this.player ? this.player.play().then(() => { - if (!this.#playButton) return - this.#playButton.innerHTML = PAUSE_SVG - this.#playButton.ariaLabel = "Pause" - }) + if (!this.#playButton) return + this.#playButton.innerHTML = PAUSE_SVG + this.#playButton.ariaLabel = "Pause" + }) : Promise.resolve() } public pause(): Promise { return this.player ? this.player.pause().then(() => { - if (!this.#playButton) return - this.#playButton.innerHTML = PLAY_SVG - this.#playButton.ariaLabel = "Play" - }) + if (!this.#playButton) return + this.#playButton.innerHTML = PLAY_SVG + this.#playButton.ariaLabel = "Play" + }) : Promise.resolve() } @@ -409,7 +412,7 @@ export class VideoMoq extends HTMLElement { await this.mute() } } catch (error) { - console.error("Error toggling mute:", error) + log.error("Error toggling mute:", error) } finally { if (this.#volumeButton) { this.#volumeButton.disabled = false @@ -420,23 +423,23 @@ export class VideoMoq extends HTMLElement { public unmute(): Promise { return this.player ? this.player.mute(false).then(() => { - if (!this.#volumeButton) return - this.#volumeButton.ariaLabel = "Mute" - this.#volumeButton.innerText = "🔊" - this.#volumeRange!.value = this.previousVolume.toString() - }) + if (!this.#volumeButton) return + this.#volumeButton.ariaLabel = "Mute" + this.#volumeButton.innerText = "🔊" + this.#volumeRange!.value = this.previousVolume.toString() + }) : Promise.resolve() } public mute(): Promise { return this.player ? this.player.mute(true).then(() => { - if (!this.#volumeButton) return - this.#volumeButton.ariaLabel = "Unmute" - this.#volumeButton.innerText = "🔇" - this.previousVolume = parseFloat(this.#volumeRange!.value) - this.#volumeRange!.value = "0" - }) + if (!this.#volumeButton) return + this.#volumeButton.ariaLabel = "Unmute" + this.#volumeButton.innerText = "🔇" + this.previousVolume = parseFloat(this.#volumeRange!.value) + this.#volumeRange!.value = "0" + }) : Promise.resolve() } @@ -462,7 +465,7 @@ export class VideoMoq extends HTMLElement { await this.#base.requestFullscreen() } } catch (error) { - console.error("Error entering fullscreen:", error) + log.error("Error entering fullscreen:", error) } } @@ -470,7 +473,7 @@ export class VideoMoq extends HTMLElement { try { await document.exitFullscreen() } catch (error) { - console.error("Error exiting fullscreen:", error) + log.error("Error exiting fullscreen:", error) } } @@ -494,12 +497,12 @@ export class VideoMoq extends HTMLElement { } if (!this.#canvas) { - console.warn("Canvas element not found.") + log.warn("Canvas element not found.") return } if (!this.#base) { - console.warn("Base element not found.") + log.warn("Base element not found.") return } @@ -511,7 +514,7 @@ export class VideoMoq extends HTMLElement { })) if (!this.#pipWindow) { - console.warn("Picture-in-Picture window not found.") + log.warn("Picture-in-Picture window not found.") return } @@ -559,13 +562,13 @@ export class VideoMoq extends HTMLElement { this.#pipWindow?.close() this.#pipWindow = undefined } else { - console.warn("Failed to restore video element! Check DOM structure.") + log.warn("Failed to restore video element! Check DOM structure.") } } private async togglePictureInPicture() { if (!("documentPictureInPicture" in window)) { - console.warn("DocumentPictureInPicture API is not supported.") + log.warn("DocumentPictureInPicture API is not supported.") return } @@ -576,7 +579,7 @@ export class VideoMoq extends HTMLElement { this.exitPictureInPicture() } } catch (error) { - console.error("Error toggling Picture-in-Picture:", error) + log.error("Error toggling Picture-in-Picture:", error) } } @@ -601,13 +604,13 @@ export class VideoMoq extends HTMLElement { this.#trackList.querySelectorAll("li").forEach((element) => { element.addEventListener("click", () => { this.switchTrack(element.dataset.name || null).catch((error) => { - console.error("Error switching track:", error) + log.error("Error switching track:", error) }) }) element.addEventListener("keydown", (e) => { if (e.key === "Enter" || e.key === " ") { this.switchTrack(element.dataset.name || null).catch((error) => { - console.error("Error switching track:", error) + log.error("Error switching track:", error) }) } }) @@ -638,7 +641,7 @@ export class VideoMoq extends HTMLElement { // Check for NaN or negative values if (isNaN(parsed) || parsed <= 0) { - console.warn(`Invalid value "${value}" for dimension, using default: ${defaultValue}px`) + log.warn(`Invalid value "${value}" for dimension, using default: ${defaultValue}px`) return defaultValue } @@ -647,7 +650,7 @@ export class VideoMoq extends HTMLElement { /** Prints error and displays it in a red box */ private fail(error?: Error) { - console.error("Moq Player failed, please reload", error) + log.error("Moq Player failed, please reload", error) this.error = error || new Error("Unknown error") @@ -679,7 +682,7 @@ export class VideoMoq extends HTMLElement { set currentTime(value: number) { if (value < this.duration) { - console.warn("Seeking within the buffer is not supported in live mode.") + log.warn("Seeking within the buffer is not supported in live mode.") } } @@ -697,3 +700,7 @@ export class VideoMoq extends HTMLElement { // Register the custom element customElements.define("video-moq", VideoMoq) export default VideoMoq + +// Re-export logger API so consumers of @moq-js/player can configure logging. +export { setGlobalLogger, getGlobalLogger, createConsoleLogger, notifyLoggerLevelChanged } from "../common/logger" +export type { Logger, LogLevel } from "../common/logger"