refactor(source): Move heartbeat responsibility into source class · foxxmd.dev/multi-scrobbler@75c0230

+17 -10

src/backend/index.ts

··· 16 16 import { getRoot } from "./ioc.js"; 17 17 import { parseVersion } from "./version.js"; 18 18 import { initServer } from "./server/index.js"; 19 - import { createHeartbeatClientsTask } from "./tasks/heartbeatClients.js"; 20 - import { createHeartbeatSourcesTask } from "./tasks/heartbeatSources.js"; 21 19 import { isDebugMode, parseBool, retry, sleep } from "./utils.js"; 22 20 import { readJson } from './utils/DataUtils.js'; 23 21 import ScrobbleClients from './scrobblers/ScrobbleClients.js'; ··· 157 155 } 158 156 159 157 for(const c of scrobbleClients.clients) { 160 - c.initHeartbeat(); 158 + c.initTasks(); 161 159 const res = await Promise.race([ 162 160 sleep(2200), 163 161 (async () => { ··· 168 166 })() 169 167 ]); 170 168 if(res === undefined) { 171 - logger.debug(`Not waiting for ${c.name} to finish init, moving on to the next client...`); 169 + logger.debug(`Not waiting for Client ${c.name} to finish init, moving on to the next Client...`); 172 170 } 173 171 } 174 172 175 - const sourceTask = createHeartbeatSourcesTask(scrobbleSources, logger); 176 - scheduler.addSimpleIntervalJob(new SimpleIntervalJob({ 177 - minutes: 20, 178 - runImmediately: true 179 - }, sourceTask, {id: 'sources_heart'})); 180 - logger.debug('Added Source Heartbeat task to scheduler'); 173 + for(const c of scrobbleSources.sources) { 174 + c.initTasks(); 175 + const res = await Promise.race([ 176 + sleep(2200), 177 + (async () => { 178 + while(!c.isReady()) { 179 + await sleep(400) 180 + } 181 + return true; 182 + })() 183 + ]); 184 + if(res === undefined) { 185 + logger.debug(`Not waiting for Source ${c.name} to finish init, moving on to the next Source...`); 186 + } 187 + } 181 188 182 189 let runRetentionNow = parseBool(process.env.RETENTION_IMMEDIATE, false); 183 190

+1 -1

src/backend/scrobblers/AbstractScrobbleClient.ts

