Elizhium/litlyx
1
0
Fork 0
mirror of https://github.com/Litlyx/litlyx synced 2025-12-10 07:48:37 +01:00
Code Issues Packages Projects Releases Wiki Activity

.

Browse Source
This commit is contained in:
Emily
2024-06-25 00:52:12 +02:00
parent 1414451d2b
commit f369880ba7
5 changed files with 37 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
shared/services/RedisStreamService.ts
View File

@@ -7,7 +7,8 @@ export type ReadingLoopOptions = {
empty?: number
},
readBlock?: number,
streamName: string
streamName: string,
consumer?: string
}
export class RedisStreamService {
@@ -23,7 +24,7 @@ export class RedisStreamService {
}
private static async readingLoop(options: ReadingLoopOptions, processFunction: (content: Record<string, string>) => Promise<any>) {
const result = await this.readFromStream(options.streamName, options.readBlock || 2500);
const result = await this.readFromStream(options.streamName, options.readBlock || 2500, options.consumer || 'base_consumer');
if (!result) {
await new Promise(r => setTimeout(r, options.delay?.empty || 5000));
setTimeout(() => this.readingLoop(options, processFunction), 1);
@@ -35,17 +36,29 @@ export class RedisStreamService {
return;
}
static startReadingLoop(options: ReadingLoopOptions, processFunction: (content: Record<string, string>) => Promise<any>) {
static async startReadingLoop(options: ReadingLoopOptions, processFunction: (content: Record<string, string>) => Promise<any>) {
try {
await this.client.xGroupCreate(options.streamName, 'broker', '0', { MKSTREAM: true, });
} catch (ex) {
}
this.readingLoop(options, processFunction)
}
private static async readFromStream(streamName: string, readBlock: number) {
const result = await this.client.xRead({ id: '0', key: streamName }, { COUNT: 1, BLOCK: readBlock });
private static async readFromStream(streamName: string, readBlock: number, consumer: string) {
const result = await this.client.xReadGroup(
'broker', consumer,
[{ key: streamName, id: '>' }],
{ COUNT: 1, BLOCK: readBlock }
);
if (!result) return;
if (result.length == 0) return;
if (!result[0].messages) return;
if (result[0].messages.length == 0) return;
const message = result[0].messages[0];
await this.client.xAck(streamName, 'broker', message.id);
await this.client.xDel(streamName, message.id);
const content = message.message;
return content;