mirror of https://github.com/nocodb/nocodb
Browse Source
* feat: clean-up job * feat: source cleanup * fix: remove unused method * feat: move jobs-redis to a static class * feat: release sources from in-memory db on update & delete * fix: skip calls if job redis not available * fix: error handling on connection delete --------- Co-authored-by: mertmit <mertmit99@gmail.com>pull/8557/head
Raju Udava
4 months ago
committed by
GitHub
9 changed files with 238 additions and 160 deletions
@ -1,91 +0,0 @@
|
||||
import { Injectable, Logger } from '@nestjs/common'; |
||||
import Redis from 'ioredis'; |
||||
import { InstanceTypes } from '~/interface/Jobs'; |
||||
|
||||
@Injectable() |
||||
export class JobsRedisService { |
||||
protected logger = new Logger(JobsRedisService.name); |
||||
|
||||
private redisClient: Redis; |
||||
private redisSubscriber: Redis; |
||||
private unsubscribeCallbacks: { [key: string]: () => void } = {}; |
||||
|
||||
public primaryCallbacks: { [key: string]: (...args) => void } = {}; |
||||
public workerCallbacks: { [key: string]: (...args) => void } = {}; |
||||
|
||||
constructor() { |
||||
this.redisClient = new Redis(process.env.NC_REDIS_JOB_URL); |
||||
this.redisSubscriber = new Redis(process.env.NC_REDIS_JOB_URL); |
||||
|
||||
if (process.env.NC_WORKER_CONTAINER === 'true') { |
||||
this.redisSubscriber.subscribe(InstanceTypes.WORKER); |
||||
} else { |
||||
this.redisSubscriber.subscribe(InstanceTypes.PRIMARY); |
||||
} |
||||
|
||||
const onMessage = (channel, message) => { |
||||
const args = message.split(':'); |
||||
const command = args.shift(); |
||||
if (channel === InstanceTypes.WORKER) { |
||||
this.workerCallbacks[command] && this.workerCallbacks[command](...args); |
||||
} else if (channel === InstanceTypes.PRIMARY) { |
||||
this.primaryCallbacks[command] && |
||||
this.primaryCallbacks[command](...args); |
||||
} |
||||
}; |
||||
|
||||
this.redisSubscriber.on('message', onMessage); |
||||
} |
||||
|
||||
publish(channel: string, message: string | any) { |
||||
if (typeof message === 'string') { |
||||
this.redisClient.publish(channel, message); |
||||
} else { |
||||
try { |
||||
this.redisClient.publish(channel, JSON.stringify(message)); |
||||
} catch (e) { |
||||
this.logger.error(e); |
||||
} |
||||
} |
||||
} |
||||
|
||||
subscribe(channel: string, callback: (message: any) => void) { |
||||
this.redisSubscriber.subscribe(channel); |
||||
|
||||
const onMessage = (_channel, message) => { |
||||
try { |
||||
message = JSON.parse(message); |
||||
} catch (e) {} |
||||
callback(message); |
||||
}; |
||||
|
||||
this.redisSubscriber.on('message', onMessage); |
||||
this.unsubscribeCallbacks[channel] = () => { |
||||
this.redisSubscriber.unsubscribe(channel); |
||||
this.redisSubscriber.off('message', onMessage); |
||||
}; |
||||
} |
||||
|
||||
unsubscribe(channel: string) { |
||||
if (this.unsubscribeCallbacks[channel]) { |
||||
this.unsubscribeCallbacks[channel](); |
||||
delete this.unsubscribeCallbacks[channel]; |
||||
} |
||||
} |
||||
|
||||
workerCount(): Promise<number> { |
||||
return new Promise((resolve, reject) => { |
||||
this.redisClient.publish( |
||||
InstanceTypes.WORKER, |
||||
'count', |
||||
(error, numberOfSubscribers) => { |
||||
if (error) { |
||||
reject(0); |
||||
} else { |
||||
resolve(numberOfSubscribers); |
||||
} |
||||
}, |
||||
); |
||||
}); |
||||
} |
||||
} |
||||
@ -0,0 +1,155 @@
|
||||
import { Logger } from '@nestjs/common'; |
||||
import Redis from 'ioredis'; |
||||
import type { InstanceCommands } from '~/interface/Jobs'; |
||||
import { InstanceTypes } from '~/interface/Jobs'; |
||||
|
||||
export class JobsRedis { |
||||
private static initialized = false; |
||||
|
||||
public static available = process.env.NC_REDIS_JOB_URL ? true : false; |
||||
|
||||
protected static logger = new Logger(JobsRedis.name); |
||||
|
||||
private static redisClient: Redis; |
||||
private static redisSubscriber: Redis; |
||||
private static unsubscribeCallbacks: { [key: string]: () => Promise<void> } = |
||||
{}; |
||||
|
||||
public static primaryCallbacks: { |
||||
[key: string]: (...args) => Promise<void>; |
||||
} = {}; |
||||
public static workerCallbacks: { [key: string]: (...args) => Promise<void> } = |
||||
{}; |
||||
|
||||
static async init() { |
||||
if (this.initialized) { |
||||
return; |
||||
} |
||||
|
||||
if (!JobsRedis.available) { |
||||
return; |
||||
} |
||||
|
||||
this.initialized = true; |
||||
|
||||
this.redisClient = new Redis(process.env.NC_REDIS_JOB_URL); |
||||
this.redisSubscriber = new Redis(process.env.NC_REDIS_JOB_URL); |
||||
|
||||
if (process.env.NC_WORKER_CONTAINER === 'true') { |
||||
await this.redisSubscriber.subscribe(InstanceTypes.WORKER); |
||||
} else { |
||||
await this.redisSubscriber.subscribe(InstanceTypes.PRIMARY); |
||||
} |
||||
|
||||
const onMessage = async (channel, message) => { |
||||
const args = message.split(':'); |
||||
const command = args.shift(); |
||||
if (channel === InstanceTypes.WORKER) { |
||||
this.workerCallbacks[command] && |
||||
(await this.workerCallbacks[command](...args)); |
||||
} else if (channel === InstanceTypes.PRIMARY) { |
||||
this.primaryCallbacks[command] && |
||||
(await this.primaryCallbacks[command](...args)); |
||||
} |
||||
}; |
||||
|
||||
this.redisSubscriber.on('message', onMessage); |
||||
} |
||||
|
||||
static async publish(channel: string, message: string | any) { |
||||
if (!this.initialized) { |
||||
if (!JobsRedis.available) { |
||||
return; |
||||
} |
||||
|
||||
await this.init(); |
||||
} |
||||
|
||||
if (typeof message === 'string') { |
||||
await this.redisClient.publish(channel, message); |
||||
} else { |
||||
try { |
||||
await this.redisClient.publish(channel, JSON.stringify(message)); |
||||
} catch (e) { |
||||
this.logger.error(e); |
||||
} |
||||
} |
||||
} |
||||
|
||||
static async subscribe( |
||||
channel: string, |
||||
callback: (message: any) => Promise<void>, |
||||
) { |
||||
if (!this.initialized) { |
||||
if (!JobsRedis.available) { |
||||
return; |
||||
} |
||||
|
||||
await this.init(); |
||||
} |
||||
|
||||
await this.redisSubscriber.subscribe(channel); |
||||
|
||||
const onMessage = async (_channel, message) => { |
||||
try { |
||||
message = JSON.parse(message); |
||||
} catch (e) {} |
||||
await callback(message); |
||||
}; |
||||
|
||||
this.redisSubscriber.on('message', onMessage); |
||||
this.unsubscribeCallbacks[channel] = async () => { |
||||
await this.redisSubscriber.unsubscribe(channel); |
||||
this.redisSubscriber.off('message', onMessage); |
||||
}; |
||||
} |
||||
|
||||
static async unsubscribe(channel: string) { |
||||
if (!this.initialized) { |
||||
if (!JobsRedis.available) { |
||||
return; |
||||
} |
||||
|
||||
await this.init(); |
||||
} |
||||
|
||||
if (this.unsubscribeCallbacks[channel]) { |
||||
await this.unsubscribeCallbacks[channel](); |
||||
delete this.unsubscribeCallbacks[channel]; |
||||
} |
||||
} |
||||
|
||||
static async workerCount(): Promise<number> { |
||||
if (!this.initialized) { |
||||
if (!JobsRedis.available) { |
||||
return; |
||||
} |
||||
|
||||
await this.init(); |
||||
} |
||||
|
||||
return new Promise((resolve, reject) => { |
||||
this.redisClient.publish( |
||||
InstanceTypes.WORKER, |
||||
'count', |
||||
(error, numberOfSubscribers) => { |
||||
if (error) { |
||||
reject(0); |
||||
} else { |
||||
resolve(numberOfSubscribers); |
||||
} |
||||
}, |
||||
); |
||||
}); |
||||
} |
||||
|
||||
static async emitWorkerCommand(command: InstanceCommands, ...args: any[]) { |
||||
const data = `${command}${args.length ? `:${args.join(':')}` : ''}`; |
||||
await JobsRedis.publish(InstanceTypes.WORKER, data); |
||||
} |
||||
|
||||
static async emitPrimaryCommand(command: InstanceCommands, ...args: any[]) { |
||||
const data = `${command}${args.length ? `:${args.join(':')}` : ''}`; |
||||
await JobsRedis.publish(InstanceTypes.PRIMARY, data); |
||||
} |
||||
} |
||||
Loading…
Reference in new issue