+5 -1

src/backend/ioc.ts

Reviewed

··· 14 14 import prom, { Counter, Gauge } from 'prom-client'; 15 15 import { CoverArtApiClient } from "./common/vendor/musicbrainz/CoverArtApiClient.js"; 16 16 import { version } from "./version.js"; 17 17 + import { StaggerOptions } from "./utils/AsyncUtils.js"; 17 18 18 19 let root: ReturnType<typeof createRoot>; 19 20 export interface RootOptions { ··· 26 27 cache?: CacheConfigOptions | MSCache | (() => MSCache) 27 28 mbMap?: MusicBrainzSingletonMap | (() => MusicBrainzSingletonMap) 28 29 transformers?: TransformerCommonConfig[] 30 30 + staggerOptions?: Partial<StaggerOptions> 29 31 } 30 32 31 33 const discovered = new prom.Counter({ ··· 60 62 cache, 61 63 mbMap, 62 64 transformers = [], 65 65 + staggerOptions, 63 66 } = options || {}; 64 67 const configDir = process.env.CONFIG_DIR || path.resolve(projectDir, `./config`); 65 68 let disableWeb = dw; ··· 146 149 transformerManager, 147 150 cache: () => maybeSingletonCache !== undefined ? () => maybeSingletonCache : cacheFunc, 148 151 mbMap: () => maybeSingletonMb !== undefined ? () => maybeSingletonMb : mbFunc, 149 149 - coverArtApi 152 152 + coverArtApi, 153 153 + staggerOptions: staggerOptions ?? {}, 150 154 }).add((items) => { 151 155 const localUrl = generateBaseURL(baseUrl, items.port) 152 156 return {

+13 -6

src/backend/scrobblers/AbstractScrobbleClient.ts

Reviewed

··· 58 58 import { AsyncTask, SimpleIntervalJob, Task, ToadScheduler } from "toad-scheduler"; 59 59 import { getRoot } from "../ioc.js"; 60 60 import { rehydratePlay } from "../utils/CacheUtils.js"; 61 61 - import { findAsyncSequential, staggerMapper } from "../utils/AsyncUtils.js"; 61 61 + import { findAsyncSequential, staggerMapper, StaggerOptions } from "../utils/AsyncUtils.js"; 62 62 import pMap, { pMapIterable } from "p-map"; 63 63 import { comparePlayArtistsNormalized, comparePlayTracksNormalized, existingScrobble, ExistingScrobbleOpts } from "../utils/PlayComparisonUtils.js"; 64 64 import { lifecyclelessInvariantTransform } from "../../core/PlayUtils.js"; ··· 123 123 protected queuedGauge: Gauge; 124 124 protected deadLetterGauge: Gauge; 125 125 protected problemGauge: Gauge; 126 126 + 127 127 + protected staggerOpts: Partial<StaggerOptions>; 126 128 127 129 constructor(type: any, name: any, config: CommonClientConfig, notifier: Notifiers, emitter: EventEmitter, logger: Logger) { 128 130 super(config); ··· 187 189 transformPlay: this.transformPlay, 188 190 existingSubmitted: this.findExistingSubmittedPlayObj 189 191 } 190 190 - this.existingScrobble = (playObjPre: PlayObject, existingScrobbles: PlayObject[], log?: boolean) => existingScrobble(playObjPre, existingScrobbles, existingScrobbleOpts, log) 192 192 + this.existingScrobble = (playObjPre: PlayObject, existingScrobbles: PlayObject[], log?: boolean) => existingScrobble(playObjPre, existingScrobbles, existingScrobbleOpts, log); 193 193 + this.staggerOpts = getRoot().items.staggerOptions; 191 194 } 192 195 193 196 protected getIdentifier() { ··· 446 449 447 450 const playObj = await this.transformPlay(playObjPre, TRANSFORM_HOOK.candidate); 448 451 449 449 - const sm = staggerMapper<ScrobbledPlayObject, ScrobbledPlayObject>({concurrency: 2}); 452 452 + const sm = staggerMapper<ScrobbledPlayObject, ScrobbledPlayObject>({...this.staggerOpts, concurrency: 2}); 450 453 const dtInvariantMatches = (await pMap(this.scrobbledPlayObjs.data, sm(async x => ({...x, play: await this.transformPlay(x.play, TRANSFORM_HOOK.existing)})), {concurrency: 2})) 451 454 .filter(x => playObjDataMatch(playObj, x.play)); 452 455 ··· 848 851 849 852 queueScrobble = async (data: PlayObject | PlayObject[], source: string) => { 850 853 const plays = (Array.isArray(data) ? data : [data]).map(x => ({...x, meta: {...x.meta, seenAt: dayjs()}})); 851 851 - const sm = staggerMapper<PlayObject, PlayObject>({concurrency: 2}); 854 854 + const sm = staggerMapper<PlayObject, PlayObject>({...this.staggerOpts, concurrency: 2}); 852 855 for await(const play of pMapIterable(plays, sm(async x => await this.transformPlay(x, TRANSFORM_HOOK.preCompare)), {concurrency: 2})) { 853 856 try { 854 857 const existingQueued = await this.existingScrobble(play, this.queuedScrobbles.map(x => x.play), false); ··· 1004 1007 1005 1008 protected updateQueuedScrobblesCache = () => { 1006 1009 this.cache.cacheScrobble.set(`${this.getMachineId()}-queue`, this.queuedScrobbles) 1007 1007 - .then(() => null) 1008 1008 - .catch((e) => this.logger.warn(new Error('Error while updating queued scrobble cache', {cause: e}))); 1010 1010 + .then(() => { 1011 1011 + return undefined; 1012 1012 + }) 1013 1013 + .catch((e) => { 1014 1014 + this.logger.warn(new Error('Error while updating queued scrobble cache', {cause: e})) 1015 1015 + }); 1009 1016 } 1010 1017 } 1011 1018

+6 -3

src/backend/sources/AbstractSource.ts

Reviewed

··· 40 40 import { WebhookPayload } from '../common/infrastructure/config/health/webhooks.js'; 41 41 import { messageWithCauses, messageWithCausesTruncatedDefault } from '../utils/ErrorUtils.js'; 42 42 import { genericSourcePlayMatch } from '../utils/PlayComparisonUtils.js'; 43 43 - import { findAsync, staggerMapper } from '../utils/AsyncUtils.js'; 43 43 + import { findAsync, staggerMapper, StaggerOptions } from '../utils/AsyncUtils.js'; 44 44 import pMap, {pMapIterable} from 'p-map'; 45 45 import prom, { Counter, Gauge } from 'prom-client'; 46 46 import { normalizeStr } from '../utils/StringUtils.js'; ··· 94 94 95 95 protected discoveredCounter: Counter; 96 96 97 97 + protected staggerOpts: Partial<StaggerOptions>; 98 98 + 97 99 constructor(type: SourceType, name: string, config: SourceConfig, internal: InternalConfig, emitter: EventEmitter) { 98 100 super(config); 99 101 const {clients = [] } = config; ··· 110 112 this.emitter = emitter; 111 113 112 114 this.discoveredCounter = getRoot().items.sourceMetics.discovered; 115 115 + this.staggerOpts = getRoot().items.staggerOptions; 113 116 } 114 117 115 118 protected getIdentifier() { ··· 219 222 discover = async (plays: PlayObject[], options: { checkAll?: boolean, [key: string]: any } = {}): Promise<PlayObject[]> => { 220 223 const newDiscoveredPlays: PlayObject[] = []; 221 224 222 222 - const sm = staggerMapper<PlayObject, PlayObject>({concurrency: 2}); 225 225 + const sm = staggerMapper<PlayObject, PlayObject>({...this.staggerOpts, concurrency: 2}); 223 226 for await(const play of pMapIterable(plays, sm(async x => await this.transformPlay(x, TRANSFORM_HOOK.preCompare)), {concurrency: 2})) { 224 227 if(!(await this.alreadyDiscovered(play, options))) { 225 228 this.addPlayToDiscovered(play); ··· 251 254 return; 252 255 } 253 256 newDiscoveredPlays.sort(sortByOldestPlayDate); 254 254 - const sm = staggerMapper<PlayObject, PlayObject>({concurrency: 2}); 257 257 + const sm = staggerMapper<PlayObject, PlayObject>({...this.staggerOpts, concurrency: 2}); 255 258 this.emitter.emit('discoveredToScrobble', { 256 259 data: await pMap(newDiscoveredPlays, sm(async (x) => await this.transformPlay(x, TRANSFORM_HOOK.postCompare)), {concurrency: 2}), 257 260 options: {

+2 -2

src/backend/tests/scrobbler/TestScrobbler.ts

Reviewed

··· 1 1 - import { loggerTest } from "@foxxmd/logging"; 2 1 import EventEmitter from "events"; 3 2 import request from "superagent"; 4 3 import { PlayObject } from "../../../core/Atomic.js"; ··· 7 6 import { CommonClientConfig, CommonClientOptions, NowPlayingOptions } from "../../common/infrastructure/config/client/index.js"; 8 7 import clone from "clone"; 9 8 import { TimeRangeListensFetcher } from "../../common/infrastructure/Atomic.js"; 9 9 + import { loggerNoop } from "../../common/MaybeLogger.js"; 10 10 11 11 export class TestScrobbler extends AbstractScrobbleClient { 12 12 ··· 14 14 getScrobblesForTimeRange: TimeRangeListensFetcher; 15 15 16 16 constructor(config: CommonClientConfig = {name: 'test'}) { 17 17 - const logger = loggerTest; 17 17 + const logger = loggerNoop; 18 18 const notifier = new Notifiers(new EventEmitter(), new EventEmitter(), new EventEmitter(), logger); 19 19 super('test', 'Test', {name: 'test', ...config}, notifier, new EventEmitter(), logger); 20 20 this.supportsNowPlaying = false;

+1 -1

src/backend/tests/setup.ts

Reviewed

··· 2 2 import { getRoot } from "../ioc.js"; 3 3 import { transientCache } from './utils/CacheTestUtils.js'; 4 4 5 5 - const root = getRoot({cache: transientCache, logger: loggerTest}); 5 5 + const root = getRoot({cache: transientCache, logger: loggerTest, staggerOptions: {initialInterval: 1, maxRandomStagger: 1}}); 6 6 root.items.cache().init();

+1 -1

src/backend/tests/utils/CacheTestUtils.ts

Reviewed

··· 1 1 import { loggerTest } from "@foxxmd/logging"; 2 2 import { MSCache } from "../../common/Cache.js"; 3 3 4 4 - export const transientCache = () => new MSCache(loggerTest, {scrobble: {provider: 'memory'}}); 4 4 + export const transientCache = () => new MSCache(loggerTest, {scrobble: {provider: 'memory'}, auth: {provider: 'memory'}});