refactor(epg): implement streaming SAX-based EPG parser and batching

Replace full-file parsing with a streaming SAX parser to avoid loading
entire XMLTV payloads into memory. Add support for gzipped feeds by
using streaming gunzip (createGunzip) and piping into a SaxesParser,
allowing incremental processing of channels and programmes.

Introduce:
- StreamingEpgParser class that incrementally builds channel and program
  objects and emits batches.
- parseXmltvDate helper to normalize XMLTV datetimes to ISO strings.
- batching constants and response types for channel/program batches and
  progress reporting.
- use of Readable/Saxes for streaming parsing and reduced memory use.

This change improves performance and memory usage for large EPG files,
enables progress reporting and batch emission to the main thread, and
prepares for handling very large or gzipped EPG sources.
This commit is contained in:
4gray committed 2025-12-26 01:32:51 +01:00
1 parent a2240fcc88
commit fe8e907e9f
2 files changed
+459 -118

No files matched your search

@@ -1,6 +1,11 @@
import { app, ipcMain } from 'electron';
import * as path from 'path';
import { EpgChannelWithPrograms, EpgData, EpgProgram } from 'shared-interfaces';
import {
EpgChannel,
EpgChannelWithPrograms,
EpgData,
EpgProgram,
} from 'shared-interfaces';
import { Worker } from 'worker_threads';
import { pathToFileURL } from 'url';
@@ -117,7 +122,13 @@ export default class EpgEvents {
try {
// Worker threads require file:// URLs wrapped in URL object for packaged apps
const workerURL = pathToFileURL(workerPath);
worker = new Worker(workerURL);
worker = new Worker(workerURL, {
// Increase memory limits for large EPG files
resourceLimits: {
maxOldGenerationSizeMb: 4096, // 4GB for parsed data
maxYoungGenerationSizeMb: 512, // 512MB for temporary allocations
},
});
} catch (error) {
console.error(
this.loggerLabel,
@@ -134,29 +145,64 @@ export default class EpgEvents {
'message',
(message: {
type: string;
data?: EpgData;
channels?: EpgChannel[];
programs?: EpgProgram[];
error?: string;
url?: string;
stats?: { totalChannels: number; totalPrograms: number };
}) => {
if (message.type === 'READY') {
worker.postMessage({ type: 'FETCH_EPG', url });
} else if (message.type === 'EPG_PARSED') {
if (message.data) {
this.mergeEpgData(message.data);
switch (message.type) {
case 'READY':
worker.postMessage({ type: 'FETCH_EPG', url });
break;
case 'EPG_CHANNELS_BATCH':
// Add channels incrementally
if (message.channels) {
this.addChannelsBatch(message.channels);
}
break;
case 'EPG_PROGRAMS_BATCH':
// Add programs incrementally
if (message.programs) {
this.addProgramsBatch(message.programs);
}
break;
case 'EPG_PROGRESS':
// Log progress for large files
if (message.stats) {
console.log(
this.loggerLabel,
`Progress: ${message.stats.totalChannels} channels, ${message.stats.totalPrograms} programs`
);
}
break;
case 'EPG_COMPLETE':
console.log(
this.loggerLabel,
`EPG parsing complete for ${url}:`,
message.stats
);
this.rebuildMergedData();
this.fetchedUrls.add(url);
}
worker.terminate();
this.workers.delete(url);
resolve();
} else if (message.type === 'EPG_ERROR') {
console.error(
this.loggerLabel,
'Worker error:',
message.error
);
worker.terminate();
this.workers.delete(url);
reject(new Error(message.error || 'Unknown error'));
worker.terminate();
this.workers.delete(url);
resolve();
break;
case 'EPG_ERROR':
console.error(
this.loggerLabel,
'Worker error:',
message.error
);
worker.terminate();
this.workers.delete(url);
reject(new Error(message.error || 'Unknown error'));
break;
}
}
);
@@ -180,29 +226,21 @@ export default class EpgEvents {
}
/**
* Merge EPG data (optimized for large datasets)
* Add a batch of channels (streaming mode)
*/
private static mergeEpgData(newData: EpgData): void {
// Merge channels (avoid duplicates)
const existingChannelIds = new Set(
this.epgData.channels.map((c) => c.id)
);
const newChannels = newData.channels.filter(
(c) => !existingChannelIds.has(c.id)
);
// Use concat instead of spread for large arrays
private static addChannelsBatch(channels: EpgChannel[]): void {
// Use Set for O(1) duplicate checking
const existingIds = new Set(this.epgData.channels.map((c) => c.id));
const newChannels = channels.filter((c) => !existingIds.has(c.id));
this.epgData.channels = this.epgData.channels.concat(newChannels);
}
// Merge programs in chunks to avoid stack overflow
const chunkSize = 10000;
for (let i = 0; i < newData.programs.length; i += chunkSize) {
const chunk = newData.programs.slice(i, i + chunkSize);
this.epgData.programs = this.epgData.programs.concat(chunk);
}
// Rebuild merged data structure
this.rebuildMergedData();
/**
* Add a batch of programs (streaming mode)
*/
private static addProgramsBatch(programs: EpgProgram[]): void {
// Simply append programs - no duplicate check needed as each program is unique
this.epgData.programs = this.epgData.programs.concat(programs);
}
/**
@@ -247,7 +285,7 @@ export default class EpgEvents {
// If not found, try to find by display name
if (!channelData) {
for (const [id, channel] of this.epgDataMerged.entries()) {
for (const [, channel] of this.epgDataMerged.entries()) {
const displayNames = channel.displayName.map((d) =>
d.value.toLowerCase()
);
@@ -1,12 +1,13 @@
import { parse as parseEpg } from 'epg-parser';
import { EpgData, EpgProgram } from 'shared-interfaces';
import { EpgChannel, EpgProgram } from 'shared-interfaces';
import { parentPort } from 'worker_threads';
import { gunzipSync } from 'zlib';
import { createGunzip } from 'zlib';
import { SaxesParser, SaxesTagPlain } from 'saxes';
import { Readable } from 'stream';
/**
* EPG Parser Worker
* Runs in a separate thread to avoid blocking the main process
* Compatible with epg-parser v0.4.0
* Streaming EPG Parser Worker
* Uses SAX parsing to process XML incrementally without loading entire file into memory.
* Supports both regular and gzipped EPG files.
*/
interface WorkerMessage {
@@ -15,84 +16,395 @@ interface WorkerMessage {
}
interface WorkerResponse {
type: 'EPG_PARSED' | 'EPG_ERROR' | 'EPG_PROGRESS' | 'READY';
data?: EpgData;
type:
| 'EPG_CHANNELS_BATCH'
| 'EPG_PROGRAMS_BATCH'
| 'EPG_COMPLETE'
| 'EPG_ERROR'
| 'EPG_PROGRESS'
| 'READY';
channels?: EpgChannel[];
programs?: EpgProgram[];
error?: string;
url?: string;
stats?: {
totalChannels: number;
totalPrograms: number;
};
}
const loggerLabel = '[EPG Worker]';
// Batch size for sending data back to main thread
const CHANNEL_BATCH_SIZE = 100;
const PROGRAM_BATCH_SIZE = 1000;
/**
* Fetches and parses EPG data from URL
* Parse XMLTV datetime format to ISO string
* Format: YYYYMMDDHHmmss +HHMM or YYYYMMDDHHmmss
*/
async function fetchAndParseEpg(url: string): Promise<EpgData> {
try {
const isGzipped = url.endsWith('.gz');
function parseXmltvDate(dateStr: string): string {
if (!dateStr) return '';
// Use native fetch (available in Node.js 18+)
const response = await fetch(url.trim());
// Extract date parts: 20231225060000 +0000
const match = dateStr.match(
/^(\d{4})(\d{2})(\d{2})(\d{2})(\d{2})(\d{2})\s*([+-]\d{4})?$/
);
if (!response.ok) {
throw new Error(`HTTP error! status: ${response.status}`);
if (!match) return dateStr;
const [, year, month, day, hour, minute, second, tz] = match;
// Build ISO string
let isoString = `${year}-${month}-${day}T${hour}:${minute}:${second}`;
if (tz) {
// Convert +0000 to +00:00
isoString += `${tz.slice(0, 3)}:${tz.slice(3)}`;
} else {
isoString += 'Z';
}
return isoString;
}
/**
* Streaming EPG parser using SAX
*/
class StreamingEpgParser {
private parser: SaxesParser;
private channels: EpgChannel[] = [];
private programs: EpgProgram[] = [];
private totalChannels = 0;
private totalPrograms = 0;
// Current element being parsed
private currentChannel: Partial<EpgChannel> | null = null;
private currentProgram: Partial<EpgProgram> | null = null;
private currentTextContent = '';
private currentLang = '';
// For nested elements
private elementStack: string[] = [];
constructor(private onChannelsBatch: (channels: EpgChannel[]) => void, private onProgramsBatch: (programs: EpgProgram[]) => void, private onProgress: (channels: number, programs: number) => void) {
this.parser = new SaxesParser();
this.setupParser();
}
private setupParser(): void {
this.parser.on('opentag', (tag: SaxesTagPlain) => {
this.elementStack.push(tag.name);
this.currentTextContent = '';
switch (tag.name) {
case 'channel':
this.currentChannel = {
id: (tag.attributes['id'] as string) || '',
displayName: [],
icon: [],
url: [],
};
break;
case 'programme':
this.currentProgram = {
start: parseXmltvDate(
(tag.attributes['start'] as string) || ''
),
stop: parseXmltvDate(
(tag.attributes['stop'] as string) || ''
),
channel: (tag.attributes['channel'] as string) || '',
title: [],
desc: [],
category: [],
date: '',
episodeNum: [],
previouslyShown: [],
subtitles: [],
icon: [],
rating: [],
credits: [],
audio: [],
};
break;
case 'icon':
if (this.currentChannel) {
this.currentChannel.icon!.push({
src: (tag.attributes['src'] as string) || '',
width: tag.attributes['width']
? parseInt(tag.attributes['width'] as string)
: undefined,
height: tag.attributes['height']
? parseInt(tag.attributes['height'] as string)
: undefined,
});
} else if (this.currentProgram) {
this.currentProgram.icon!.push({
src: (tag.attributes['src'] as string) || '',
width: tag.attributes['width']
? parseInt(tag.attributes['width'] as string)
: undefined,
height: tag.attributes['height']
? parseInt(tag.attributes['height'] as string)
: undefined,
});
}
break;
case 'display-name':
case 'title':
case 'desc':
case 'category':
this.currentLang = (tag.attributes['lang'] as string) || '';
break;
case 'rating':
if (this.currentProgram) {
// Rating has system attribute, value comes as child <value> element
const system =
(tag.attributes['system'] as string) || '';
// We'll complete this when we see the value element
this.currentProgram.rating!.push({ system, value: '' });
}
break;
case 'episode-num':
if (this.currentProgram) {
const system =
(tag.attributes['system'] as string) || '';
this.currentProgram.episodeNum!.push({
system,
value: '',
});
}
break;
}
});
this.parser.on('text', (text: string) => {
this.currentTextContent += text;
});
this.parser.on('closetag', (tag: SaxesTagPlain) => {
const text = this.currentTextContent.trim();
if (this.currentChannel) {
switch (tag.name) {
case 'display-name':
this.currentChannel.displayName!.push({
lang: this.currentLang,
value: text,
});
break;
case 'url':
if (text) this.currentChannel.url!.push(text);
break;
case 'channel':
// Channel complete, add to batch
this.channels.push(this.currentChannel as EpgChannel);
this.totalChannels++;
this.currentChannel = null;
if (this.channels.length >= CHANNEL_BATCH_SIZE) {
this.flushChannels();
}
break;
}
}
if (this.currentProgram) {
switch (tag.name) {
case 'title':
this.currentProgram.title!.push({
lang: this.currentLang,
value: text,
});
break;
case 'desc':
this.currentProgram.desc!.push({
lang: this.currentLang,
value: text,
});
break;
case 'category':
this.currentProgram.category!.push({
lang: this.currentLang,
value: text,
});
break;
case 'date':
this.currentProgram.date = text;
break;
case 'value':
// This is for rating value
if (
this.elementStack.includes('rating') &&
this.currentProgram.rating!.length > 0
) {
this.currentProgram.rating![
this.currentProgram.rating!.length - 1
].value = text;
}
break;
case 'episode-num':
if (this.currentProgram.episodeNum!.length > 0) {
this.currentProgram.episodeNum![
this.currentProgram.episodeNum!.length - 1
].value = text;
}
break;
case 'programme':
// Program complete, add to batch
this.programs.push(this.currentProgram as EpgProgram);
this.totalPrograms++;
this.currentProgram = null;
if (this.programs.length >= PROGRAM_BATCH_SIZE) {
this.flushPrograms();
}
break;
}
}
this.elementStack.pop();
this.currentTextContent = '';
});
this.parser.on('error', (err: Error) => {
console.error(loggerLabel, 'Parser error:', err.message);
});
}
private flushChannels(): void {
if (this.channels.length > 0) {
this.onChannelsBatch([...this.channels]);
this.channels = [];
this.onProgress(this.totalChannels, this.totalPrograms);
}
}
let xmlString: string;
private flushPrograms(): void {
if (this.programs.length > 0) {
this.onProgramsBatch([...this.programs]);
this.programs = [];
this.onProgress(this.totalChannels, this.totalPrograms);
}
}
write(chunk: string): void {
this.parser.write(chunk);
}
finish(): { totalChannels: number; totalPrograms: number } {
this.parser.close();
// Flush any remaining data
this.flushChannels();
this.flushPrograms();
return {
totalChannels: this.totalChannels,
totalPrograms: this.totalPrograms,
};
}
}
/**
* Fetches and parses EPG data from URL using streaming
*/
async function fetchAndParseEpgStreaming(url: string): Promise<void> {
const isGzipped = url.endsWith('.gz');
console.log(
loggerLabel,
`Fetching EPG from ${url} (gzipped: ${isGzipped})`
);
const response = await fetch(url.trim());
if (!response.ok) {
throw new Error(`HTTP error! status: ${response.status}`);
}
if (!response.body) {
throw new Error('Response body is null');
}
const parser = new StreamingEpgParser(
(channels) => {
const response: WorkerResponse = {
type: 'EPG_CHANNELS_BATCH',
channels,
};
parentPort?.postMessage(response);
},
(programs) => {
const response: WorkerResponse = {
type: 'EPG_PROGRAMS_BATCH',
programs,
};
parentPort?.postMessage(response);
},
(totalChannels, totalPrograms) => {
const response: WorkerResponse = {
type: 'EPG_PROGRESS',
stats: { totalChannels, totalPrograms },
};
parentPort?.postMessage(response);
}
);
// Convert web stream to Node.js stream
const nodeStream = Readable.fromWeb(response.body as any);
return new Promise((resolve, reject) => {
let dataStream: Readable = nodeStream;
if (isGzipped) {
// For gzipped files, get as ArrayBuffer
const arrayBuffer = await response.arrayBuffer();
const buffer = Buffer.from(arrayBuffer);
const decompressed = gunzipSync(buffer);
xmlString = decompressed.toString('utf-8');
} else {
// For regular files, get as text
xmlString = await response.text();
const gunzip = createGunzip();
dataStream = nodeStream.pipe(gunzip);
gunzip.on('error', (err) => {
console.error(loggerLabel, 'Gunzip error:', err);
reject(err);
});
}
const parsed = parseEpg(xmlString);
dataStream.on('data', (chunk: Buffer) => {
try {
parser.write(chunk.toString('utf-8'));
} catch (err) {
console.error(loggerLabel, 'Parse error:', err);
reject(err);
}
});
// Normalize data to match our interface
const normalized: EpgData = {
channels: parsed.channels.map(
(channel: Record<string, unknown>) => ({
id: channel.id as string,
displayName:
(channel.displayName as {
lang: string;
value: string;
}[]) || [],
icon:
(channel.icon as {
src: string;
width?: number;
height?: number;
}[]) || [],
url: (channel.url as string[]) || [],
})
),
programs: parsed.programs.map(
(program: Record<string, unknown>) => ({
...program,
start: program.start as string, // Already ISO string in v0.4.0
stop: program.stop as string, // Already ISO string in v0.4.0
date:
typeof program.date === 'string'
? program.date
: (program.date as string[])?.[0] || '',
})
) as EpgProgram[],
};
dataStream.on('end', () => {
try {
const stats = parser.finish();
console.log(
loggerLabel,
`Parsing complete: ${stats.totalChannels} channels, ${stats.totalPrograms} programs`
);
return normalized;
} catch (error) {
console.error(
loggerLabel,
'Error fetching EPG:',
error instanceof Error ? error.message : String(error)
);
throw error;
}
const response: WorkerResponse = {
type: 'EPG_COMPLETE',
url,
stats,
};
parentPort?.postMessage(response);
resolve();
} catch (err) {
reject(err);
}
});
dataStream.on('error', (err) => {
console.error(loggerLabel, 'Stream error:', err);
reject(err);
});
});
}
/**
@@ -105,15 +417,7 @@ if (parentPort) {
message.type === 'FETCH_EPG' ||
message.type === 'FORCE_FETCH'
) {
const parsedData = await fetchAndParseEpg(message.url);
const response: WorkerResponse = {
type: 'EPG_PARSED',
data: parsedData,
url: message.url,
};
parentPort?.postMessage(response);
await fetchAndParseEpgStreaming(message.url);
}
} catch (error) {
console.error(loggerLabel, 'Worker error:', error);
@@ -122,7 +426,6 @@ if (parentPort) {
error: error instanceof Error ? error.message : String(error),
url: message.url,
};
parentPort?.postMessage(errorResponse);
}
});