refactor(core): Simplify Redis client types (no-changelog) (#10397)

This commit is contained in:
Iván Ovejero 2024-08-15 10:41:49 +02:00 committed by GitHub
parent 613cdd2ba2
commit 60b15e16cb
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 27 additions and 20 deletions

View file

@ -45,7 +45,7 @@ export class CacheService extends TypedEmitter<CacheEvents> {
);
const redisClient = redisClientService.createClient({
type: 'client(cache)',
type: 'cache(n8n)',
extraOptions: { keyPrefix: prefix },
});

View file

@ -4,20 +4,7 @@ import { Service } from 'typedi';
import config from '@/config';
import { Logger } from '@/Logger';
import { RedisClientService } from './redis-client.service';
export type RedisClientType =
| 'subscriber'
| 'client'
| 'bclient'
| 'subscriber(bull)'
| 'client(bull)'
| 'bclient(bull)'
| 'client(cache)'
| 'publisher'
| 'consumer'
| 'producer'
| 'list-sender'
| 'list-receiver';
import type { RedisClientType } from './redis.types';
export type RedisServiceMessageHandler =
| ((channel: string, message: string) => void)
@ -34,7 +21,7 @@ class RedisServiceBase {
private readonly redisClientService: RedisClientService,
) {}
async init(type: RedisClientType = 'client'): Promise<void> {
async init(type: RedisClientType): Promise<void> {
if (this.redisClient && this.isInitialized) {
return;
}
@ -62,7 +49,7 @@ class RedisServiceBase {
export abstract class RedisServiceBaseSender extends RedisServiceBase {
senderId: string;
async init(type: RedisClientType = 'client'): Promise<void> {
async init(type: RedisClientType): Promise<void> {
await super.init(type);
this.senderId = config.get('redis.queueModeId');
}

View file

@ -9,7 +9,7 @@ import { RedisServiceBaseSender } from './RedisServiceBaseClasses';
@Service()
export class RedisServicePubSubPublisher extends RedisServiceBaseSender {
async init(): Promise<void> {
await super.init('publisher');
await super.init('publisher(n8n)');
}
async publish(channel: string, message: string): Promise<void> {

View file

@ -5,7 +5,7 @@ import { RedisServiceBaseReceiver } from './RedisServiceBaseClasses';
@Service()
export class RedisServicePubSubSubscriber extends RedisServiceBaseReceiver {
async init(): Promise<void> {
await super.init('subscriber');
await super.init('subscriber(n8n)');
this.redisClient?.on('message', (channel: string, message: string) => {
this.messageHandlers.forEach((handler: (channel: string, message: string) => void) =>

View file

@ -2,7 +2,8 @@ import { Service } from 'typedi';
import { Logger } from '@/Logger';
import ioRedis from 'ioredis';
import type { Cluster, RedisOptions } from 'ioredis';
import type { RedisClientType } from './RedisServiceBaseClasses';
import type { RedisClientType } from './redis.types';
import { OnShutdown } from '@/decorators/OnShutdown';
import { LOWEST_SHUTDOWN_PRIORITY } from '@/constants';
import { GlobalConfig } from '@n8n/config';

View file

@ -0,0 +1,19 @@
export type RedisClientType = N8nRedisClientType | BullRedisClientType;
/**
* Redis client used by n8n.
*
* - `subscriber(n8n)` to listen for messages from scaling mode communication channels
* - `publisher(n8n)` to send messages into scaling mode communication channels
* - `cache(n8n)` for caching operations (variables, resource ownership, etc.)
*/
type N8nRedisClientType = 'subscriber(n8n)' | 'publisher(n8n)' | 'cache(n8n)';
/**
* Redis client used internally by Bull. Suffixed with `(bull)` at `ScalingService.setupQueue`.
*
* - `subscriber(bull)` for event listening
* - `client(bull)` for general queue operations
* - `bclient(bull)` for blocking operations when processing jobs
*/
type BullRedisClientType = 'subscriber(bull)' | 'client(bull)' | 'bclient(bull)';