fix(playback): wait for mpv's reply to every reused-instance IPC command (#1876)

Every reused-instance MPV IPC command now carries a request_id and waits, within the existing 2 s bound, for mpv's matching reply; an error or missing reply falls back to a fresh launch like a socket failure, and the racing post-loadfile seek is gone. MPV launches run one at a time through a launch queue so overlapping launches never act on the same child; Stop on a queued launch delegates to the in-flight launch's own closer, a failed exit stays retryable, and the dock keeps the newest launch. Position polling and output-error reporting moved out of mpv-session.service.ts, which leaves the max-lines baseline.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
4grayandClaude Opus 5.5 authored and GitHub committed 2026-10-11 22:42:43 +02:00
1 parent 94570bc387
commit 4e9795ac54
21 files changed
+2920 -394

No files matched your search

@@ -0,0 +1,9 @@
---
type: fix
area: playback
---
With "Reuse player instance" on, IPTVnator now waits for MPV to confirm each
command sent to the running player. A command MPV rejects or never answers
falls back to a fresh MPV launch instead of reporting the stream as opened
while nothing changed.
@@ -30,6 +30,26 @@ describe('ExternalPlayerSessionRegistry', () => {
expect(updates.at(-1)?.status).toBe('opened');
});
it('numbers sessions in the order they began', () => {
const first = registry.beginSession({
player: 'mpv',
title: 'First',
streamUrl: 'https://example.com/one.m3u8',
});
const second = registry.beginSession({
player: 'mpv',
title: 'Second',
streamUrl: 'https://example.com/two.m3u8',
});
expect(first.launchOrder).toEqual(expect.any(Number));
expect(second.launchOrder).toBeGreaterThan(first.launchOrder ?? 0);
// Later updates keep the order the session began with.
expect(registry.markOpened(first.id)?.launchOrder).toBe(
first.launchOrder
);
});
it('keeps close capability and closes the session explicitly', async () => {
const close = jest.fn();
const session = registry.beginSession({
@@ -144,9 +164,9 @@ describe('ExternalPlayerSessionRegistry', () => {
registry.markClosed(closed.id);
expect(registry.markPlaying(closed.id)?.status).toBe('closed');
expect(
registry.markError(closed.id, 'Late process exit')?.status
).toBe('closed');
expect(registry.markError(closed.id, 'Late process exit')?.status).toBe(
'closed'
);
expect(updates.at(-1)).toMatchObject({
id: closed.id,
status: 'closed',
@@ -44,6 +44,7 @@ function isRestorableSession(session: ExternalPlayerSession): boolean {
export class ExternalPlayerSessionRegistry {
private readonly sessions = new Map<string, ExternalPlayerSessionRuntime>();
private activeSessionId: string | null = null;
private launchCount = 0;
constructor(
private readonly onUpdate: (session: ExternalPlayerSession) => void
@@ -63,6 +64,7 @@ export class ExternalPlayerSessionRegistry {
contentInfo: options.contentInfo,
startedAt,
updatedAt: startedAt,
launchOrder: ++this.launchCount,
canClose: false,
};
@@ -0,0 +1,298 @@
jest.mock('net', () => ({
createConnection: jest.fn(),
}));
jest.mock('./external-player-runtime', () => ({
traceExternalPlayer: jest.fn(),
}));
import { EventEmitter } from 'events';
import { createConnection } from 'net';
import {
MPV_IPC_COMMAND_TIMEOUT_MS,
MpvIpcCommandError,
sendMpvCommand,
} from './mpv-ipc-command';
type SocketMock = EventEmitter & {
write: jest.Mock;
destroy: jest.Mock;
destroyed: boolean;
};
const createConnectionMock = createConnection as unknown as jest.Mock;
const socketPath = '/tmp/mpv-ipc-command.sock';
const streamUrl = 'https://example.com/stream.m3u8?token=secret-token';
function nextSocket(): SocketMock {
const socket: SocketMock = Object.assign(new EventEmitter(), {
write: jest.fn(() => true),
destroy: jest.fn(() => {
socket.destroyed = true;
}),
destroyed: false,
});
createConnectionMock.mockReturnValueOnce(socket);
return socket;
}
function writtenRequest(socket: SocketMock): {
command: Array<string | number>;
request_id: number;
} {
return JSON.parse(String(socket.write.mock.calls[0][0]).trim());
}
function reply(socket: SocketMock, ...lines: Array<object | string>): void {
socket.emit(
'data',
Buffer.from(
lines
.map((line) =>
typeof line === 'string' ? line : JSON.stringify(line)
)
.join('\n') + '\n'
)
);
}
async function settle(): Promise<void> {
await new Promise<void>((resolve) => setImmediate(resolve));
}
describe('sendMpvCommand', () => {
beforeEach(() => {
jest.clearAllMocks();
});
afterEach(() => {
jest.useRealTimers();
});
it('sends a request_id and resolves with the matching reply', async () => {
const socket = nextSocket();
const pending = sendMpvCommand(socketPath, 'loadfile', [
streamUrl,
'replace',
-1,
'start=0',
]);
socket.emit('connect');
const request = writtenRequest(socket);
expect(request.command).toEqual([
'loadfile',
streamUrl,
'replace',
-1,
'start=0',
]);
expect(Number.isInteger(request.request_id)).toBe(true);
expect(String(socket.write.mock.calls[0][0]).endsWith('\n')).toBe(true);
reply(socket, {
data: { playlist_entry_id: 1 },
request_id: request.request_id,
error: 'success',
});
await expect(pending).resolves.toEqual({
request_id: request.request_id,
error: 'success',
data: { playlist_entry_id: 1 },
});
expect(socket.destroy).toHaveBeenCalledTimes(1);
});
it('gives consecutive commands distinct request ids', async () => {
const first = nextSocket();
const second = nextSocket();
const firstPending = sendMpvCommand(socketPath, 'set_property', [
'user-agent',
'agent',
]);
const secondPending = sendMpvCommand(socketPath, 'set_property', [
'referrer',
'https://example.com/',
]);
first.emit('connect');
second.emit('connect');
const firstId = writtenRequest(first).request_id;
const secondId = writtenRequest(second).request_id;
expect(secondId).not.toBe(firstId);
reply(first, { request_id: firstId, error: 'success' });
reply(second, { request_id: secondId, error: 'success' });
await expect(firstPending).resolves.toMatchObject({
request_id: firstId,
});
await expect(secondPending).resolves.toMatchObject({
request_id: secondId,
});
});
it('resolves false without writing when dispatch is vetoed', async () => {
const socket = nextSocket();
const onDispatched = jest.fn();
const pending = sendMpvCommand(socketPath, 'loadfile', [streamUrl], {
shouldDispatch: () => false,
onDispatched,
});
socket.emit('connect');
await expect(pending).resolves.toBe(false);
expect(socket.write).not.toHaveBeenCalled();
expect(onDispatched).not.toHaveBeenCalled();
expect(socket.destroy).toHaveBeenCalledTimes(1);
});
it('rejects with mpv error text after the command was dispatched', async () => {
const socket = nextSocket();
const onDispatched = jest.fn();
const pending = sendMpvCommand(socketPath, 'seek', ['10', 'absolute'], {
onDispatched,
});
socket.emit('connect');
expect(onDispatched).toHaveBeenCalledTimes(1);
reply(socket, {
request_id: writtenRequest(socket).request_id,
error: 'error running command',
});
await expect(pending).rejects.toMatchObject({
name: 'MpvIpcCommandError',
command: 'seek',
dispatched: true,
mpvError: 'error running command',
message: 'mpv seek: error running command',
});
expect(socket.destroy).toHaveBeenCalledTimes(1);
});
it.each([
['a missing error', {}],
['a non-string error', { error: 0 }],
])(
'rejects a matching reply with %s instead of confirming it',
async (_label, fields) => {
const socket = nextSocket();
const pending = sendMpvCommand(socketPath, 'loadfile', [streamUrl]);
socket.emit('connect');
reply(socket, {
request_id: writtenRequest(socket).request_id,
data: { playlist_entry_id: 1 },
...fields,
});
await expect(pending).rejects.toMatchObject({
name: 'MpvIpcCommandError',
dispatched: true,
mpvError: null,
message: 'mpv loadfile: malformed reply',
});
}
);
it('rejects an undispatched command on a socket error without echoing arguments', async () => {
const socket = nextSocket();
const pending = sendMpvCommand(socketPath, 'loadfile', [streamUrl]);
socket.emit('error', new Error('connect ENOENT'));
const error = (await pending.catch((caught) => caught)) as unknown;
expect(error).toBeInstanceOf(MpvIpcCommandError);
expect(error).toMatchObject({
dispatched: false,
mpvError: null,
message: 'mpv loadfile: connect ENOENT',
});
expect(socket.write).not.toHaveBeenCalled();
expect(String((error as Error).message)).not.toContain('secret');
});
it('rejects when the socket closes before a reply', async () => {
const socket = nextSocket();
const pending = sendMpvCommand(socketPath, 'quit', []);
socket.emit('connect');
socket.emit('close');
await expect(pending).rejects.toMatchObject({
command: 'quit',
dispatched: true,
message: 'mpv quit: socket closed before reply',
});
});
it('times out a reply that never comes', async () => {
jest.useFakeTimers();
const socket = nextSocket();
const pending = sendMpvCommand(socketPath, 'loadfile', [streamUrl]);
socket.emit('connect');
let settled = false;
void pending.catch(() => (settled = true));
await jest.advanceTimersByTimeAsync(MPV_IPC_COMMAND_TIMEOUT_MS - 1);
expect(settled).toBe(false);
await jest.advanceTimersByTimeAsync(1);
expect(settled).toBe(true);
await expect(pending).rejects.toMatchObject({
dispatched: true,
message: 'mpv loadfile: reply timed out',
});
expect(socket.destroy).toHaveBeenCalledTimes(1);
});
it('skips events, foreign replies and malformed lines, and joins split chunks', async () => {
const socket = nextSocket();
const pending = sendMpvCommand(socketPath, 'loadfile', [streamUrl]);
socket.emit('connect');
const requestId = writtenRequest(socket).request_id;
let settled = false;
void pending.then(
() => (settled = true),
() => (settled = true)
);
reply(
socket,
{ event: 'start-file', playlist_entry_id: 1 },
{
event: 'end-file',
reason: 'error',
file_error: 'loading failed',
},
{ request_id: requestId + 1, error: 'error running command' },
'not json',
'42'
);
await settle();
expect(settled).toBe(false);
const ownReply = JSON.stringify({
request_id: requestId,
error: 'success',
});
socket.emit('data', Buffer.from(ownReply.slice(0, 10)));
await settle();
expect(settled).toBe(false);
socket.emit('data', Buffer.from(ownReply.slice(10) + '\n'));
await expect(pending).resolves.toMatchObject({
request_id: requestId,
error: 'success',
});
});
it('ignores data that arrives after settling', async () => {
const socket = nextSocket();
const pending = sendMpvCommand(socketPath, 'quit', []);
socket.emit('connect');
const requestId = writtenRequest(socket).request_id;
reply(socket, { request_id: requestId, error: 'success' });
await expect(pending).resolves.toMatchObject({ error: 'success' });
expect(() =>
reply(socket, { request_id: requestId, error: 'late failure' })
).not.toThrow();
expect(() => socket.emit('close')).not.toThrow();
expect(socket.destroy).toHaveBeenCalledTimes(1);
});
});
@@ -0,0 +1,158 @@
import { createConnection } from 'net';
import { traceExternalPlayer } from './external-player-runtime';
/**
* Bounds connect, write and the wait for mpv's reply. It matches the IPC
* quit bound the reuse teardown relies on before its exit windows start.
*/
export const MPV_IPC_COMMAND_TIMEOUT_MS = 2_000;
/** One newline-delimited JSON reply of mpv's IPC protocol. */
export interface MpvIpcReply {
request_id: number;
error: string;
data?: unknown;
}
export interface MpvIpcCommandOptions {
/** Checked once connected; a `false` skips the write and resolves `false`. */
shouldDispatch?: () => boolean;
/** Runs right after the command was written, before any reply. */
onDispatched?: () => void;
}
/**
* A command mpv answered with an error, or one whose reply never came.
* `dispatched` tells whether the command reached the socket, so callers can
* treat the player state as possibly changed. The message never carries
* command arguments, which may include stream URLs.
*/
export class MpvIpcCommandError extends Error {
constructor(
readonly command: string,
message: string,
readonly dispatched: boolean,
readonly mpvError: string | null = null
) {
super(`mpv ${command}: ${message}`);
this.name = 'MpvIpcCommandError';
}
}
let nextRequestId = 1;
function parseReplyLine(line: string): Partial<MpvIpcReply> | null {
try {
const parsed: unknown = JSON.parse(line);
return parsed && typeof parsed === 'object'
? (parsed as Partial<MpvIpcReply>)
: null;
} catch {
return null;
}
}
/**
* Sends one command to a reused mpv over its IPC socket and resolves with
* mpv's matching reply. Every command carries a `request_id`; event lines
* and replies to other requests on the same connection are skipped. Resolves
* `false` when `shouldDispatch` vetoed the write, and rejects with
* {@link MpvIpcCommandError} on an error reply, a missing reply, or a socket
* failure.
*/
export function sendMpvCommand(
socketPath: string,
command: string,
args: Array<string | number>,
options: MpvIpcCommandOptions = {}
): Promise<MpvIpcReply | false> {
return new Promise((resolve, reject) => {
const requestId = nextRequestId++;
const client = createConnection(socketPath);
const request =
JSON.stringify({
command: [command, ...args],
request_id: requestId,
}) + '\n';
let settled = false;
let dispatched = false;
let buffered = '';
let timeoutHandle: NodeJS.Timeout | null = null;
const settle = (
outcome: { reply: MpvIpcReply | false } | { error: Error }
) => {
if (settled) return;
settled = true;
if (timeoutHandle) clearTimeout(timeoutHandle);
if (!client.destroyed) client.destroy();
if ('error' in outcome) {
reject(outcome.error);
} else {
resolve(outcome.reply);
}
};
const fail = (message: string, mpvError: string | null = null) =>
settle({
error: new MpvIpcCommandError(
command,
message,
dispatched,
mpvError
),
});
const consumeLine = (line: string) => {
const reply = parseReplyLine(line);
if (!reply || reply.request_id !== requestId) return;
// mpv always answers with a string `error`; anything else is not
// a confirmation and must not skip the fallback.
if (typeof reply.error !== 'string') {
fail('malformed reply');
return;
}
const error = reply.error;
traceExternalPlayer('mpv ipc reply', { command, error });
if (error === 'success') {
settle({
reply: { request_id: requestId, error, data: reply.data },
});
} else {
fail(error, error);
}
};
client.on('connect', () => {
if (options.shouldDispatch && !options.shouldDispatch()) {
settle({ reply: false });
return;
}
traceExternalPlayer('mpv ipc command', {
command,
argsCount: args.length,
});
try {
client.write(request);
dispatched = true;
options.onDispatched?.();
} catch (error) {
fail(error instanceof Error ? error.message : String(error));
}
});
client.on('data', (chunk) => {
if (settled) return;
buffered += chunk.toString();
const lines = buffered.split('\n');
buffered = lines.pop() ?? '';
for (const line of lines) {
if (line.trim()) consumeLine(line);
if (settled) return;
}
});
client.on('error', (error) => fail(error.message));
client.on('close', () => fail('socket closed before reply'));
timeoutHandle = setTimeout(
() => fail('reply timed out'),
MPV_IPC_COMMAND_TIMEOUT_MS
);
timeoutHandle.unref();
});
}
@@ -0,0 +1,55 @@
import { EventEmitter } from 'events';
export type MpvSocketMock = EventEmitter & {
write: jest.Mock;
end: jest.Mock;
destroy: jest.Mock;
};
export interface MpvSocketMockOptions {
/** Collects every chunk written to the socket. */
written?: string[];
/** Emit `connect` on the next tick; `false` leaves that to the test. */
connect?: boolean;
/**
* mpv's `error` field for a written command, `'success'` by default.
* `null` leaves the command unanswered.
*/
reply?: (command: Array<string | number>) => string | null;
}
interface MpvIpcRequest {
command: Array<string | number>;
request_id: number;
}
/**
* A reused-instance mpv IPC socket: records writes and answers each command
* on the next tick with a reply carrying its `request_id`, as mpv does.
*/
export function createMpvSocketMock({
written,
connect = true,
reply = () => 'success',
}: MpvSocketMockOptions = {}): MpvSocketMock {
const socket: MpvSocketMock = Object.assign(new EventEmitter(), {
write: jest.fn((chunk: string) => {
written?.push(chunk);
const request = JSON.parse(chunk.trim()) as MpvIpcRequest;
const error = reply(request.command);
if (error !== null) {
const line =
JSON.stringify({ request_id: request.request_id, error }) +
'\n';
setImmediate(() => socket.emit('data', Buffer.from(line)));
}
return true;
}),
end: jest.fn(),
destroy: jest.fn(),
});
if (connect) {
setImmediate(() => socket.emit('connect'));
}
return socket;
}
@@ -0,0 +1,856 @@
/**
* MPV launch queue: launches run one at a time, only the newest queued one
* runs, and Stop on a queued launch delegates to the in-flight launch's and
* the stored player's own closers. Reply handling lives in
* `mpv-session.service.ipc-reply.spec.ts`.
*/
jest.mock('electron', () => ({
ipcMain: {
handle: jest.fn(),
},
}));
jest.mock('child_process', () => ({
spawn: jest.fn(),
}));
jest.mock('net', () => ({
createConnection: jest.fn(),
createServer: jest.fn(),
}));
jest.mock('../app', () => ({
__esModule: true,
default: {
mainWindow: null,
},
}));
jest.mock('../services/store.service', () => ({
MPV_PLAYER_ARGUMENTS: 'MPV_PLAYER_ARGUMENTS',
MPV_PLAYER_PATH: 'MPV_PLAYER_PATH',
MPV_REUSE_INSTANCE: 'MPV_REUSE_INSTANCE',
store: {
get: jest.fn(),
set: jest.fn(),
},
}));
import {
MPV_PLAYER_PATH,
MPV_REUSE_INSTANCE,
store,
} from '../services/store.service';
import { externalPlayerSessions } from './external-player-runtime';
import { MPV_IPC_COMMAND_TIMEOUT_MS } from './mpv-ipc-command';
import { MpvSocketMock } from './mpv-ipc-socket.test-helpers';
import {
openMpvPlayer,
setMpvReuseInstance,
shutdownMpvSession,
} from './mpv-session.service';
import {
spawnMock,
createMockChildProcess,
mockMpvSockets,
nextTick,
launchReusableSession,
mockUnansweredSockets,
mockSocketsQuittingInto,
commandsOf,
writtenRequestId,
} from './mpv-session.test-helpers';
describe('MPV launch queue', () => {
beforeEach(() => {
jest.clearAllMocks();
shutdownMpvSession();
});
afterEach(() => {
jest.useRealTimers();
shutdownMpvSession();
});
it('runs a quick second launch only after the first one settles', async () => {
const { proc } = await launchReusableSession();
const sockets = mockUnansweredSockets();
const commandOf = (socket: MpvSocketMock) =>
(
JSON.parse(String(socket.write.mock.calls[0][0])) as {
command: string[];
}
).command[0];
const answer = (socket: MpvSocketMock, id: number) =>
socket.emit(
'data',
Buffer.from(
JSON.stringify({ request_id: id, error: 'success' }) + '\n'
)
);
const first = openMpvPlayer({
title: 'Channel A',
url: 'https://example.com/a.m3u8',
});
const firstId = await writtenRequestId(sockets[0]);
const second = openMpvPlayer({
title: 'Channel B',
url: 'https://example.com/b.m3u8',
userAgent: 'IPTVnator test agent',
});
await nextTick();
await nextTick();
// The second launch waits in the queue: nothing else reached mpv.
expect(sockets).toHaveLength(1);
answer(sockets[0], firstId);
await expect(first).resolves.toMatchObject({ status: 'opened' });
answer(sockets[1], await writtenRequestId(sockets[1]));
while (sockets.length < 3) await nextTick();
answer(sockets[2], await writtenRequestId(sockets[2]));
await expect(second).resolves.toMatchObject({ status: 'opened' });
expect(sockets.map(commandOf)).toEqual([
'loadfile',
'set_property',
'loadfile',
]);
expect(proc.kill).not.toHaveBeenCalled();
expect(spawnMock).toHaveBeenCalledTimes(1);
});
it('lets Stop on a queued launch cancel it and stop the in-flight player', async () => {
const { proc } = await launchReusableSession();
const sockets = mockSocketsQuittingInto(proc);
const first = openMpvPlayer({
title: 'Channel A',
url: 'https://example.com/a.m3u8',
});
await writtenRequestId(sockets[0]);
const second = openMpvPlayer({
title: 'Channel B',
url: 'https://example.com/b.m3u8',
});
// The queued launch is already the session the dock shows and stops.
const queuedId = externalPlayerSessions.getActiveSessionId() as string;
expect(externalPlayerSessions.getSession(queuedId)).toMatchObject({
title: 'Channel B',
status: 'launching',
canClose: true,
});
// Stop goes through the in-flight launch's own closer: quit, exit.
await expect(
externalPlayerSessions.closeSession(queuedId)
).resolves.toMatchObject({ id: queuedId, status: 'closed' });
expect(proc.exitCode).toBe(0);
// The quit player's socket closes before any reply.
sockets[0].emit('close');
await expect(first).resolves.toMatchObject({ status: 'closed' });
await expect(second).resolves.toMatchObject({
id: queuedId,
status: 'closed',
});
// Nothing started playing again after Stop.
expect(commandsOf(sockets)).toEqual(['loadfile', 'quit']);
expect(spawnMock).toHaveBeenCalledTimes(1);
});
it('lets Stop on the newest queued launch cancel every launch queued before it', async () => {
const { proc } = await launchReusableSession();
const sockets = mockSocketsQuittingInto(proc);
const first = openMpvPlayer({
title: 'Channel A',
url: 'https://example.com/a.m3u8',
});
await writtenRequestId(sockets[0]);
const second = openMpvPlayer({
title: 'Channel B',
url: 'https://example.com/b.m3u8',
});
const third = openMpvPlayer({
title: 'Channel C',
url: 'https://example.com/c.m3u8',
});
const newestId = externalPlayerSessions.getActiveSessionId() as string;
await expect(
externalPlayerSessions.closeSession(newestId)
).resolves.toMatchObject({ status: 'closed' });
sockets[0].emit('close');
await expect(first).resolves.toMatchObject({ status: 'closed' });
await expect(second).resolves.toMatchObject({ status: 'closed' });
await expect(third).resolves.toMatchObject({ status: 'closed' });
expect(proc.exitCode).toBe(0);
expect(commandsOf(sockets)).toEqual(['loadfile', 'quit']);
expect(spawnMock).toHaveBeenCalledTimes(1);
});
it('hands the dock back to the live player when a queued launch is blocked by its teardown', async () => {
const { proc } = await launchReusableSession();
// The first launch's loadfile fails and its teardown never sees the
// child exit, so the queued launch finds the teardown gate closed.
(proc.kill as jest.Mock).mockImplementation(() => true);
mockMpvSockets({
reply: (command) =>
command[0] === 'loadfile' ? 'error running command' : 'success',
});
const errorSpy = jest
.spyOn(console, 'error')
.mockImplementation(() => undefined);
jest.useFakeTimers();
try {
const first = openMpvPlayer({
title: 'Channel A',
url: 'https://example.com/a.m3u8',
});
const firstId =
externalPlayerSessions.getActiveSessionId() as string;
const firstRejection = expect(first).rejects.toThrow(
'External player process did not exit'
);
const second = openMpvPlayer({
title: 'Channel B',
url: 'https://example.com/b.m3u8',
});
const secondId =
externalPlayerSessions.getActiveSessionId() as string;
const secondRejection = expect(second).rejects.toThrow(
'previous external player is still shutting down'
);
await jest.advanceTimersByTimeAsync(6_000);
await firstRejection;
await secondRejection;
// The session owning the still-running child is active again
// and can be stopped; the blocked launch is a terminal error.
expect(externalPlayerSessions.getActiveSessionId()).toBe(firstId);
expect(externalPlayerSessions.getSession(firstId)).toMatchObject({
status: 'error',
canClose: true,
});
expect(externalPlayerSessions.getSession(secondId)).toMatchObject({
status: 'error',
canClose: false,
});
expect(spawnMock).toHaveBeenCalledTimes(1);
} finally {
errorSpy.mockRestore();
Object.defineProperty(proc, 'exitCode', { value: 0 });
proc.emit('exit', 0);
}
});
it('runs only the newest queued launch once the in-flight one settles', async () => {
await launchReusableSession();
const sockets = mockUnansweredSockets();
const answer = (socket: MpvSocketMock, id: number) =>
socket.emit(
'data',
Buffer.from(
JSON.stringify({ request_id: id, error: 'success' }) + '\n'
)
);
const first = openMpvPlayer({
title: 'Channel A',
url: 'https://example.com/a.m3u8',
});
const firstId = await writtenRequestId(sockets[0]);
const skipped = openMpvPlayer({
title: 'Channel B',
url: 'https://example.com/b.m3u8',
});
const newest = openMpvPlayer({
title: 'Channel C',
url: 'https://example.com/c.m3u8',
});
const newestId = externalPlayerSessions.getActiveSessionId();
answer(sockets[0], firstId);
await expect(first).resolves.toMatchObject({ status: 'opened' });
// The skipped launch never reached mpv nor took the dock.
await expect(skipped).resolves.toMatchObject({ status: 'closed' });
expect(externalPlayerSessions.getActiveSessionId()).toBe(newestId);
answer(sockets[1], await writtenRequestId(sockets[1]));
await expect(newest).resolves.toMatchObject({
id: newestId,
status: 'opened',
});
expect(sockets).toHaveLength(2);
expect(spawnMock).toHaveBeenCalledTimes(1);
});
it('keeps a queued launch stoppable when its Stop cannot confirm the exit', async () => {
const { proc } = await launchReusableSession();
(proc.kill as jest.Mock).mockImplementation(() => true);
mockUnansweredSockets();
jest.useFakeTimers();
try {
const first = openMpvPlayer({
title: 'Channel A',
url: 'https://example.com/a.m3u8',
});
const queued = openMpvPlayer({
title: 'Channel B',
url: 'https://example.com/b.m3u8',
});
const queuedId =
externalPlayerSessions.getActiveSessionId() as string;
await jest.advanceTimersByTimeAsync(0);
// Stop goes through the in-flight launch's closer: its quit gets
// no reply, and the bounded teardown cannot confirm the exit.
const closing = externalPlayerSessions.closeSession(queuedId);
const closeRejection = expect(closing).rejects.toThrow(
'External player process did not exit'
);
const firstRejection = expect(first).rejects.toThrow(
'External player process did not exit'
);
await jest.advanceTimersByTimeAsync(10_000);
await closeRejection;
await firstRejection;
await expect(queued).resolves.toMatchObject({
id: queuedId,
status: 'error',
canClose: true,
});
// A retried Stop reaches the still-running child again.
const killsBeforeRetry = (proc.kill as jest.Mock).mock.calls.length;
const retry = externalPlayerSessions.closeSession(queuedId);
await jest.advanceTimersByTimeAsync(
MPV_IPC_COMMAND_TIMEOUT_MS + 100
);
expect((proc.kill as jest.Mock).mock.calls.length).toBeGreaterThan(
killsBeforeRetry
);
Object.defineProperty(proc, 'exitCode', { value: 0 });
proc.emit('exit', 0);
await expect(retry).resolves.toMatchObject({
id: queuedId,
status: 'closed',
});
expect(spawnMock).toHaveBeenCalledTimes(1);
} finally {
if (proc.exitCode === null) {
Object.defineProperty(proc, 'exitCode', { value: 0 });
proc.emit('exit', 0);
}
}
});
it('lets a late Stop for an older queued launch cancel only that launch', async () => {
const { proc } = await launchReusableSession();
const sockets = mockUnansweredSockets();
const answer = (socket: MpvSocketMock, id: number) =>
socket.emit(
'data',
Buffer.from(
JSON.stringify({ request_id: id, error: 'success' }) + '\n'
)
);
const first = openMpvPlayer({
title: 'Channel A',
url: 'https://example.com/a.m3u8',
});
const firstId = await writtenRequestId(sockets[0]);
const older = openMpvPlayer({
title: 'Channel B',
url: 'https://example.com/b.m3u8',
});
const olderId = externalPlayerSessions.getActiveSessionId() as string;
const newest = openMpvPlayer({
title: 'Channel C',
url: 'https://example.com/c.m3u8',
});
await expect(
externalPlayerSessions.closeSession(olderId)
).resolves.toMatchObject({ id: olderId, status: 'closed' });
expect(proc.kill).not.toHaveBeenCalled();
answer(sockets[0], firstId);
await expect(first).resolves.toMatchObject({ status: 'opened' });
await expect(older).resolves.toMatchObject({ status: 'closed' });
answer(sockets[1], await writtenRequestId(sockets[1]));
await expect(newest).resolves.toMatchObject({ status: 'opened' });
expect(commandsOf(sockets)).toEqual(['loadfile', 'loadfile']);
expect(proc.kill).not.toHaveBeenCalled();
});
it('settles a fresh launch and its queued Stop when the exit cannot be confirmed', async () => {
const proc = createMockChildProcess();
(proc.kill as jest.Mock).mockImplementation(() => {
Object.defineProperty(proc, 'killed', {
value: true,
configurable: true,
});
return true;
});
spawnMock.mockReturnValueOnce(proc);
(store.get as unknown as jest.Mock).mockImplementation(
(key: string, fallback?: unknown) =>
({
[MPV_PLAYER_PATH]: '/usr/bin/mpv',
[MPV_REUSE_INSTANCE]: true,
})[key] ?? fallback
);
jest.useFakeTimers();
try {
// The first launch spawns fresh; the second queues during its
// start-up confirmation window.
const first = openMpvPlayer({
title: 'Channel A',
url: 'https://example.com/a.m3u8',
});
const queued = openMpvPlayer({
title: 'Channel B',
url: 'https://example.com/b.m3u8',
});
const queuedId =
externalPlayerSessions.getActiveSessionId() as string;
// The Stop arrives over IPC, after the first launch spawned.
await jest.advanceTimersByTimeAsync(0);
let firstSettled = false;
void first.then(
() => (firstSettled = true),
() => (firstSettled = true)
);
const firstRejection = expect(first).rejects.toThrow(
'External player process did not exit'
);
const closeRejection = expect(
externalPlayerSessions.closeSession(queuedId)
).rejects.toThrow('External player process did not exit');
await jest.advanceTimersByTimeAsync(6_000);
expect(firstSettled).toBe(true);
await firstRejection;
await closeRejection;
await expect(queued).resolves.toMatchObject({
id: queuedId,
status: 'error',
canClose: true,
});
expect(spawnMock).toHaveBeenCalledTimes(1);
} finally {
Object.defineProperty(proc, 'exitCode', { value: 0 });
proc.emit('exit', 0);
}
});
it('never lets a retried queued Stop reach a player started later', async () => {
const { proc } = await launchReusableSession();
(proc.kill as jest.Mock).mockImplementation(() => true);
const sockets = mockUnansweredSockets();
jest.useFakeTimers();
const later = createMockChildProcess();
try {
const first = openMpvPlayer({
title: 'Channel A',
url: 'https://example.com/a.m3u8',
});
const queued = openMpvPlayer({
title: 'Channel B',
url: 'https://example.com/b.m3u8',
});
const queuedId =
externalPlayerSessions.getActiveSessionId() as string;
// The Stop arrives after the first launch wrote its loadfile.
while (!sockets[0]?.write.mock.calls.length) {
await jest.advanceTimersByTimeAsync(0);
}
const firstRejection = expect(first).rejects.toThrow(
'External player process did not exit'
);
const closeRejection = expect(
externalPlayerSessions.closeSession(queuedId)
).rejects.toThrow('External player process did not exit');
await jest.advanceTimersByTimeAsync(10_000);
await firstRejection;
await closeRejection;
await expect(queued).resolves.toMatchObject({ status: 'error' });
// The stuck child exits on its own and a new stream starts.
Object.defineProperty(proc, 'exitCode', { value: 0 });
proc.emit('exit', 0);
spawnMock.mockReturnValueOnce(later);
const next = openMpvPlayer({
title: 'Channel C',
url: 'https://example.com/c.m3u8',
});
await jest.advanceTimersByTimeAsync(100);
await expect(next).resolves.toMatchObject({ status: 'opened' });
// A retried Stop for the old queued launch stays on its own
// targets and leaves the new player alone.
const retry = externalPlayerSessions.closeSession(queuedId);
await jest.advanceTimersByTimeAsync(
MPV_IPC_COMMAND_TIMEOUT_MS + 100
);
await expect(retry).resolves.toMatchObject({ status: 'closed' });
expect(later.kill).not.toHaveBeenCalled();
} finally {
if (proc.exitCode === null) {
Object.defineProperty(proc, 'exitCode', { value: 0 });
proc.emit('exit', 0);
}
}
});
it('restores the last started launch when a queued one is blocked by its teardown', async () => {
const proc = createMockChildProcess();
(proc.kill as jest.Mock).mockImplementation(() => {
Object.defineProperty(proc, 'killed', {
value: true,
configurable: true,
});
return true;
});
spawnMock.mockReturnValueOnce(proc);
// Reuse off: the detached child is never stored for reuse.
(store.get as unknown as jest.Mock).mockImplementation(
(key: string, fallback?: unknown) =>
({
[MPV_PLAYER_PATH]: '/usr/bin/mpv',
[MPV_REUSE_INSTANCE]: false,
})[key] ?? fallback
);
jest.useFakeTimers();
try {
const first = openMpvPlayer({
title: 'Channel A',
url: 'https://example.com/a.m3u8',
});
const firstId =
externalPlayerSessions.getActiveSessionId() as string;
const skipped = openMpvPlayer({
title: 'Channel B',
url: 'https://example.com/b.m3u8',
});
const newest = openMpvPlayer({
title: 'Channel C',
url: 'https://example.com/c.m3u8',
});
const newestId =
externalPlayerSessions.getActiveSessionId() as string;
const firstRejection = expect(first).rejects.toThrow(
'External player process did not exit'
);
const newestRejection = expect(newest).rejects.toThrow(
'previous external player is still shutting down'
);
// A delayed Stop for the first launch cannot confirm its exit.
const closeRejection = expect(
externalPlayerSessions.closeSession(firstId)
).rejects.toThrow('External player process did not exit');
await jest.advanceTimersByTimeAsync(6_000);
await closeRejection;
await firstRejection;
await expect(skipped).resolves.toMatchObject({ status: 'closed' });
await newestRejection;
// The dock goes back to the session owning the stuck child.
expect(externalPlayerSessions.getActiveSessionId()).toBe(firstId);
expect(externalPlayerSessions.getSession(firstId)).toMatchObject({
status: 'error',
canClose: true,
});
expect(externalPlayerSessions.getSession(newestId)).toMatchObject({
status: 'error',
canClose: false,
});
expect(spawnMock).toHaveBeenCalledTimes(1);
} finally {
Object.defineProperty(proc, 'exitCode', { value: 0 });
proc.emit('exit', 0);
}
});
it('lets a queued Stop reach the child a failed launch is tearing down', async () => {
const { proc } = await launchReusableSession();
(proc.kill as jest.Mock).mockImplementation(() => true);
// The first launch's loadfile fails; its teardown then waits for an
// exit that never comes.
mockMpvSockets({
reply: (command) =>
command[0] === 'loadfile' ? 'error running command' : null,
});
const errorSpy = jest
.spyOn(console, 'error')
.mockImplementation(() => undefined);
jest.useFakeTimers();
try {
const first = openMpvPlayer({
title: 'Channel A',
url: 'https://example.com/a.m3u8',
});
const firstRejection = expect(first).rejects.toThrow(
'External player process did not exit'
);
const queued = openMpvPlayer({
title: 'Channel B',
url: 'https://example.com/b.m3u8',
});
const queuedId =
externalPlayerSessions.getActiveSessionId() as string;
for (
let tick = 0;
tick < 50 && !(proc.kill as jest.Mock).mock.calls.length;
tick += 1
) {
await jest.advanceTimersByTimeAsync(10);
}
expect(proc.kill).toHaveBeenCalled();
// Stop during that teardown must not report success while the
// child is still alive.
const closeRejection = expect(
externalPlayerSessions.closeSession(queuedId)
).rejects.toThrow('External player process did not exit');
await jest.advanceTimersByTimeAsync(10_000);
await closeRejection;
await firstRejection;
await expect(queued).resolves.toMatchObject({
id: queuedId,
status: 'error',
canClose: true,
});
// A retry reaches the same child and settles once it exits; the
// retried quit gets no reply within its bound.
const retry = externalPlayerSessions.closeSession(queuedId);
await jest.advanceTimersByTimeAsync(0);
Object.defineProperty(proc, 'exitCode', { value: 0 });
proc.emit('exit', 0);
await jest.advanceTimersByTimeAsync(
MPV_IPC_COMMAND_TIMEOUT_MS + 100
);
await expect(retry).resolves.toMatchObject({
id: queuedId,
status: 'closed',
});
expect(spawnMock).toHaveBeenCalledTimes(1);
} finally {
errorSpy.mockRestore();
if (proc.exitCode === null) {
Object.defineProperty(proc, 'exitCode', { value: 0 });
proc.emit('exit', 0);
}
}
});
it('keeps a queued Stop working after the previous owner closed during a failed handback', async () => {
const { proc, session: previous } = await launchReusableSession();
(proc.kill as jest.Mock).mockImplementation(() => true);
const sockets = mockUnansweredSockets();
const errorSpy = jest
.spyOn(console, 'error')
.mockImplementation(() => undefined);
jest.useFakeTimers();
try {
const first = openMpvPlayer({
title: 'Channel A',
url: 'https://example.com/a.m3u8',
});
const firstRejection = expect(first).rejects.toThrow(
'External player process did not exit'
);
const loadfileId = await (async () => {
while (!sockets[0]?.write.mock.calls.length) {
await jest.advanceTimersByTimeAsync(0);
}
return (
JSON.parse(String(sockets[0].write.mock.calls[0][0])) as {
request_id: number;
}
).request_id;
})();
// A late Stop for the previous session closes it; the child now
// belongs to the first launch.
await expect(
externalPlayerSessions.closeSession(previous.id)
).resolves.toMatchObject({ status: 'closed' });
const queued = openMpvPlayer({
title: 'Channel B',
url: 'https://example.com/b.m3u8',
});
const queuedId =
externalPlayerSessions.getActiveSessionId() as string;
// mpv rejects the load: the child is handed back to the closed
// previous session while the first launch tears it down.
sockets[0].emit(
'data',
Buffer.from(
JSON.stringify({
request_id: loadfileId,
error: 'error running command',
}) + '\n'
)
);
await jest.advanceTimersByTimeAsync(10);
const closeRejection = expect(
externalPlayerSessions.closeSession(queuedId)
).rejects.toThrow('External player process did not exit');
await jest.advanceTimersByTimeAsync(10_000);
await closeRejection;
await firstRejection;
await expect(queued).resolves.toMatchObject({
id: queuedId,
status: 'error',
canClose: true,
});
expect(spawnMock).toHaveBeenCalledTimes(1);
} finally {
errorSpy.mockRestore();
if (proc.exitCode === null) {
Object.defineProperty(proc, 'exitCode', { value: 0 });
proc.emit('exit', 0);
}
}
});
it('keeps a queued Stop working when reuse is disabled before the in-flight load', async () => {
const { proc } = await launchReusableSession();
(proc.kill as jest.Mock).mockImplementation(() => true);
const sockets = mockUnansweredSockets();
const errorSpy = jest
.spyOn(console, 'error')
.mockImplementation(() => undefined);
jest.useFakeTimers();
try {
// The first launch still waits on its user-agent reply, so the
// child still belongs to the session before it.
const first = openMpvPlayer({
title: 'Channel A',
url: 'https://example.com/a.m3u8',
userAgent: 'IPTVnator test agent',
});
const firstRejection = expect(first).rejects.toThrow(
'External player process did not exit'
);
while (!sockets[0]?.write.mock.calls.length) {
await jest.advanceTimersByTimeAsync(0);
}
const queued = openMpvPlayer({
title: 'Channel B',
url: 'https://example.com/b.m3u8',
});
const queuedId =
externalPlayerSessions.getActiveSessionId() as string;
// Disabling reuse clears the stored child before it has exited.
setMpvReuseInstance(false);
const closeRejection = expect(
externalPlayerSessions.closeSession(queuedId)
).rejects.toThrow('External player process did not exit');
await jest.advanceTimersByTimeAsync(10_000);
await closeRejection;
await firstRejection;
await expect(queued).resolves.toMatchObject({
id: queuedId,
status: 'error',
canClose: true,
});
// The first launch has settled; a retried Stop must still reach
// the live child and settle only once it exits.
const killsBeforeRetry = (proc.kill as jest.Mock).mock.calls.length;
const retry = externalPlayerSessions.closeSession(queuedId);
let retried = false;
void retry.then(
() => (retried = true),
() => (retried = true)
);
await jest.advanceTimersByTimeAsync(
MPV_IPC_COMMAND_TIMEOUT_MS + 100
);
expect(retried).toBe(false);
expect((proc.kill as jest.Mock).mock.calls.length).toBeGreaterThan(
killsBeforeRetry
);
Object.defineProperty(proc, 'exitCode', { value: 0 });
proc.emit('exit', 0);
await expect(retry).resolves.toMatchObject({
id: queuedId,
status: 'closed',
});
expect(spawnMock).toHaveBeenCalledTimes(1);
} finally {
errorSpy.mockRestore();
if (proc.exitCode === null) {
Object.defineProperty(proc, 'exitCode', { value: 0 });
proc.emit('exit', 0);
}
}
});
it('starts a queued launch on the fresh player a failed launch spawned', async () => {
const { proc } = await launchReusableSession();
const freshProc = createMockChildProcess();
spawnMock.mockReturnValueOnce(freshProc);
// The first launch's loadfile fails; every later command succeeds.
// Each loadfile records how many players had been spawned by then.
let rejectedLoadfile = false;
const spawnsAtLoadfile: number[] = [];
mockMpvSockets({
reply: (command) => {
if (command[0] === 'loadfile') {
spawnsAtLoadfile.push(spawnMock.mock.calls.length);
}
if (command[0] === 'loadfile' && !rejectedLoadfile) {
rejectedLoadfile = true;
return 'error running command';
}
return 'success';
},
});
const errorSpy = jest
.spyOn(console, 'error')
.mockImplementation(() => undefined);
try {
const first = openMpvPlayer({
title: 'Channel A',
url: 'https://example.com/a.m3u8',
});
const second = openMpvPlayer({
title: 'Channel B',
url: 'https://example.com/b.m3u8',
});
await expect(first).resolves.toMatchObject({ status: 'opened' });
const secondSession = await second;
// The second launch loaded into the replacement the first one's
// fallback spawned, not into the child that fallback tore down.
expect(spawnsAtLoadfile).toEqual([1, 2]);
expect(
externalPlayerSessions.getSession(secondSession.id)?.status
).toBe('opened');
expect(proc.kill).toHaveBeenCalledTimes(1);
expect(spawnMock).toHaveBeenCalledTimes(2);
expect(freshProc.kill).not.toHaveBeenCalled();
} finally {
errorSpy.mockRestore();
}
});
});
@@ -0,0 +1,182 @@
import {
ExternalPlayerSession,
readExternalPlayerErrorCode,
} from '@iptvnator/shared/interfaces';
import { externalPlayerProcessTeardownGate } from './external-player-process';
import { externalPlayerSessions } from './external-player-runtime';
/** What a launch knows about its place in the queue when it starts. */
export interface QueuedMpvLaunch {
/** Whether this launch waited behind another one. */
queued: boolean;
/**
* Whether Stop or shutdown cancelled this launch while it waited, or a
* newer launch is queued behind it.
*/
cancelled: () => boolean;
/** This launch's own queued Stop, if one was requested. */
pendingStop: () => Promise<void> | null;
/** Called once the launch really starts acting on a player. */
markInFlight: () => void;
/** The last launch that started before this one runs. */
lastStartedSessionId: () => string | null;
}
/**
* Runs MPV launches one at a time. A reused launch waits for mpv's replies
* (bounded per command) and a failed one for its teardown and fallback
* spawn, so overlapping launches would act on the same child. Queued, each
* launch starts from the player the previous one left behind, and only the
* newest queued launch runs. An idle queue starts a launch synchronously.
*
* Until a queued launch starts, its Stop cancels it and every launch queued
* before it. Stop on the newest one also closes, through their own closers,
* the in-flight launch and the session owning the stored player, so the
* exact-child, retry and settle rules of those closers apply.
*/
export class MpvLaunchQueue {
private pending: Promise<unknown> | null = null;
private latest = 0;
private cancelledThrough = 0;
private inFlightSessionId: string | null = null;
// The last launch that actually started. With reuse off its detached
// child is not stored, so it is the only handle on that player once the
// launch settled with an unconfirmed teardown.
private lastStartedSessionId: string | null = null;
constructor(private readonly storedOwnerSessionId: () => string | null) {}
/** Cancels every launch queued so far, for app shutdown. */
cancelAll(): void {
this.cancelledThrough = this.latest;
}
enqueue(
session: ExternalPlayerSession,
launch: (queue: QueuedMpvLaunch) => Promise<ExternalPlayerSession>
): Promise<ExternalPlayerSession> {
const previous = this.pending;
const number = ++this.latest;
let stop: Promise<void> | null = null;
// Fixed at the first Stop, so a retry never reaches a newer player.
let stopTargets: string[] | null = null;
if (previous) {
externalPlayerSessions.attachCloser(session.id, () => {
this.cancelledThrough = Math.max(this.cancelledThrough, number);
// A late Stop for an older queued launch only cancels it.
stopTargets ??=
number === this.latest ? this.stopTargets(session.id) : [];
stop = Promise.all(
stopTargets.map((id) =>
externalPlayerSessions.closeSession(id)
)
).then(() => undefined);
return stop;
});
}
const queue: QueuedMpvLaunch = {
queued: previous !== null,
cancelled: () =>
number <= this.cancelledThrough || number !== this.latest,
pendingStop: () => stop,
markInFlight: () => {
this.inFlightSessionId = session.id;
this.lastStartedSessionId = session.id;
},
lastStartedSessionId: () => this.lastStartedSessionId,
};
const run = () => launch(queue);
// The queue is released before the caller sees the result, so a
// launch made right after this one settles starts synchronously.
let settled: Promise<unknown> | null = null;
const result = (previous ? previous.then(run) : run()).finally(() => {
if (this.inFlightSessionId === session.id) {
this.inFlightSessionId = null;
}
if (this.pending === settled) this.pending = null;
});
settled = result.catch(() => undefined);
this.pending = settled;
return result;
}
private stopTargets(sessionId: string): string[] {
const targets: string[] = [];
for (const id of [
this.inFlightSessionId,
this.storedOwnerSessionId(),
this.lastStartedSessionId,
]) {
// Only a session with its own closer can confirm its teardown;
// closing one without would just mark it closed.
if (
id &&
id !== sessionId &&
!targets.includes(id) &&
externalPlayerSessions.getSession(id)?.canClose
) {
targets.push(id);
}
}
return targets;
}
}
/**
* Settles a queued launch that will not start. When its own Stop is still
* closing the in-flight player, the outcome waits for it: a Stop that
* cannot confirm the exit leaves a closable error, so Stop can be retried.
*/
export async function settleCancelledQueuedLaunch(
session: ExternalPlayerSession,
pendingStop: Promise<void> | null
): Promise<ExternalPlayerSession> {
if (pendingStop) {
try {
await pendingStop;
} catch (error) {
return (
externalPlayerSessions.markError(
session.id,
error instanceof Error ? error.message : String(error),
{ canClose: true }
) ?? session
);
}
}
return externalPlayerSessions.markClosed(session.id) ?? session;
}
/**
* A queued launch starts after an earlier one may have left a child whose
* exit is unconfirmed. Then the dock goes back to the session that still
* owns that child; failing that, this session keeps its queued closer so
* Stop can still reach the child.
*/
export function assertQueuedLaunchAllowed(
session: ExternalPlayerSession,
restoreCandidates: Array<string | null>
): void {
try {
externalPlayerProcessTeardownGate.assertLaunchAllowed();
} catch (error) {
let restored: ExternalPlayerSession | null = null;
for (const id of restoreCandidates) {
if (!id || id === session.id) continue;
restored = externalPlayerSessions.restoreActiveSession(
id,
session.id
);
if (restored) break;
}
externalPlayerSessions.markError(
session.id,
error instanceof Error ? error.message : String(error),
{
code: readExternalPlayerErrorCode(error) ?? undefined,
canClose: !restored,
}
);
throw error;
}
}
@@ -0,0 +1,69 @@
import type { ChildProcess } from 'child_process';
import {
sendPlayerErrorNotification,
traceExternalPlayer,
} from './external-player-runtime';
function shouldIgnoreMpvStdoutLine(line: string): boolean {
return /^(\(Paused\)\s*)?AV:\s/.test(line);
}
/**
* Reports what a reusable MPV child prints: stream errors on stdout or stderr
* reach the renderer as player errors; the periodic status line is ignored.
*/
export function attachMpvOutputHandlers(proc: ChildProcess): void {
if (proc.stdout) {
proc.stdout.on('data', (data) => {
const lines = data
.toString()
.split('\n')
.map((line) => line.trim())
.filter(Boolean);
for (const output of lines) {
if (shouldIgnoreMpvStdoutLine(output)) {
continue;
}
traceExternalPlayer('mpv stdout', { output });
if (
output.includes('Failed to open') ||
output.includes('Error opening') ||
output.includes('Protocol not found') ||
output.includes('Connection refused') ||
output.includes('error') ||
output.includes('403') ||
output.includes('404')
) {
console.error('[MPV ERROR from stdout]:', output);
sendPlayerErrorNotification('MPV', output);
}
}
});
}
if (proc.stderr) {
proc.stderr.on('data', (data) => {
const output = data.toString().trim();
if (output) {
console.error('[MPV stderr]:', output);
if (
output.includes('Failed to open') ||
output.includes('Exiting... (Errors when loading file)') ||
output.includes('Error opening') ||
output.includes('Protocol not found') ||
output.includes('Connection refused') ||
output.includes('error') ||
output.includes('403') ||
output.includes('404')
) {
console.error('[MPV ERROR]:', output);
sendPlayerErrorNotification('MPV', output);
}
}
});
}
}
@@ -0,0 +1,121 @@
import { createConnection } from 'net';
import { PlayerContentInfo } from '@iptvnator/shared/interfaces';
import { sendPlaybackPositionUpdate } from './external-player-runtime';
/**
* Playback-position polling of an external MPV over its IPC socket. One poll
* runs at a time across launches: starting a new one stops the previous one.
*/
let positionPollingInterval: NodeJS.Timeout | null = null;
// The first poll waits for MPV to open its IPC socket. The handle is kept so
// a player that exits during that wait does not start an orphaned interval.
let positionPollingDelay: NodeJS.Timeout | null = null;
// Bumped on every stop. A poll that was already awaiting an IPC reply when
// polling stopped (for example when a reused player loads the next stream)
// must not report what it reads under the previous content.
let positionPollingGeneration = 0;
async function getMpvProperty(
socketPath: string,
property: string
): Promise<number | null> {
return new Promise((resolve) => {
const client = createConnection(socketPath);
const request =
JSON.stringify({
command: ['get_property', property],
}) + '\n';
let data = '';
client.on('connect', () => {
client.write(request);
});
const timeoutHandle = setTimeout(() => {
client.destroy();
resolve(null);
}, 2000);
client.on('data', (chunk) => {
data += chunk.toString();
if (data.includes('\n')) {
clearTimeout(timeoutHandle);
try {
const lines = data.split('\n');
for (const line of lines) {
if (!line.trim()) continue;
const response = JSON.parse(line);
if (response.data !== undefined) {
client.destroy();
resolve(response.data);
return;
}
}
} catch {
// Ignore partial JSON while MPV is still writing.
}
}
});
client.on('end', () => {
clearTimeout(timeoutHandle);
if (data && !data.includes('\n')) {
try {
const response = JSON.parse(data);
resolve(response.data ?? null);
} catch {
resolve(null);
}
}
});
client.on('error', () => {
clearTimeout(timeoutHandle);
resolve(null);
});
});
}
export function stopPositionPolling(): void {
positionPollingGeneration += 1;
if (positionPollingDelay) {
clearTimeout(positionPollingDelay);
positionPollingDelay = null;
}
if (positionPollingInterval) {
clearInterval(positionPollingInterval);
positionPollingInterval = null;
}
}
export function startPositionPolling(
socketPath: string,
contentInfo: PlayerContentInfo,
sessionId: string
): void {
stopPositionPolling();
const generation = positionPollingGeneration;
const isCurrent = () => generation === positionPollingGeneration;
positionPollingDelay = setTimeout(() => {
positionPollingDelay = null;
positionPollingInterval = setInterval(async () => {
try {
const position = await getMpvProperty(socketPath, 'time-pos');
if (!isCurrent()) return;
const duration = await getMpvProperty(socketPath, 'duration');
if (!isCurrent()) return;
if (position !== null) {
sendPlaybackPositionUpdate(sessionId, contentInfo, {
positionSeconds: Math.floor(position),
durationSeconds: duration ? Math.floor(duration) : null,
});
}
} catch {
stopPositionPolling();
}
}, 5000);
}, 2000);
}
@@ -1,20 +1,22 @@
import { ChildProcess } from 'child_process';
import { createConnection } from 'net';
import {
ExternalPlayerSession,
PlayerContentInfo,
} from '@iptvnator/shared/interfaces';
import { redactSensitiveData } from '@iptvnator/shared/logging';
import { joinMpvHeaderFields } from '../util/mpv-string-list.util';
import { externalPlayerProcessTeardownGate } from './external-player-process';
import {
externalPlayerSessions,
traceExternalPlayer,
} from './external-player-runtime';
import { sendMpvCommand } from './mpv-ipc-command';
/**
* The per-file start of a reuse load: the requested offset, else zero. A
* `seek` right after `loadfile` runs before the file is loaded and fails,
* so the offset travels with the load itself.
* `seek` right after `loadfile` runs before the file is loaded and mpv
* rejects it, so the offset travels with the load itself and no seek is
* sent on the reuse path.
*/
function reuseStartSeconds(startTime: number | undefined): number {
return typeof startTime === 'number' &&
@@ -24,7 +26,9 @@ function reuseStartSeconds(startTime: number | undefined): number {
: 0;
}
const MPV_IPC_COMMAND_TIMEOUT_MS = 2_000;
function hasExited(child: ChildProcess): boolean {
return child.exitCode !== null || child.signalCode !== null;
}
export interface MpvReuseAttemptState {
contentMutated: boolean;
@@ -50,68 +54,33 @@ interface MpvReuseOptions {
stopPositionPolling: () => void;
}
function sendMpvCommand(
socketPath: string,
command: string,
args: Array<string | number>,
shouldDispatch?: () => boolean
): Promise<boolean> {
return new Promise((resolve, reject) => {
const client = createConnection(socketPath);
const request = JSON.stringify({ command: [command, ...args] }) + '\n';
let settled = false;
let timeoutHandle: NodeJS.Timeout | null = null;
const complete = (error?: Error, dispatched = true) => {
if (settled) return;
settled = true;
if (timeoutHandle) clearTimeout(timeoutHandle);
if (!dispatched && !client.destroyed) client.destroy();
if (error) {
reject(error);
} else {
resolve(dispatched);
}
};
client.on('connect', () => {
if (shouldDispatch && !shouldDispatch()) {
complete(undefined, false);
return;
}
traceExternalPlayer('mpv ipc command', {
command,
argsCount: args.length,
});
try {
client.write(request);
client.end();
complete();
} catch (error) {
complete(
error instanceof Error ? error : new Error(String(error))
);
}
});
client.on('error', (error) => complete(error));
timeoutHandle = setTimeout(() => {
complete(new Error('MPV IPC command timed out'));
client.destroy();
}, MPV_IPC_COMMAND_TIMEOUT_MS);
timeoutHandle.unref();
});
}
/** Owns the one MPV child/socket retained when instance reuse is enabled. */
export class MpvReusableProcess {
private process: ChildProcess | null = null;
private socketPath: string | null = null;
private processSessionId: string | null = null;
private readonly processSessionIds = new WeakMap<ChildProcess, string>();
// Bumped whenever the stored player is stopped (reuse disabled or app
// quitting). Launches run one at a time, so the only thing that can
// overtake an in-flight reuse attempt is that stop, which cancels it.
private stopGeneration = 0;
currentSessionId(): string | null {
return this.processSessionId;
}
/**
* The session that owns the stored child right now. A failed reuse
* attempt hands the child back to its previous owner while it tears it
* down, so this reads the per-child mapping, not the last recorded id.
*/
storedOwnerSessionId(): string | null {
if (!this.process) return null;
return (
this.processSessionIds.get(this.process) ?? this.processSessionId
);
}
sessionIdFor(process: ChildProcess, fallback: string): string {
return this.processSessionIds.get(process) ?? fallback;
}
@@ -140,6 +109,10 @@ export class MpvReusableProcess {
stopPositionPolling: () => void,
guardFutureLaunches = false
): void {
// A launch still waiting for mpv's reply must not fall back to a
// fresh, again-reusable player once reuse is disabled or the app
// quits: cancel it so it settles its own session instead.
this.stopGeneration += 1;
const process = this.process;
if (!process || process.killed) return;
traceExternalPlayer(reason);
@@ -163,6 +136,29 @@ export class MpvReusableProcess {
traceExternalPlayer('reuse existing mpv instance');
const { session, state } = options;
const generation = this.stopGeneration;
const cancelled = () => generation !== this.stopGeneration;
// The stored-player stop is already tearing the child down; settle
// only once its exit is confirmed, else stay a closable error so Stop
// can retry while the teardown gate blocks further launches.
const settleCancelled = async () => {
traceExternalPlayer('mpv reuse cancelled by a stored-player stop');
try {
await externalPlayerProcessTeardownGate.terminate(
reusedProcess
);
} catch (error) {
state.teardownUnconfirmed = true;
return (
externalPlayerSessions.markError(
session.id,
error instanceof Error ? error.message : String(error),
{ canClose: true }
) ?? session
);
}
return externalPlayerSessions.markClosed(session.id) ?? session;
};
const reusedProcessSessionId =
this.processSessionIds.get(reusedProcess) ??
options.previousProcessSessionId;
@@ -170,7 +166,18 @@ export class MpvReusableProcess {
let retryableClose: Promise<void> | null = null;
let launchClose: Promise<void> | null = null;
// Launches run one at a time, so nothing else acts on this child while
// this attempt is in flight: its Stop always reaches the child then,
// whoever the per-child mapping names (the previous owner before
// loadfile or during a failed attempt's teardown, possibly closed
// already). Once the attempt settles, ownership decides as before.
let attemptInFlight = true;
// A Stop of this session that could not confirm the exit keeps the
// right to retry that exact child after the attempt settles.
let ownsFailedTeardown = false;
const ownsReusedProcess = () =>
attemptInFlight ||
ownsFailedTeardown ||
this.processSessionIds.get(reusedProcess) === session.id ||
externalPlayerSessions.getActiveSessionId() === session.id;
const closeReusedProcess = async () => {
@@ -208,6 +215,7 @@ export class MpvReusableProcess {
launchClose ??= closeAttempt;
void closeAttempt.catch((error) => {
if (retryableClose === closeAttempt) retryableClose = null;
ownsFailedTeardown = true;
state.teardownUnconfirmed = true;
externalPlayerSessions.markError(
session.id,
@@ -219,28 +227,34 @@ export class MpvReusableProcess {
return retryableClose;
});
try {
await this.applyReuseCommands(
options,
reusedSocketPath,
() => !closeRequested
);
if (closeRequested) return await finishRequestedClose();
// Once `loadfile` reaches the socket the player content may have
// changed, whatever mpv replies, so the attempted session owns the
// child from the write onwards: an exit or a stale Stop for the
// previous session during the reply wait then lands on the right
// session instead of terminating the new playback, and the previous
// session's position poll can no longer read the new stream's
// position under the old content.
const onContentDispatched = () => {
state.contentMutated = true;
this.processSessionId = session.id;
this.processSessionIds.set(reusedProcess, session.id);
options.stopPositionPolling();
};
if (options.startTime) {
await sendMpvCommand(
reusedSocketPath,
'seek',
[String(options.startTime), 'absolute'],
() => !closeRequested
);
}
try {
await this.applyReuseCommands(
options,
reusedSocketPath,
() => !closeRequested && !cancelled(),
onContentDispatched
);
if (closeRequested) return await finishRequestedClose();
if (cancelled()) return await settleCancelled();
if (hasExited(reusedProcess)) {
// The child exited while its reply was in flight; its exit
// handler has already settled the session that owned it.
return externalPlayerSessions.getSession(session.id) ?? session;
}
if (options.contentInfo) {
options.startPositionPolling(
@@ -256,7 +270,17 @@ export class MpvReusableProcess {
const current = externalPlayerSessions.getSession(session.id);
if (current?.status === 'closed') return current;
if (closeRequested) return await finishRequestedClose();
console.error('Failed to send command to existing MPV:', error);
if (cancelled()) return await settleCancelled();
if (current?.status === 'error' && hasExited(reusedProcess)) {
// The child died while its reply was in flight and its exit
// handler already reported the failure; a fresh launch under
// this terminal session could never be shown as opened.
return current;
}
console.error(
'Failed to send command to existing MPV:',
redactSensitiveData(error)
);
if (state.contentMutated) {
if (reusedProcessSessionId) {
@@ -282,21 +306,27 @@ export class MpvReusableProcess {
this.clear(reusedProcess);
options.stopPositionPolling();
if (closeRequested) return await finishRequestedClose();
// Reuse may have been disabled, or the app may be quitting,
// while the teardown above waited for the child to exit.
if (cancelled()) return await settleCancelled();
return null;
} finally {
attemptInFlight = false;
}
}
private async applyReuseCommands(
options: MpvReuseOptions,
socketPath: string,
shouldDispatch: () => boolean
shouldDispatch: () => boolean,
onContentDispatched: () => void
): Promise<void> {
if (options.effectiveUserAgent) {
const dispatched = await sendMpvCommand(
socketPath,
'set_property',
['user-agent', options.effectiveUserAgent],
shouldDispatch
{ shouldDispatch }
);
if (!dispatched) return;
}
@@ -305,7 +335,7 @@ export class MpvReusableProcess {
socketPath,
'set_property',
['referrer', options.effectiveReferer],
shouldDispatch
{ shouldDispatch }
);
if (!dispatched) return;
}
@@ -317,7 +347,7 @@ export class MpvReusableProcess {
'http-header-fields',
joinMpvHeaderFields(options.headerFields),
],
shouldDispatch
{ shouldDispatch }
);
if (!dispatched) return;
}
@@ -335,13 +365,13 @@ export class MpvReusableProcess {
-1,
fileOptions.join(','),
];
const dispatched = await sendMpvCommand(
const reply = await sendMpvCommand(
socketPath,
'loadfile',
loadFileArgs,
shouldDispatch
{ shouldDispatch, onDispatched: onContentDispatched }
);
if (!dispatched) return;
if (!reply) return;
traceExternalPlayer('loaded new url in existing mpv instance');
}
}
@@ -0,0 +1,663 @@
/**
* Reply handling of the reused-instance MPV IPC path. Every command carries a
* `request_id` and the launch waits, bounded, for mpv's matching reply; an
* error reply or a missing one falls back to a fresh launch exactly like a
* socket failure. Teardown, ownership and Stop-interleaving coverage lives in
* `mpv-session.service.spec.ts`.
*/
jest.mock('electron', () => ({
ipcMain: {
handle: jest.fn(),
},
}));
jest.mock('child_process', () => ({
spawn: jest.fn(),
}));
jest.mock('net', () => ({
createConnection: jest.fn(),
createServer: jest.fn(),
}));
jest.mock('../app', () => ({
__esModule: true,
default: {
mainWindow: null,
},
}));
jest.mock('../services/store.service', () => ({
MPV_PLAYER_ARGUMENTS: 'MPV_PLAYER_ARGUMENTS',
MPV_PLAYER_PATH: 'MPV_PLAYER_PATH',
MPV_REUSE_INSTANCE: 'MPV_REUSE_INSTANCE',
store: {
get: jest.fn(),
set: jest.fn(),
},
}));
import {
MPV_PLAYER_PATH,
MPV_REUSE_INSTANCE,
store,
} from '../services/store.service';
import * as externalPlayerRuntime from './external-player-runtime';
import { externalPlayerSessions } from './external-player-runtime';
import { MPV_IPC_COMMAND_TIMEOUT_MS } from './mpv-ipc-command';
import {
createMpvSocketMock,
MpvSocketMock,
} from './mpv-ipc-socket.test-helpers';
import {
openMpvPlayer,
setMpvReuseInstance,
shutdownMpvSession,
} from './mpv-session.service';
import {
spawnMock,
createConnectionMock,
createMockChildProcess,
mockMpvSockets,
nextTick,
launchReusableInstance,
launchReusableSession,
mockUnansweredSockets,
writtenRequestId,
} from './mpv-session.test-helpers';
describe('reused MPV IPC replies', () => {
beforeEach(() => {
jest.clearAllMocks();
shutdownMpvSession();
});
afterEach(() => {
jest.useRealTimers();
shutdownMpvSession();
});
it('opens the session once mpv acknowledges the content command', async () => {
const reusedProc = await launchReusableInstance();
const written: string[] = [];
mockMpvSockets({ written });
const session = await openMpvPlayer({
title: 'Second stream',
url: 'https://example.com/two.m3u8',
startTime: 30,
});
expect(session.status).toBe('opened');
expect(spawnMock).toHaveBeenCalledTimes(1);
expect(reusedProc.kill).not.toHaveBeenCalled();
const requests = written.map(
(chunk) =>
JSON.parse(chunk.trim()) as {
command: Array<string | number>;
request_id: number;
}
);
// The offset travels with loadfile; no seek races the load.
expect(requests.map((request) => request.command[0])).toEqual([
'loadfile',
]);
expect(Number.isInteger(requests[0].request_id)).toBe(true);
});
it('falls back to a fresh launch when mpv rejects the content command', async () => {
const reusedProc = await launchReusableInstance();
const freshProc = createMockChildProcess();
spawnMock.mockReturnValueOnce(freshProc);
mockMpvSockets({
reply: (command) =>
command[0] === 'loadfile' ? 'error running command' : 'success',
});
const errorSpy = jest
.spyOn(console, 'error')
.mockImplementation(() => undefined);
try {
const session = await openMpvPlayer({
title: 'Second stream',
url: 'https://example.com/two.m3u8?token=secret-token',
});
expect(reusedProc.kill).toHaveBeenCalledTimes(1);
expect(spawnMock).toHaveBeenCalledTimes(2);
expect(session.status).toBe('opened');
expect(externalPlayerSessions.getSession(session.id)?.status).toBe(
'opened'
);
const logged = JSON.stringify(errorSpy.mock.calls);
expect(logged).toContain('mpv loadfile: error running command');
expect(logged).not.toContain('secret-token');
} finally {
errorSpy.mockRestore();
}
});
it('falls back to a fresh launch when mpv never replies', async () => {
const reusedProc = await launchReusableInstance();
const freshProc = createMockChildProcess();
spawnMock.mockReturnValueOnce(freshProc);
mockMpvSockets({ reply: () => null });
const errorSpy = jest
.spyOn(console, 'error')
.mockImplementation(() => undefined);
jest.useFakeTimers();
try {
const opening = openMpvPlayer({
title: 'Second stream',
url: 'https://example.com/two.m3u8',
});
await jest.advanceTimersByTimeAsync(MPV_IPC_COMMAND_TIMEOUT_MS - 1);
expect(reusedProc.kill).not.toHaveBeenCalled();
expect(spawnMock).toHaveBeenCalledTimes(1);
await jest.advanceTimersByTimeAsync(1);
expect(reusedProc.kill).toHaveBeenCalledTimes(1);
// Exit of the reused child, the fresh spawn and its start
// confirmation all run on timers.
await jest.advanceTimersByTimeAsync(500);
await expect(opening).resolves.toMatchObject({
status: 'opened',
});
expect(spawnMock).toHaveBeenCalledTimes(2);
expect(JSON.stringify(errorSpy.mock.calls)).toContain(
'mpv loadfile: reply timed out'
);
} finally {
errorSpy.mockRestore();
}
});
it('ignores events and other replies while waiting for its own', async () => {
await launchReusableInstance();
const sockets: MpvSocketMock[] = [];
createConnectionMock.mockImplementation(() => {
const socket = createMpvSocketMock({ reply: () => null });
sockets.push(socket);
return socket;
});
const opening = openMpvPlayer({
title: 'Second stream',
url: 'https://example.com/two.m3u8',
});
let settled = false;
void opening.then(
() => (settled = true),
() => (settled = true)
);
while (!sockets[0]?.write.mock.calls.length) {
await nextTick();
}
const { request_id: requestId } = JSON.parse(
sockets[0].write.mock.calls[0][0] as string
) as { request_id: number };
const push = (...lines: object[]) =>
sockets[0].emit(
'data',
Buffer.from(
lines.map((line) => JSON.stringify(line)).join('\n') + '\n'
)
);
// mpv broadcasts events on every connection and may still be
// answering an earlier request; none of these is this reply.
push(
{ event: 'start-file', playlist_entry_id: 1 },
{
event: 'end-file',
reason: 'error',
file_error: 'loading failed',
},
{ request_id: requestId + 1000, error: 'error running command' }
);
await nextTick();
await nextTick();
expect(settled).toBe(false);
expect(spawnMock).toHaveBeenCalledTimes(1);
push({ request_id: requestId, error: 'success' });
await expect(opening).resolves.toMatchObject({ status: 'opened' });
expect(spawnMock).toHaveBeenCalledTimes(1);
expect(sockets).toHaveLength(1);
});
it('lets a stale Stop for the previous session pass while the loadfile reply is pending', async () => {
const { proc, session: previous } = await launchReusableSession();
const sockets = mockUnansweredSockets();
const opening = openMpvPlayer({
title: 'Second stream',
url: 'https://example.com/two.m3u8',
});
const requestId = await writtenRequestId(sockets[0]);
// The child already belongs to the new session: the old closer
// must neither kill it nor send quit.
await expect(
externalPlayerSessions.closeSession(previous.id)
).resolves.toMatchObject({ id: previous.id, status: 'closed' });
expect(proc.kill).not.toHaveBeenCalled();
expect(sockets).toHaveLength(1);
sockets[0].emit(
'data',
Buffer.from(
JSON.stringify({ request_id: requestId, error: 'success' }) +
'\n'
)
);
await expect(opening).resolves.toMatchObject({ status: 'opened' });
expect(proc.kill).not.toHaveBeenCalled();
expect(spawnMock).toHaveBeenCalledTimes(1);
});
it('settles the new session as closed when the child exits during the reply wait', async () => {
const { proc, session: previous } = await launchReusableSession();
const sockets = mockUnansweredSockets();
const opening = openMpvPlayer({
title: 'Second stream',
url: 'https://example.com/two.m3u8',
contentInfo: {
playlistId: 'playlist-1',
contentXtreamId: 7,
contentType: 'vod',
},
});
const requestId = await writtenRequestId(sockets[0]);
const replacementId =
externalPlayerSessions.getActiveSessionId() as string;
Object.defineProperty(proc, 'exitCode', { value: 0 });
proc.emit('exit', 0);
sockets[0].emit(
'data',
Buffer.from(
JSON.stringify({ request_id: requestId, error: 'success' }) +
'\n'
)
);
// The exit landed on the session that owned the child at that
// moment, and a dead player is never reported as opened.
await expect(opening).resolves.toMatchObject({
id: replacementId,
status: 'closed',
});
expect(externalPlayerSessions.getSession(previous.id)?.status).toBe(
'opened'
);
expect(spawnMock).toHaveBeenCalledTimes(1);
});
it('stops the previous position poll as soon as loadfile is written', async () => {
jest.useFakeTimers();
const contentInfo = {
playlistId: 'playlist-1',
contentXtreamId: 7,
contentType: 'vod' as const,
};
const proc = createMockChildProcess();
spawnMock.mockReturnValueOnce(proc);
(store.get as unknown as jest.Mock).mockImplementation(
(key: string, fallback?: unknown) =>
({
[MPV_PLAYER_PATH]: '/usr/bin/mpv',
[MPV_REUSE_INSTANCE]: true,
})[key] ?? fallback
);
const sockets = mockUnansweredSockets();
const first = openMpvPlayer({
title: 'First stream',
url: 'https://example.com/one.m3u8',
contentInfo,
});
await jest.advanceTimersByTimeAsync(100);
await first;
// The first session polls 2 s after launch, then every 5 s: the
// next poll is due at 7 s, inside the reply window of a loadfile
// written at 6.5 s.
await jest.advanceTimersByTimeAsync(6_500);
const pollsBefore = sockets.length;
const opening = openMpvPlayer({
title: 'Second stream',
url: 'https://example.com/two.m3u8',
contentInfo: { ...contentInfo, contentXtreamId: 8 },
});
await jest.advanceTimersByTimeAsync(0);
const loadfileSocket = sockets[pollsBefore];
const requestId = await writtenRequestId(loadfileSocket);
await jest.advanceTimersByTimeAsync(1_000);
const commandsAfterLoadfile = sockets
.slice(pollsBefore + 1)
.flatMap((socket) =>
socket.write.mock.calls.map(
([request]) =>
(JSON.parse(String(request)) as { command: string[] })
.command[0]
)
);
expect(commandsAfterLoadfile).not.toContain('get_property');
loadfileSocket.emit(
'data',
Buffer.from(
JSON.stringify({ request_id: requestId, error: 'success' }) +
'\n'
)
);
await jest.advanceTimersByTimeAsync(0);
await expect(opening).resolves.toMatchObject({ status: 'opened' });
});
it('drops a position read that was in flight when loadfile was written', async () => {
// The previous test's player exits on a real tick, and any MPV exit
// stops polling; let it land before this test starts its poll.
await nextTick();
await nextTick();
jest.useFakeTimers();
const sendPosition = jest.spyOn(
externalPlayerRuntime,
'sendPlaybackPositionUpdate'
);
const proc = createMockChildProcess();
spawnMock.mockReturnValueOnce(proc);
(store.get as unknown as jest.Mock).mockImplementation(
(key: string, fallback?: unknown) =>
({
[MPV_PLAYER_PATH]: '/usr/bin/mpv',
[MPV_REUSE_INSTANCE]: true,
})[key] ?? fallback
);
const sockets = mockUnansweredSockets();
const commandOf = (socket: MpvSocketMock) =>
socket.write.mock.calls.length
? (
JSON.parse(String(socket.write.mock.calls[0][0])) as {
command: string[];
}
).command
: [];
const first = openMpvPlayer({
title: 'First stream',
url: 'https://example.com/one.m3u8',
contentInfo: {
playlistId: 'playlist-1',
contentXtreamId: 7,
contentType: 'vod',
},
});
await jest.advanceTimersByTimeAsync(100);
const previous = await first;
// The first poll fires 2 s + 5 s after launch and waits for time-pos.
await jest.advanceTimersByTimeAsync(6_950);
const timePosSocket = sockets.find(
(socket) => commandOf(socket)[1] === 'time-pos'
);
expect(timePosSocket).toBeDefined();
const opening = openMpvPlayer({
title: 'Second stream',
url: 'https://example.com/two.m3u8',
});
await jest.advanceTimersByTimeAsync(0);
const loadfileSocket = sockets.find(
(socket) => commandOf(socket)[0] === 'loadfile'
) as MpvSocketMock;
const requestId = await writtenRequestId(loadfileSocket);
// The pending read now returns the new stream's position.
timePosSocket?.emit(
'data',
Buffer.from(JSON.stringify({ data: 12.5, error: 'success' }) + '\n')
);
await jest.advanceTimersByTimeAsync(10);
loadfileSocket.emit(
'data',
Buffer.from(
JSON.stringify({ request_id: requestId, error: 'success' }) +
'\n'
)
);
await jest.advanceTimersByTimeAsync(0);
await expect(opening).resolves.toMatchObject({ status: 'opened' });
// An unguarded poll would now ask for the duration and, once that
// read times out, report the new position under the old content.
await jest.advanceTimersByTimeAsync(MPV_IPC_COMMAND_TIMEOUT_MS + 100);
expect(
sockets.some((socket) => commandOf(socket)[1] === 'duration')
).toBe(false);
expect(sendPosition).not.toHaveBeenCalledWith(
previous.id,
expect.anything(),
expect.anything()
);
sendPosition.mockRestore();
});
it('keeps a reuse cancelled by disabling reuse stoppable until its child exits', async () => {
const { proc } = await launchReusableSession();
(proc.kill as jest.Mock).mockImplementation(() => true);
const sockets = mockUnansweredSockets();
jest.useFakeTimers();
try {
const opening = openMpvPlayer({
title: 'Channel B',
url: 'https://example.com/b.m3u8',
});
while (!sockets[0]?.write.mock.calls.length) {
await jest.advanceTimersByTimeAsync(0);
}
setMpvReuseInstance(false);
sockets[0].emit('close');
await jest.advanceTimersByTimeAsync(6_000);
// The child never confirmed its exit: the session stays a
// closable error instead of a closed one Stop cannot retry.
await expect(opening).resolves.toMatchObject({
status: 'error',
canClose: true,
});
expect(spawnMock).toHaveBeenCalledTimes(1);
} finally {
Object.defineProperty(proc, 'exitCode', { value: 0 });
proc.emit('exit', 0);
}
});
it('keeps Stop working for a reused session restored after a later attempt fails', async () => {
const { proc } = await launchReusableSession();
// Exit is never confirmed, so a failing later attempt cannot finish
// its teardown and the previous owner is restored.
(proc.kill as jest.Mock).mockImplementation(() => true);
mockMpvSockets();
const restored = await openMpvPlayer({
title: 'Channel A',
url: 'https://example.com/a.m3u8',
});
expect(restored.status).toBe('opened');
createConnectionMock.mockImplementation(() => {
const socket = createMpvSocketMock({ connect: false });
setImmediate(() =>
socket.emit('error', new Error('reuse command failed'))
);
return socket;
});
const errorSpy = jest
.spyOn(console, 'error')
.mockImplementation(() => undefined);
jest.useFakeTimers();
try {
const failing = openMpvPlayer({
title: 'Channel B',
url: 'https://example.com/b.m3u8',
});
const rejection = expect(failing).rejects.toThrow(
'External player process did not exit'
);
await jest.advanceTimersByTimeAsync(6_000);
await rejection;
expect(externalPlayerSessions.getActiveSessionId()).toBe(
restored.id
);
const written: string[] = [];
mockMpvSockets({ written });
let closed = false;
const closing = externalPlayerSessions
.closeSession(restored.id)
.then((session) => {
closed = true;
return session;
});
await jest.advanceTimersByTimeAsync(10);
// Stop reaches the child and waits for its exit.
expect(
written.map(
(chunk) =>
(JSON.parse(chunk.trim()) as { command: string[] })
.command[0]
)
).toContain('quit');
expect(closed).toBe(false);
Object.defineProperty(proc, 'exitCode', { value: 0 });
proc.emit('exit', 0);
await expect(closing).resolves.toMatchObject({
id: restored.id,
status: 'closed',
});
} finally {
errorSpy.mockRestore();
if (proc.exitCode === null) {
Object.defineProperty(proc, 'exitCode', { value: 0 });
proc.emit('exit', 0);
}
}
});
it.each([
['reuse is disabled', () => setMpvReuseInstance(false)],
['the app shuts down', () => shutdownMpvSession()],
])(
'does not fall back to a fresh player when %s during the reply wait',
async (_label, stopStored) => {
const { proc } = await launchReusableSession();
// The kill is sent but its exit is reported only later, after
// the socket close already rejected the pending loadfile.
(proc.kill as jest.Mock).mockImplementation(() => true);
const sockets = mockUnansweredSockets();
const opening = openMpvPlayer({
title: 'Channel B',
url: 'https://example.com/b.m3u8',
});
await writtenRequestId(sockets[0]);
stopStored();
sockets[0].emit('close');
await nextTick();
await nextTick();
Object.defineProperty(proc, 'exitCode', { value: 0 });
proc.emit('exit', 0);
await expect(opening).resolves.toMatchObject({
status: 'closed',
});
expect(spawnMock).toHaveBeenCalledTimes(1);
}
);
it.each([
['reuse is disabled', () => setMpvReuseInstance(false)],
['the app shuts down', () => shutdownMpvSession()],
])(
'does not fall back to a fresh player when %s during the failed-reuse teardown',
async (_label, stopStored) => {
const { proc } = await launchReusableSession();
// mpv rejects the content command, and the teardown that
// follows waits for an exit that is reported only later.
(proc.kill as jest.Mock).mockImplementation(() => true);
mockMpvSockets({
reply: (command) =>
command[0] === 'loadfile'
? 'error running command'
: 'success',
});
const errorSpy = jest
.spyOn(console, 'error')
.mockImplementation(() => undefined);
try {
const opening = openMpvPlayer({
title: 'Channel B',
url: 'https://example.com/b.m3u8',
});
while (!(proc.kill as jest.Mock).mock.calls.length) {
await nextTick();
}
stopStored();
Object.defineProperty(proc, 'exitCode', { value: 0 });
proc.emit('exit', 0);
await expect(opening).resolves.toMatchObject({
status: 'closed',
});
expect(spawnMock).toHaveBeenCalledTimes(1);
} finally {
errorSpy.mockRestore();
}
}
);
it('keeps the errored session when the child crashes during the reply wait', async () => {
const { proc, session: previous } = await launchReusableSession();
const sockets = mockUnansweredSockets();
const errorSpy = jest
.spyOn(console, 'error')
.mockImplementation(() => undefined);
try {
const opening = openMpvPlayer({
title: 'Second stream',
url: 'https://example.com/two.m3u8',
});
await writtenRequestId(sockets[0]);
const replacementId =
externalPlayerSessions.getActiveSessionId() as string;
// A crash closes the IPC socket before any reply: the exit
// handler reports it, and no fresh player is launched under a
// session that could never be shown as opened.
Object.defineProperty(proc, 'exitCode', { value: 1 });
proc.emit('exit', 1);
sockets[0].emit('close');
await expect(opening).resolves.toMatchObject({
id: replacementId,
status: 'error',
errorCode: 'closed-unexpectedly',
});
expect(externalPlayerSessions.getSession(previous.id)?.status).toBe(
'opened'
);
expect(spawnMock).toHaveBeenCalledTimes(1);
} finally {
errorSpy.mockRestore();
}
});
});
@@ -61,6 +61,11 @@ import {
setVlcReuseInstance,
shutdownVlcSession,
} from './vlc-session.service';
import {
createMpvSocketMock,
MpvSocketMock,
MpvSocketMockOptions,
} from './mpv-ipc-socket.test-helpers';
function createMockChildProcess(): ChildProcess {
return Object.assign(new EventEmitter(), {
@@ -95,6 +100,13 @@ function mockStoreValues(values: Record<string, unknown>): void {
);
}
/** Every reused-instance IPC connection answers like a healthy mpv. */
function mockMpvSockets(options: MpvSocketMockOptions = {}): void {
(createConnection as unknown as jest.Mock).mockImplementation(() =>
createMpvSocketMock(options)
);
}
describe('external player shutdown on app quit', () => {
beforeEach(() => {
jest.clearAllMocks();
@@ -263,15 +275,7 @@ describe('external player shutdown on app quit', () => {
title: 'First stream',
url: 'https://example.com/one.m3u8',
});
(createConnection as unknown as jest.Mock).mockImplementation(() => {
const socket = Object.assign(new EventEmitter(), {
write: jest.fn(),
end: jest.fn(),
destroy: jest.fn(),
});
setImmediate(() => socket.emit('connect'));
return socket;
});
mockMpvSockets();
const current = await openMpvPlayer({
title: 'Second stream',
url: 'https://example.com/two.m3u8',
@@ -305,15 +309,7 @@ describe('external player shutdown on app quit', () => {
url: 'https://example.com/one.m3u8',
});
const written: string[] = [];
(createConnection as unknown as jest.Mock).mockImplementation(() => {
const socket = Object.assign(new EventEmitter(), {
write: jest.fn((chunk: string) => written.push(chunk)),
end: jest.fn(),
destroy: jest.fn(),
});
setImmediate(() => socket.emit('connect'));
return socket;
});
mockMpvSockets({ written });
const reused = await openMpvPlayer({
title: 'Second stream',
url: 'https://example.com/two.m3u8',
@@ -394,15 +390,7 @@ describe('external player shutdown on app quit', () => {
expect(launchArgs).toContain('--start=42');
const written: string[] = [];
(createConnection as unknown as jest.Mock).mockImplementation(() => {
const socket = Object.assign(new EventEmitter(), {
write: jest.fn((chunk: string) => written.push(chunk)),
end: jest.fn(),
destroy: jest.fn(),
});
setImmediate(() => socket.emit('connect'));
return socket;
});
mockMpvSockets({ written });
const loadfileOptionsOf = async (startTime: number | undefined) => {
written.length = 0;
await openMpvPlayer({
@@ -456,15 +444,7 @@ describe('external player shutdown on app quit', () => {
// Capture the JSON IPC traffic of the second, reused launch.
const written: string[] = [];
(createConnection as unknown as jest.Mock).mockImplementation(() => {
const socket = Object.assign(new EventEmitter(), {
write: jest.fn((chunk: string) => written.push(chunk)),
end: jest.fn(),
destroy: jest.fn(),
});
setImmediate(() => socket.emit('connect'));
return socket;
});
mockMpvSockets({ written });
await openMpvPlayer({
title: 'Stalker live stream',
@@ -509,15 +489,7 @@ describe('external player shutdown on app quit', () => {
url: 'https://example.com/one.m3u8',
});
(createConnection as unknown as jest.Mock).mockImplementation(() => {
const socket = Object.assign(new EventEmitter(), {
write: jest.fn(),
end: jest.fn(),
destroy: jest.fn(),
});
setImmediate(() => socket.emit('connect'));
return socket;
});
mockMpvSockets();
const reused = await openMpvPlayer({
title: 'Second stream',
url: 'https://example.com/two.m3u8',
@@ -560,15 +532,7 @@ describe('external player shutdown on app quit', () => {
title: 'First stream',
url: 'https://example.com/one.m3u8',
});
(createConnection as unknown as jest.Mock).mockImplementation(() => {
const socket = Object.assign(new EventEmitter(), {
write: jest.fn(),
end: jest.fn(),
destroy: jest.fn(),
});
setImmediate(() => socket.emit('connect'));
return socket;
});
mockMpvSockets();
const reused = await openMpvPlayer({
title: 'Second stream',
url: 'https://example.com/two.m3u8',
@@ -622,27 +586,15 @@ describe('external player shutdown on app quit', () => {
url: 'https://example.com/one.m3u8',
});
(createConnection as unknown as jest.Mock).mockImplementation(() => {
const socket = Object.assign(new EventEmitter(), {
write: jest.fn(),
end: jest.fn(),
destroy: jest.fn(),
});
setImmediate(() => socket.emit('connect'));
return socket;
});
mockMpvSockets();
const session = await openMpvPlayer({
title: 'Second stream',
url: 'https://example.com/two.m3u8',
});
const sockets: EventEmitter[] = [];
const sockets: MpvSocketMock[] = [];
(createConnection as unknown as jest.Mock).mockImplementation(() => {
const socket = Object.assign(new EventEmitter(), {
write: jest.fn(),
end: jest.fn(),
destroy: jest.fn(),
});
const socket = createMpvSocketMock({ connect: false });
sockets.push(socket);
return socket;
});
@@ -676,19 +628,12 @@ describe('external player shutdown on app quit', () => {
url: 'https://example.com/one.m3u8',
});
const sockets: Array<
EventEmitter & { write: jest.Mock; end: jest.Mock }
> = [];
const sockets: MpvSocketMock[] = [];
(createConnection as unknown as jest.Mock).mockImplementation(() => {
const socket = Object.assign(new EventEmitter(), {
write: jest.fn(),
end: jest.fn(),
destroy: jest.fn(),
const socket = createMpvSocketMock({
connect: sockets.length >= 2,
});
sockets.push(socket);
if (sockets.length > 2) {
setImmediate(() => socket.emit('connect'));
}
return socket;
});
@@ -720,7 +665,7 @@ describe('external player shutdown on app quit', () => {
expect(spawn).toHaveBeenCalledTimes(1);
});
it('does not spawn a replacement when Stop interrupts reused MPV seek', async () => {
it('does not spawn a replacement when Stop interrupts a reused MPV loadfile reply', async () => {
shutdownMpvSession();
const proc = createMockChildProcess();
(spawn as unknown as jest.Mock).mockReturnValue(proc);
@@ -733,17 +678,14 @@ describe('external player shutdown on app quit', () => {
url: 'https://example.com/one.m3u8',
});
const sockets: EventEmitter[] = [];
// The content command is written but mpv never answers it; the quit
// connection that Stop opens answers normally.
const sockets: MpvSocketMock[] = [];
(createConnection as unknown as jest.Mock).mockImplementation(() => {
const socket = Object.assign(new EventEmitter(), {
write: jest.fn(),
end: jest.fn(),
destroy: jest.fn(),
const socket = createMpvSocketMock({
reply: sockets.length === 0 ? () => null : undefined,
});
sockets.push(socket);
if (sockets.length === 1 || sockets.length === 3) {
setImmediate(() => socket.emit('connect'));
}
return socket;
});
@@ -752,19 +694,19 @@ describe('external player shutdown on app quit', () => {
url: 'https://example.com/two.m3u8',
startTime: 120,
});
while (sockets.length < 2) {
while (!sockets[0]?.write.mock.calls.length) {
await new Promise<void>((resolve) => setImmediate(resolve));
}
const replacementId =
externalPlayerSessions.getActiveSessionId() as string;
const closing = externalPlayerSessions.closeSession(replacementId);
while (sockets.length < 3) {
while (sockets.length < 2) {
await new Promise<void>((resolve) => setImmediate(resolve));
}
Object.defineProperty(proc, 'exitCode', { value: 0 });
proc.emit('exit', 0);
sockets[1].emit('error', new Error('connection closed'));
sockets[0].emit('error', new Error('connection closed'));
await expect(closing).resolves.toMatchObject({
id: replacementId,
@@ -972,22 +914,11 @@ describe('external player shutdown on app quit', () => {
title: 'First stream',
url: 'https://example.com/one.m3u8',
});
let connectionCount = 0;
(createConnection as unknown as jest.Mock).mockImplementation(() => {
connectionCount += 1;
const socket = Object.assign(new EventEmitter(), {
write: jest.fn(),
end: jest.fn(),
destroy: jest.fn(),
});
setImmediate(() => {
if (connectionCount === 1) {
socket.emit('connect');
} else {
socket.emit('error', new Error('seek failed'));
}
});
return socket;
// mpv rejects the content command after it was written, so the
// player may already have changed.
mockMpvSockets({
reply: (command) =>
command[0] === 'loadfile' ? 'error running command' : 'success',
});
jest.useFakeTimers();
@@ -1153,22 +1084,11 @@ describe('external player shutdown on app quit', () => {
title: 'First stream',
url: 'https://example.com/one.m3u8',
});
let connectionCount = 0;
(createConnection as unknown as jest.Mock).mockImplementation(() => {
connectionCount += 1;
const socket = Object.assign(new EventEmitter(), {
write: jest.fn(),
end: jest.fn(),
destroy: jest.fn(),
});
setImmediate(() => {
if (connectionCount === 1) {
socket.emit('connect');
} else {
socket.emit('error', new Error('seek failed'));
}
});
return socket;
// mpv rejects the content command after it was written, so the
// player may already have changed.
mockMpvSockets({
reply: (command) =>
command[0] === 'loadfile' ? 'error running command' : 'success',
});
const session = await openMpvPlayer({
@@ -1,6 +1,6 @@
import { spawn } from 'child_process';
import { createConnection } from 'net';
import {
ExternalPlayerSession,
PlayerContentInfo,
readExternalPlayerErrorCode,
} from '@iptvnator/shared/interfaces';
@@ -28,15 +28,25 @@ import {
buildPlayerStartError,
externalPlayerSessions,
maskUrlForLogs,
sendPlaybackPositionUpdate,
sendPlayerErrorNotification,
traceExternalPlayer,
} from './external-player-runtime';
import { externalPlayerProcessTeardownGate } from './external-player-process';
import {
assertQueuedLaunchAllowed,
MpvLaunchQueue,
QueuedMpvLaunch,
settleCancelledQueuedLaunch,
} from './mpv-launch-queue';
import {
MpvReusableProcess,
MpvReuseAttemptState,
} from './mpv-reusable-process';
import { attachMpvOutputHandlers } from './mpv-output-errors';
import {
startPositionPolling,
stopPositionPolling,
} from './mpv-position-polling';
export interface OpenExternalPlayerRequest {
url: string;
@@ -51,11 +61,6 @@ export interface OpenExternalPlayerRequest {
}
const reusableMpvProcess = new MpvReusableProcess();
let positionPollingInterval: NodeJS.Timeout | null = null;
// The first poll waits for MPV to open its IPC socket. The handle is kept so
// a player that exits during that wait does not start an orphaned interval.
let positionPollingDelay: NodeJS.Timeout | null = null;
function getMpvPath(options: PlayerPathOptions = {}): string {
return (
normalizeCustomPlayerPath(store.get(MPV_PLAYER_PATH)) ??
@@ -63,110 +68,6 @@ function getMpvPath(options: PlayerPathOptions = {}): string {
);
}
async function getMpvProperty(
socketPath: string,
property: string
): Promise<number | null> {
return new Promise((resolve) => {
const client = createConnection(socketPath);
const request =
JSON.stringify({
command: ['get_property', property],
}) + '\n';
let data = '';
client.on('connect', () => {
client.write(request);
});
const timeoutHandle = setTimeout(() => {
client.destroy();
resolve(null);
}, 2000);
client.on('data', (chunk) => {
data += chunk.toString();
if (data.includes('\n')) {
clearTimeout(timeoutHandle);
try {
const lines = data.split('\n');
for (const line of lines) {
if (!line.trim()) continue;
const response = JSON.parse(line);
if (response.data !== undefined) {
client.destroy();
resolve(response.data);
return;
}
}
} catch {
// Ignore partial JSON while MPV is still writing.
}
}
});
client.on('end', () => {
clearTimeout(timeoutHandle);
if (data && !data.includes('\n')) {
try {
const response = JSON.parse(data);
resolve(response.data ?? null);
} catch {
resolve(null);
}
}
});
client.on('error', () => {
clearTimeout(timeoutHandle);
resolve(null);
});
});
}
function stopPositionPolling(): void {
if (positionPollingDelay) {
clearTimeout(positionPollingDelay);
positionPollingDelay = null;
}
if (positionPollingInterval) {
clearInterval(positionPollingInterval);
positionPollingInterval = null;
}
}
function shouldIgnoreMpvStdoutLine(line: string): boolean {
return /^(\(Paused\)\s*)?AV:\s/.test(line);
}
function startPositionPolling(
socketPath: string,
contentInfo: PlayerContentInfo,
sessionId: string
): void {
stopPositionPolling();
positionPollingDelay = setTimeout(() => {
positionPollingDelay = null;
positionPollingInterval = setInterval(async () => {
try {
const position = await getMpvProperty(socketPath, 'time-pos');
const duration = await getMpvProperty(socketPath, 'duration');
if (position !== null) {
sendPlaybackPositionUpdate(sessionId, contentInfo, {
positionSeconds: Math.floor(position),
durationSeconds: duration ? Math.floor(duration) : null,
});
}
} catch {
stopPositionPolling();
}
}, 5000);
}, 2000);
}
export function setMpvReuseInstance(reuseInstance: boolean): void {
traceExternalPlayer('set mpv reuse instance', { reuseInstance });
store.set(MPV_REUSE_INSTANCE, reuseInstance);
@@ -186,33 +87,71 @@ export function setMpvReuseInstance(reuseInstance: boolean): void {
* app and keeps playing after quit.
*/
export function shutdownMpvSession(): void {
mpvLaunchQueue.cancelAll();
reusableMpvProcess.stopStored(
'kill reused mpv process on app shutdown',
stopPositionPolling
);
}
export async function openMpvPlayer({
url,
title,
thumbnail,
userAgent,
referer,
origin,
contentInfo,
startTime,
headers,
}: OpenExternalPlayerRequest) {
externalPlayerProcessTeardownGate.assertLaunchAllowed();
// MPV launches run one at a time; see MpvLaunchQueue. Each launch begins its
// session at once, so the newest launch is the active one Stop targets.
const mpvLaunchQueue = new MpvLaunchQueue(() =>
reusableMpvProcess.storedOwnerSessionId()
);
export function openMpvPlayer(
request: OpenExternalPlayerRequest
): Promise<ExternalPlayerSession> {
try {
externalPlayerProcessTeardownGate.assertLaunchAllowed();
} catch (error) {
return Promise.reject(error);
}
const displacedSessionId = externalPlayerSessions.getActiveSessionId();
const previousProcessSessionId = reusableMpvProcess.currentSessionId();
const session = externalPlayerSessions.beginSession({
player: 'mpv',
title,
thumbnail,
streamUrl: url,
contentInfo,
title: request.title,
thumbnail: request.thumbnail,
streamUrl: request.url,
contentInfo: request.contentInfo,
});
return mpvLaunchQueue.enqueue(session, (queue) =>
launchMpvPlayer(request, session, displacedSessionId, queue)
);
}
async function launchMpvPlayer(
{
url,
title,
userAgent,
referer,
origin,
contentInfo,
startTime,
headers,
}: OpenExternalPlayerRequest,
session: ExternalPlayerSession,
displacedSessionId: string | null,
queue: QueuedMpvLaunch
): Promise<ExternalPlayerSession> {
const current = externalPlayerSessions.getSession(session.id);
// Stopped while it waited in the queue, directly or by a later Stop.
if (current?.status === 'closed') return current;
if (queue.queued && queue.cancelled()) {
return settleCancelledQueuedLaunch(session, queue.pendingStop());
}
if (queue.queued) {
// The last started launch owns the child the gate waits for; the
// session displaced at enqueue time is the fallback.
assertQueuedLaunchAllowed(session, [
queue.lastStartedSessionId(),
displacedSessionId,
]);
}
queue.markInFlight();
const previousProcessSessionId = reusableMpvProcess.currentSessionId();
const reuseState: MpvReuseAttemptState = {
teardownUnconfirmed: false,
contentMutated: false,
@@ -356,61 +295,7 @@ export async function openMpvPlayer({
stdio: reuseInstance ? ['ignore', 'pipe', 'pipe'] : 'ignore',
});
if (proc.stdout) {
proc.stdout.on('data', (data) => {
const lines = data
.toString()
.split('\n')
.map((line) => line.trim())
.filter(Boolean);
for (const output of lines) {
if (shouldIgnoreMpvStdoutLine(output)) {
continue;
}
traceExternalPlayer('mpv stdout', { output });
if (
output.includes('Failed to open') ||
output.includes('Error opening') ||
output.includes('Protocol not found') ||
output.includes('Connection refused') ||
output.includes('error') ||
output.includes('403') ||
output.includes('404')
) {
console.error('[MPV ERROR from stdout]:', output);
sendPlayerErrorNotification('MPV', output);
}
}
});
}
if (proc.stderr) {
proc.stderr.on('data', (data) => {
const output = data.toString().trim();
if (output) {
console.error('[MPV stderr]:', output);
if (
output.includes('Failed to open') ||
output.includes(
'Exiting... (Errors when loading file)'
) ||
output.includes('Error opening') ||
output.includes('Protocol not found') ||
output.includes('Connection refused') ||
output.includes('error') ||
output.includes('403') ||
output.includes('404')
) {
console.error('[MPV ERROR]:', output);
sendPlayerErrorNotification('MPV', output);
}
}
});
}
attachMpvOutputHandlers(proc);
proc.on('error', (err) => {
console.error('Failed to start MPV player:', err);
@@ -0,0 +1,132 @@
/**
* Shared fixtures for the reused-MPV specs: a mock child, a reusable
* instance to hand launches over to, and IPC sockets that answer, stay
* silent, or let `quit` stop the child.
*/
import { spawn, type ChildProcess } from 'child_process';
import { EventEmitter } from 'events';
import { createConnection } from 'net';
import type { ExternalPlayerSession } from '@iptvnator/shared/interfaces';
import {
MPV_PLAYER_PATH,
MPV_REUSE_INSTANCE,
store,
} from '../services/store.service';
import {
createMpvSocketMock,
MpvSocketMock,
MpvSocketMockOptions,
} from './mpv-ipc-socket.test-helpers';
import { openMpvPlayer } from './mpv-session.service';
export const spawnMock = spawn as unknown as jest.Mock;
export const createConnectionMock = createConnection as unknown as jest.Mock;
export function createMockChildProcess(): ChildProcess {
const proc = Object.assign(new EventEmitter(), {
exitCode: null,
killed: false,
kill: jest.fn(),
signalCode: null,
stderr: null,
stdout: null,
unref: jest.fn(),
}) as unknown as ChildProcess;
(proc.kill as jest.Mock).mockImplementation(() => {
setImmediate(() => {
Object.defineProperty(proc, 'exitCode', { value: 0 });
proc.emit('exit', 0);
});
return true;
});
return proc;
}
export function mockMpvSockets(options: MpvSocketMockOptions = {}): void {
createConnectionMock.mockImplementation(() => createMpvSocketMock(options));
}
export function nextTick(): Promise<void> {
return new Promise<void>((resolve) => setImmediate(resolve));
}
/** Spawns the reusable instance a later launch is handed over to. */
export async function launchReusableInstance(): Promise<ChildProcess> {
return (await launchReusableSession()).proc;
}
export async function launchReusableSession(): Promise<{
proc: ChildProcess;
session: ExternalPlayerSession;
}> {
const proc = createMockChildProcess();
spawnMock.mockReturnValueOnce(proc);
(store.get as unknown as jest.Mock).mockImplementation(
(key: string, fallback?: unknown) =>
({
[MPV_PLAYER_PATH]: '/usr/bin/mpv',
[MPV_REUSE_INSTANCE]: true,
})[key] ?? fallback
);
const session = await openMpvPlayer({
title: 'First stream',
url: 'https://example.com/one.m3u8',
});
return { proc, session };
}
/** A socket whose written command mpv never answers; the test replies. */
export function mockUnansweredSockets(): MpvSocketMock[] {
const sockets: MpvSocketMock[] = [];
createConnectionMock.mockImplementation(() => {
const socket = createMpvSocketMock({ reply: () => null });
sockets.push(socket);
return socket;
});
return sockets;
}
/**
* Like mockUnansweredSockets, but a `quit` succeeds and the child then exits,
* as a real mpv does.
*/
export function mockSocketsQuittingInto(proc: ChildProcess): MpvSocketMock[] {
const sockets: MpvSocketMock[] = [];
createConnectionMock.mockImplementation(() => {
const socket = createMpvSocketMock({
reply: (command) => {
if (command[0] !== 'quit') return null;
setImmediate(() => {
Object.defineProperty(proc, 'exitCode', { value: 0 });
proc.emit('exit', 0);
});
return 'success';
},
});
sockets.push(socket);
return socket;
});
return sockets;
}
export function commandsOf(sockets: MpvSocketMock[]): string[] {
return sockets.map(
(socket) =>
(
JSON.parse(String(socket.write.mock.calls[0][0])) as {
command: string[];
}
).command[0]
);
}
export async function writtenRequestId(socket: MpvSocketMock): Promise<number> {
while (!socket?.write.mock.calls.length) {
await nextTick();
}
return (
JSON.parse(socket.write.mock.calls[0][0] as string) as {
request_id: number;
}
).request_id;
}
@@ -62,6 +62,49 @@ describe('ExternalPlaybackService', () => {
expect(service.visibleSession()).toEqual(next);
});
it('keeps the newest launch when an older launch republishes as launching', () => {
const older = createSession({
id: 'session-older',
startedAt: '2026-03-07T10:00:00.000Z',
});
const newer = createSession({
id: 'session-newer',
title: 'Newer Example',
startedAt: '2026-03-07T10:00:01.000Z',
});
listener?.(older);
listener?.(newer);
// The older launch is still settling and republishes itself.
listener?.({
...older,
canClose: true,
updatedAt: '2026-03-07T10:00:02.000Z',
});
expect(service.activeSession()?.id).toBe('session-newer');
});
it('orders launches made in the same millisecond by launch order', () => {
const older = createSession({ id: 'session-older', launchOrder: 1 });
const newer = createSession({
id: 'session-newer',
title: 'Newer Example',
launchOrder: 2,
});
listener?.(older);
listener?.(newer);
// Same startedAt: only the launch order tells the two apart.
listener?.({
...older,
canClose: true,
updatedAt: '2026-03-07T10:00:02.000Z',
});
expect(service.activeSession()?.id).toBe('session-newer');
});
it('matches only active non-error sessions to content info', () => {
const playing = createSession({
status: 'playing',
@@ -4,6 +4,17 @@ import {
PlayerContentInfo,
} from '@iptvnator/shared/interfaces';
/** Launch order when both carry it, else the start time. */
function startedBefore(
session: ExternalPlayerSession,
other: ExternalPlayerSession
): boolean {
if (session.launchOrder !== undefined && other.launchOrder !== undefined) {
return session.launchOrder < other.launchOrder;
}
return Date.parse(session.startedAt) < Date.parse(other.startedAt);
}
@Injectable({
providedIn: 'root',
})
@@ -105,17 +116,25 @@ export class ExternalPlaybackService {
const current = this.activeSession();
const restoresCurrentReplacement =
current?.id === session.restoredFromSessionId;
// A launching update means a new launch only when it did not start
// before the current one: an older launch still settling (MPV runs
// launches one at a time) must not take the dock back from it.
const isLatestLaunch =
session.status === 'launching' &&
(!current ||
current.id === session.id ||
!startedBefore(session, current));
if (
!current ||
current.id === session.id ||
session.status === 'launching' ||
isLatestLaunch ||
restoresCurrentReplacement
) {
this.activeSession.set(session);
}
if (session.status === 'launching' || restoresCurrentReplacement) {
if (isLatestLaunch || restoresCurrentReplacement) {
this.dismissedSessionId.set(null);
}
}
+62 -2
View File
@@ -1029,6 +1029,61 @@ is also idempotent: it returns the closed snapshot without invoking the saved
closer, and a later process error cannot revive it as a visible failure. Reused
MPV commands use the socket captured for their exact child, so a delayed close
cannot send `quit` to a replacement process through a newer global socket.
Every reused MPV command carries a `request_id` and waits, within the same 2 s
IPC bound, for mpv's matching newline-delimited reply; event lines and replies
to other requests on that connection are skipped. An error reply or a missing
reply fails the handoff exactly like a socket error: the reused child is torn
down and a fresh launch follows. `loadfile` marks the content as possibly
changed and moves ownership of the child to the attempted session from the
moment it is written, whatever mpv replies, and stops the previous session's
position poll there, including a read already awaiting mpv, so an exit or a
stale Stop for the previous session during
the reply wait settles the right session and no poll can save the new stream's
position under the old content; a child that exited while its reply was in
flight is returned as the session its exit handler settled, never reported as
opened and never replaced by a fresh launch under that terminal session.
MPV launches run one at a time. A reused launch waits for mpv's replies and a
failed one for its teardown and fallback spawn, so a launch that arrives
meanwhile, such as a quick channel switch, starts only once the previous one
has settled and acts on the player that one left behind; two launches never
act on the same child. Each launch begins its session at once, so the newest
launch is the active session; the renderer lets a `launching` update take the
dock only when that launch did not start before the current one, by the
session's `launchOrder` (which separates launches in the same millisecond) or
else its `startedAt`, so an older launch still settling cannot pull the dock
back. Only the newest queued launch
runs: an older
one that comes up while a newer one waits behind it settles closed without
reaching mpv or taking the dock. Stop on a launch still waiting in the queue
cancels it and every launch queued before it; on the newest queued launch it
also closes, through their own closers, the in-flight launch and the session
owning the stored player at that moment (read from the per-child mapping, which
a failed attempt hands back to the previous owner during its teardown), so
their exact-child, retry and settle rules apply. Because launches run one at a
time, nothing else acts on a reused child while an attempt is in flight, so
that attempt's own Stop always reaches its child until the attempt settles,
whatever the per-child mapping names (the previous owner before `loadfile` or
during a failed attempt's teardown, possibly closed already); after it settles,
ownership decides as before, except that a Stop of that attempt which could
not confirm the exit keeps the right to retry that exact child. These rules apply
and nothing starts playing after Stop. Those targets are fixed at the first
Stop, so a retried Stop never reaches a player started later, and a late Stop
for an older queued launch cancels only that launch. The cancelled launch
settles only once its own Stop has: if the exit cannot be confirmed it stays a
closable error, so Stop can be retried. App shutdown cancels the whole queue.
A queued launch that finds the teardown gate closed, because the launch before
it could not confirm its child's exit, restores the session that still owns
that child, preferring the last launch that started (with reuse off its
detached child is not stored anywhere else); if that is not possible, it stays
a closable error so Stop can still tear the child down. Stopping the stored player, when reuse is disabled or
the app quits, cancels an in-flight reuse attempt: it dispatches no further
commands, never falls back to a fresh reusable player, also after waiting for
a failed child's exit, and settles its own session as closed once the child's
exit is confirmed, or as a closable error when it is not. No `seek` follows
`loadfile`: mpv rejects a seek before the file is loaded, so the resume offset
travels as the per-file `start` option. The `loadfile` reply only acknowledges
the command; a stream that later fails to open is reported through the reused
child's output and exit, not through the reply.
If Stop is observed before a pending MPV content command or VLC enqueue command
is dispatched, that command is skipped. A source handoff also fails closed
while a live session has no closer (`canClose: false`); renderer Dismiss is not
@@ -1142,8 +1197,13 @@ External MPV/VLC integration is split across focused main-process modules:
session tracking, trace logging, renderer notifications, playback-position
forwarding, and user-facing start errors.
- `apps/electron-backend/src/app/events/mpv-session.service.ts` owns fresh MPV
launches and progress polling; `mpv-reusable-process.ts` owns the tracked
child, captured socket, remapping, retryable close, and reuse handoff.
launches; `mpv-position-polling.ts` owns progress polling over the IPC socket
and `mpv-output-errors.ts` the child's stdout/stderr error reporting;
`mpv-reusable-process.ts` owns the tracked
child, captured socket, remapping, retryable close, and reuse handoff;
`mpv-ipc-command.ts` owns the bounded request/reply IPC command the reuse
handoff and its quit use; `mpv-launch-queue.ts` runs MPV launches one at a
time and owns queued-launch cancellation and Stop.
- `apps/electron-backend/src/app/events/vlc-session.service.ts` owns fresh VLC
launches and progress polling; `vlc-reusable-process.ts` owns the tracked
child, RC-port remapping, retryable close, and reuse handoff, while
+1 -1
View File
@@ -210,7 +210,7 @@ user who leaves that screen open pays them indefinitely, minimized included.
| Source | What it does | Period | Cost per firing | Justified | Evidence | Follow-up |
| --- | --- | --- | --- | --- | --- | --- |
| [mpv-session.service.ts:140–141](../../apps/electron-backend/src/app/events/mpv-session.service.ts) | External MPV position poll | 2 s delay, then 5 s | Two IPC-socket round-trips and one `playback-position-update` | Yes while playing. The early-exit leak found here (an unstored start-delay handle leaving an orphaned 5 s poll) is **resolved by #1720**. Still open: in reuse mode the poll keeps querying an idle MPV (no IPC while `time-pos` is null). | Static | Early-exit leak: done (#1720). Reuse mode: needs an idle signal from MPV before the poll can stop; not worth its own thread at two local socket calls per 5 s. |
| [mpv-position-polling.ts](../../apps/electron-backend/src/app/events/mpv-position-polling.ts) | External MPV position poll | 2 s delay, then 5 s | Two IPC-socket round-trips and one `playback-position-update` | Yes while playing. The early-exit leak found here (an unstored start-delay handle leaving an orphaned 5 s poll) is **resolved by #1720**. Still open: in reuse mode the poll keeps querying an idle MPV (no IPC while `time-pos` is null). | Static | Early-exit leak: done (#1720). Reuse mode: needs an idle signal from MPV before the poll can stop; not worth its own thread at two local socket calls per 5 s. |
| [vlc-session.service.ts:81–82](../../apps/electron-backend/src/app/events/vlc-session.service.ts) | External VLC position poll | 1.5 s delay, then 2 s | Up to three localhost TCP connections, then IPC | Yes while playing. The early-exit leak (orphaned 2 s poll) is **resolved by #1720**. | Static | Done (#1720). |
| [embedded-mpv-native.service.ts:1068](../../apps/electron-backend/src/app/services/embedded-mpv-native.service.ts) | Embedded MPV session snapshot poll | 500 ms while a session exists | Native snapshot, diff, and IPC only on change | Yes during playback. It is cleared when the last session closes. | Static | None |
| [embedded-mpv-session-controller.ts:197](../../libs/ui/playback/src/lib/embedded-mpv-player/embedded-mpv-session-controller.ts) | Renderer-side bounds poll for a native-view embedded MPV session: compares `getBoundingClientRect()` with the last synced bounds | 500 ms while a native-view session is open (skipped for frame-copy) | One layout read outside the Angular zone; on drift a rAF and a `setEmbeddedMpvBounds` IPC | Yes during playback: a position-only layout shift would otherwise leave the native view misplaced | Static | Cover with the throttling thread: it kept its rate while minimized under `backgroundThrottling: false`. |
@@ -92,6 +92,11 @@ export interface ExternalPlayerSession {
contentInfo?: PlayerContentInfo;
startedAt: string;
updatedAt: string;
/**
* Order in which sessions began in this app run. Unlike `startedAt` it
* tells apart launches made in the same millisecond.
*/
launchOrder?: number;
/** Raw failure detail; the dock translates `errorCode` instead. */
error?: string;
errorCode?: ExternalPlayerErrorCode;
-1
View File
@@ -13,7 +13,6 @@ export const maxLinesBaseline = [
'apps/electron-backend/src/app/database/operations/content.operations.ts',
'apps/electron-backend/src/app/database/operations/playlist.operations.ts',
'apps/electron-backend/src/app/events/epg-query.service.ts',
'apps/electron-backend/src/app/events/mpv-session.service.ts',
'apps/electron-backend/src/app/events/vlc-session.service.ts',
'apps/electron-backend/src/app/services/app-update.service.ts',
'apps/electron-backend/src/app/services/embedded-mpv-native.service.ts',