Spilum.Net/mtf-backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Replace initializeStreams with improved ensureGroup method

Browse Source
This commit is contained in:
Apher 2026-07-03 06:09:19 +00:00
parent d49f527cea
commit bf809a0309
2 changed files with 14 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
src/models/index.js
View File

@ -49,48 +49,21 @@ redis.client = Redis.createClient({
url: `redis://default@${redisConfig.REDIS_HOST}:${redisConfig.REDIS_PORT}` url: `redis://default@${redisConfig.REDIS_HOST}:${redisConfig.REDIS_PORT}`
}); });
redis.initializeStreams = async () => { redis.ensureGroup = async (stream, group) => {
redis.client.on('error', (err) => console.error('Redis error:', err));
try {
await redis.client.connect().catch(console.error); await redis.client.connect().catch(console.error);
console.log("Initializing Redis Streams...");
// Create transcriptions streams consumer group
try { try {
await redis.client.xGroupCreate("communication", "mtf-devices", '0', {MKSTREAM: true}); await redis.client.xGroupCreate(stream, group, "0", {MKSTREAM: true});
console.log("Created consumer group: mtf-devices (communication)"); console.log(`Created consumer group: ${group} (${stream})`);
} catch (err) { } catch (err) {
if (err.message.includes('BUSYGROUP')) { if (err.message.includes('BUSYGROUP')) {
console.log("Consumer group already exists: mtf-devices"); console.log(`Consumer group already exists: ${group} (${stream})`);
} else { } else {
console.error(err); console.error(err);
} }
} }
const deviceIds = [config.DEVICE_ID];
for (const deviceId of deviceIds) {
const groupName = `${deviceId}_group`;
try {
await redis.client.xGroupCreate('communication', groupName, '0', {MKSTREAM: true});
console.log(`Created consumer group: ${groupName} (communication)`);
} catch (err) {
if (err.message.includes('BUSYGROUP')) {
console.log(`Consumer group already exists: ${groupName}`);
} else {
console.error(err);
}
}
}
console.log("Streams initialized successfully");
await redis.client.quit(); await redis.client.quit();
} catch (err) {
console.error("Error initializing streams:", err.message);
}
} }
module.exports = { module.exports = {

6
src/server.js
View File

@ -17,8 +17,10 @@ const Role = db.role;
const authConfig = require("./config/auth.config"); const authConfig = require("./config/auth.config");
const config = require("./config"); const config = require("./config");
const createApp = () => { const createApp = async () => {
redis.initializeStreams(); await redis.ensureGroup("speech:commands", `${config.DEVICE_ID}-cg`);
await redis.ensureGroup("speech:events", `${config.DEVICE_ID}-cg`);
const app = express(); const app = express();
// Security headers // Security headers