Revert "fix: redis pub/sub and streaming response issue" (#6084)

Revert "fix: redis pub/sub and streaming response issue (#6008)"

This reverts commit 34aa82e.

Author: yau-wd

packages/server/src/controllers/internal-predictions/index.ts

@@ -33,7 +33,6 @@ const createInternalPrediction = async (req: Request, res: Response, next: NextF
 const createAndStreamInternalPrediction = async (req: Request, res: Response, next: NextFunction) => {
     const chatId = req.body.chatId
     const sseStreamer = getRunningExpressApp().sseStreamer
-    const isQueueMode = process.env.MODE === MODE.QUEUE
 
     try {
         sseStreamer.addClient(chatId, res)
@@ -43,8 +42,8 @@ const createAndStreamInternalPrediction = async (req: Request, res: Response, ne
         res.setHeader('X-Accel-Buffering', 'no') //nginx config: https://serverfault.com/a/801629
         res.flushHeaders()
 
-        if (isQueueMode) {
-            await getRunningExpressApp().redisSubscriber.subscribe(chatId)
+        if (process.env.MODE === MODE.QUEUE) {
+            getRunningExpressApp().redisSubscriber.subscribe(chatId)
         }
 
         const apiResponse = await utilBuildChatflow(req, true)
@@ -55,9 +54,6 @@ const createAndStreamInternalPrediction = async (req: Request, res: Response, ne
         }
         next(error)
     } finally {
-        if (isQueueMode && chatId) {
-            await getRunningExpressApp().redisSubscriber.unsubscribe(chatId)
-        }
         sseStreamer.removeClient(chatId)
     }
 }

packages/server/src/controllers/predictions/index.ts

@@ -64,7 +64,6 @@ const createPrediction = async (req: Request, res: Response, next: NextFunction)
                     chatId = req.body.chatId ?? req.body.overrideConfig?.sessionId ?? uuidv4()
                     req.body.chatId = chatId
                 }
-                const isQueueMode = process.env.MODE === MODE.QUEUE
                 try {
                     sseStreamer.addExternalClient(chatId, res)
                     res.setHeader('Content-Type', 'text/event-stream')
@@ -73,8 +72,8 @@ const createPrediction = async (req: Request, res: Response, next: NextFunction)
                     res.setHeader('X-Accel-Buffering', 'no') //nginx config: https://serverfault.com/a/801629
                     res.flushHeaders()
 
-                    if (isQueueMode) {
-                        await getRunningExpressApp().redisSubscriber.subscribe(chatId)
+                    if (process.env.MODE === MODE.QUEUE) {
+                        getRunningExpressApp().redisSubscriber.subscribe(chatId)
                     }
 
                     const apiResponse = await predictionsServices.buildChatflow(req)
@@ -85,9 +84,6 @@ const createPrediction = async (req: Request, res: Response, next: NextFunction)
                     }
                     next(error)
                 } finally {
-                    if (isQueueMode && chatId) {
-                        await getRunningExpressApp().redisSubscriber.unsubscribe(chatId)
-                    }
                     sseStreamer.removeClient(chatId)
                 }
             } else {

packages/server/src/index.ts

@@ -128,7 +128,6 @@ export class App {
 
             // Initialize SSE Streamer
             this.sseStreamer = new SSEStreamer()
-            this.sseStreamer.startHeartbeat()
             logger.info('🌊 [server]: SSE Streamer initialized successfully')
 
             // Init Queues
@@ -149,7 +148,6 @@ export class App {
 
                 this.redisSubscriber = new RedisEventSubscriber(this.sseStreamer)
                 await this.redisSubscriber.connect()
-                this.redisSubscriber.startPeriodicCleanup()
                 logger.info('🔗 [server]: Redis event subscriber connected successfully')
             }
 
@@ -358,7 +356,6 @@ export class App {
 
     async stopApp() {
         try {
-            this.sseStreamer.stopHeartbeat()
             const removePromises: any[] = []
             removePromises.push(this.telemetry.flush())
             if (this.queueManager) {

packages/server/src/queue/PredictionQueue.ts

@@ -63,9 +63,6 @@ export class PredictionQueue extends BaseQueue {
     }
 
     async processJob(data: IExecuteFlowParams | IGenerateAgentflowv2Params) {
-        if (this.redisPublisher) {
-            await this.redisPublisher.connect()
-        }
         if (this.appDataSource) data.appDataSource = this.appDataSource
         if (this.telemetry) data.telemetry = this.telemetry
         if (this.cachePool) data.cachePool = this.cachePool