··· 235 235 await this.tryStopScrobbling(); 236 236 } 237 237 238 - public initHeartbeat(opts: {deadDelay?: number} = {}) { 238 + public initTasks(opts: {deadDelay?: number} = {}) { 239 239 if(this.scheduler.existsById('heartbeat') === false) { 240 240 this.logger.info('Adding Heartbeat Task and running immediately'); 241 241 this.scheduler.addSimpleIntervalJob(new SimpleIntervalJob({

+55

src/backend/sources/AbstractSource.ts

··· 48 48 import { AbortedError, generateLoggableAbortReason } from '../common/errors/MSErrors.js'; 49 49 import { DrizzlePlayRepository, playToRepositoryCreatePlayOpts, queryArgsFromRequest, QueryPlaysOpts, RequestPlayQuery } from '../common/database/drizzle/repositories/PlayRepository.js'; 50 50 import { asPlay } from '../../core/PlayMarshalUtils.js'; 51 + import { AsyncTask, SimpleIntervalJob, ToadScheduler } from 'toad-scheduler'; 51 52 52 53 export interface RecentlyPlayedOptions { 53 54 limit?: number ··· 90 91 91 92 manualListening?: boolean 92 93 94 + scheduler: ToadScheduler = new ToadScheduler(); 93 95 emitter: EventEmitter; 94 96 95 97 protected SCROBBLE_BACKLOG_COUNT: number = 30; ··· 144 146 if(this.canPoll) { 145 147 await this.tryStopPolling('Source is being disposed'); 146 148 } 149 + } 150 + 151 + public initTasks() { 152 + if(this.scheduler.existsById('heartbeat') === false) { 153 + this.logger.info('Adding Heartbeat Task and running immediately'); 154 + this.scheduler.addSimpleIntervalJob(new SimpleIntervalJob({ 155 + minutes: 20, 156 + runImmediately: true 157 + }, new AsyncTask( 158 + 'Heartbeat', 159 + (): Promise<any> => { 160 + return this.heartbeatTask().then(() => null).catch((err) => { 161 + this.logger.error(err); 162 + }); 163 + }, 164 + (err: Error) => { 165 + this.logger.error(err); 166 + } 167 + ), {id: 'heartbeat'})); 168 + } else { 169 + this.logger.warn('Heartbeat task is already added to scheduler.'); 170 + } 171 + } 172 + 173 + protected async heartbeatTask(): Promise<boolean> { 174 + if(!this.isReady()) { 175 + if(!this.canAuthUnattended()) { 176 + this.logger.warn({labels: 'Heartbeat'}, 'Source is not ready but will not try to initialize because auth state is not good and cannot be corrected unattended.') 177 + return false; 178 + } 179 + try { 180 + await this.tryInitialize({force: false, notify: true, notifyTitle: 'Could not initialize automatically'}); 181 + } catch (e) { 182 + this.logger.error(new Error('Could not initialize automatically', {cause: e})); 183 + return false; 184 + } 185 + 186 + if('discoverDevices' in this && typeof this.discoverDevices === 'function') { 187 + this.discoverDevices(); 188 + } 189 + 190 + if (this.canPoll && !this.polling) { 191 + if(!this.canAuthUnattended()) { 192 + this.logger.warn({labels: 'Heartbeat'}, 'Should be polling but will not attempt to start because auth state is not good and cannot be correct unattended.'); 193 + return false; 194 + } else { 195 + this.logger.info({labels: 'Heartbeat'}, 'Should be polling, attempting to start polling...'); 196 + this.poll({force: false, notify: true}).catch(e => this.logger.error(e)); 197 + } 198 + return true; 199 + } 200 + } 201 + return true; 147 202 } 148 203 149 204 protected async postCache(): Promise<void> {

-59

src/backend/tasks/heartbeatClients.ts

··· 1 - import { childLogger, Logger } from '@foxxmd/logging'; 2 - import { PromisePool } from "@supercharge/promise-pool"; 3 - import { AsyncTask } from "toad-scheduler"; 4 - import ScrobbleClients from "../scrobblers/ScrobbleClients.js"; 5 - 6 - export const createHeartbeatClientsTask = (clients: ScrobbleClients, parentLogger: Logger) => { 7 - const logger = childLogger(parentLogger, ['Heartbeat', 'Clients']); 8 - 9 - return new AsyncTask( 10 - 'Heartbeat', 11 - (): Promise<any> => { 12 - logger.verbose('Starting check...'); 13 - return PromisePool 14 - .withConcurrency(1) 15 - .for(clients.clients) 16 - .process(async (client) => { 17 - if(!client.isReady()) { 18 - if(!client.canAuthUnattended()) { 19 - client.logger.warn({labels: 'Heartbeat'}, 'Client is not ready but will not try to initialize because auth state is not good and cannot be correct unattended.') 20 - return 0; 21 - } 22 - try { 23 - await client.tryInitialize({force: false, notify: true, notifyTitle: 'Could not initialize automatically'}); 24 - } catch (e) { 25 - client.logger.error(new Error('Could not initialize automatically', {cause: e})); 26 - return 1; 27 - } 28 - } 29 - 30 - if(!client.canAuthUnattended()) { 31 - client.logger.warn({label: 'Heartbeat'}, 'Should be monitoring scrobbles but will not attempt to start because auth state is not good and cannot be correct unattended.'); 32 - return 0; 33 - } 34 - 35 - await client.processDeadLetterQueue(); 36 - if(!client.scrobbling) { 37 - client.logger.info({labels: 'Heartbeat'}, 'Should be processing scrobbles! Attempting to restart scrobbling...'); 38 - client.initScrobbleMonitoring(); 39 - return 1; 40 - } 41 - }).then(({results, errors}) => { 42 - logger.verbose(`Checked Dead letter queue for ${clients.clients.length} clients.`); 43 - const restarted = results.reduce((acc, curr) => acc += curr, 0); 44 - if (restarted > 0) { 45 - logger.info(`Attempted to start ${restarted} clients that were not processing scrobbles.`); 46 - } 47 - if (errors.length > 0) { 48 - logger.error(`Encountered errors!`); 49 - for (const err of errors) { 50 - logger.error(err); 51 - } 52 - } 53 - }); 54 - }, 55 - (err: Error) => { 56 - logger.error(err); 57 - } 58 - ); 59 - }

-65

src/backend/tasks/heartbeatSources.ts

··· 1 - import { childLogger, Logger } from '@foxxmd/logging'; 2 - import { PromisePool } from "@supercharge/promise-pool"; 3 - import { AsyncTask } from "toad-scheduler"; 4 - import { ChromecastSource } from "../sources/ChromecastSource.js"; 5 - import ScrobbleSources from "../sources/ScrobbleSources.js"; 6 - 7 - export const createHeartbeatSourcesTask = (sources: ScrobbleSources, parentLogger: Logger) => { 8 - const logger = childLogger(parentLogger, ['Heartbeat', 'Sources']); 9 - 10 - return new AsyncTask( 11 - 'Heartbeat', 12 - (): Promise<any> => { 13 - logger.verbose('Starting check...'); 14 - return PromisePool 15 - .withConcurrency(1) 16 - .for(sources.sources) 17 - .process(async (source) => { 18 - if(!source.isReady()) { 19 - if(!source.canAuthUnattended()) { 20 - source.logger.warn({label: 'Heartbeat'}, 'Source is not ready but will not try to initialize because auth state is not good and cannot be correct unattended.'); 21 - return 0; 22 - } 23 - try { 24 - await source.tryInitialize({force: false, notify: true, notifyTitle: 'Could not initialize automatically'}); 25 - } catch (e) { 26 - source.logger.error(new Error('Could not initialize source automatically', {cause: e})); 27 - return 1; 28 - } 29 - } 30 - 31 - if(source.type === 'chromecast') { 32 - (source as ChromecastSource).discoverDevices(); 33 - } 34 - 35 - if (source.canPoll && !source.polling) { 36 - if(!source.canAuthUnattended()) { 37 - source.logger.warn({label: 'Heartbeat'}, 'Should be polling but will not attempt to start because auth state is not good and cannot be correct unattended.'); 38 - return 0; 39 - } else { 40 - source.logger.info({label: 'Heartbeat'}, 'Should be polling, attempting to start polling...'); 41 - source.poll({force: false, notify: true}).catch(e => source.logger.error(e)); 42 - } 43 - return 1; 44 - } 45 - 46 - return 0; 47 - }).then(({results, errors}) => { 48 - logger.verbose(`Checked ${sources.sources.length} sources for start signals.`); 49 - const restarted = results.reduce((acc, curr) => acc += curr, 0); 50 - if (restarted > 0) { 51 - logger.info(`Attempted to start ${restarted} sources.`); 52 - } 53 - if (errors.length > 0) { 54 - logger.error(`Encountered errors!`); 55 - for (const err of errors) { 56 - logger.error(err); 57 - } 58 - } 59 - }); 60 - }, 61 - (err: Error) => { 62 - logger.error(err); 63 - } 64 - ); 65 - }

+2 -2

src/backend/tests/scrobbler/scrobblers.test.ts

··· 1040 1040 await using npScrobbler = new NowPlayingScrobbler(); 1041 1041 npScrobbler.nowPlayingTaskInterval = 10; 1042 1042 await npScrobbler.initialize(); 1043 - npScrobbler.initHeartbeat(); 1043 + npScrobbler.initTasks(); 1044 1044 1045 1045 await npScrobbler.queuePlayingNow(generateSourcePlayerObj({play:generatePlay({}, {deviceId: genGroupIdStr(generatePlayPlatformId())})}), {type: 'jellyfin', name: 'test'}); 1046 1046 ··· 1054 1054 await using npScrobbler = new NowPlayingScrobbler(); 1055 1055 npScrobbler.nowPlayingTaskInterval = 10; 1056 1056 await npScrobbler.initialize(); 1057 - npScrobbler.initHeartbeat(); 1057 + npScrobbler.initTasks(); 1058 1058 1059 1059 const now = dayjs(); 1060 1060

Configure Feed

Configure Feed