Something went wrong. Try again.
[READ-ONLY] Mirror of https://github.com/FoxxMD/multi-scrobbler. Scrobble plays from multiple sources to multiple clients docs.multi-scrobbler.app
deezer docker jellyfin koito lastfm listenbrainz maloja mopidy mpris music music-assistant plex scrobble self-hosted spotify subsonic tautulli youtube-music
Something went wrong. Try again.
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125import { RockskyClient } from "@rocksky/sdk";import { RateLimiterMemory, type IRateLimiterOptions, RateLimiterQueue } from "rate-limiter-flexible";import { loggerNoop } from "../../MaybeLogger.ts";import type { Logger } from "@foxxmd/logging";import AbstractApiClient from "../AbstractApiClient.ts";import { ROCKSKY_URL, type RockskyApiClientConfig } from "./interfaces.ts";import type { Cacheable } from "cacheable";import type { AbstractApiOptions } from "../../infrastructure/Atomic.ts";import { getRoot } from "../../../ioc.ts";import { type CircuitBreakerProxy, ProxyWithCircuitBreaker } from "@foxxmd/load-balancer-proxy";import { maxRequestsPerSecond, normalizeWebAddress } from "../../../utils/NetworkUtils.ts";import { formatNumber } from "../../../../core/DataUtils.ts";import { ConsecutiveBreaker } from "cockatiel";import { SimpleError } from "../../errors/MSErrors.ts";
interface RockskyClientWrappedOptions { rate?: Partial<IRateLimiterOptions> logger?: Logger}
// https://stackoverflow.com/a/62561508type Append<I, T extends unknown[]> = [...T, I]type WrappedConstructorArgs = Append<RockskyClientWrappedOptions, ConstructorParameters<typeof RockskyClient>>;
export class RockskyClientWrapped extends RockskyClient {
public rateLimiterQueue: RateLimiterQueue; protected logger: Logger;
constructor(...args: WrappedConstructorArgs) { const [view, token, wrappedOpts] = args; super(view, token); const { rate: { points = 1, duration = 1 } = {}, logger = loggerNoop } = wrappedOpts;
this.logger = logger; this.rateLimiterQueue = new RateLimiterQueue(new RateLimiterMemory({ points, duration }), { maxQueueSize: 20 }); }
async matchSong(...args: Parameters<RockskyClient['matchSong']>): ReturnType<RockskyClient['matchSong']> {
const remainingTokens = await this.rateLimiterQueue.removeTokens(1); this.logger.trace(`Rate Tokens => Used 1 | Remaining ${remainingTokens}`);
return super.matchSong(...args); }}
export type RockskySingletonMap = Map<string, RockskyClientWrapped>;
export class RockskyClientPool extends AbstractApiClient { declare config: RockskyApiClientConfig; public rsProxy: CircuitBreakerProxy<RockskyClientWrapped> cache: Cacheable;
constructor(name: any, config: RockskyApiClientConfig, options: AbstractApiOptions & { cache?: Cacheable }) { const {apis = [{enable: true}]} = config; super('Rocksky API', name, {...config, apis}, options); this.cache = options.cache ?? getRoot().items.cache().cacheApi;
const rsMap = getRoot().items.rsMap();
const usedApis: RockskyClientWrapped[] = []; const hosts: string[] = []; for (const rsConfig of apis) { if ((rsConfig.enable ?? true) === false) { this.logger.verbose(`Not using config for ${rsConfig.url ?? ROCKSKY_URL} because it is disabled`); continue; } const { rate = {}, url } = rsConfig; const u = normalizeWebAddress((url ?? ROCKSKY_URL).toLocaleLowerCase()); hosts.push(u.url.hostname); const rs = rsMap.get(u.url.hostname); let points: number, duration: number; if (rs === undefined) { switch (u.url.hostname) { case 'api.rocksky.app': { points = rate.requests ?? 1000; duration = rate.perTime ?? 30; const reqRate = maxRequestsPerSecond(points, duration); if (reqRate > 33.4) { this.logger.warn(`Cannot use a rate greater than 33req/s for rocksky.app. Reverting to 33req/s | Given: ${formatNumber(reqRate)}req/s`); points = 1000; duration = 30; } } break; default: points = rate.requests ?? 1; duration = rate.perTime ?? 1; break; } const api = new RockskyClientWrapped(undefined, rsConfig.token, {rate: {points, duration}}); rsMap.set(u.url.hostname, api); usedApis.push(api); this.logger.verbose(`Created Rocksky API for ${u.url.hostname} with Rate Limit ${points}req/${duration}s`); } else { usedApis.push(rs); } } this.rsProxy = ProxyWithCircuitBreaker.create<RockskyClientWrapped>(usedApis,() => ({ halfOpenAfter: 30000, breaker: new ConsecutiveBreaker(3), onFailure: ({reason, duration}: {reason: any, duration: number}) => { this.logger.warn(new SimpleError(`Error occurred after ${duration}ms, will try next host`, {cause: reason, shortStack: true})); }, }), { comparer: async (a, b) => await b.rateLimiterQueue.getTokensRemaining() - await a.rateLimiterQueue.getTokensRemaining() }) this.logger.debug(`Rate limit prioritized API calls using hosts: ${hosts.join(' | ')}`); }
protected getIdentifier(): string { return 'RSAPI'; }
}