Skip to content

fix: redis pub/sub and streaming response issue #6008

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yau-wd merged 9 commits into from
Mar 30, 2026
Merged

fix: redis pub/sub and streaming response issue #6008

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
b59664c
fix: unsubscribe redis pubsub when connection closes during predictio…
0xi4o Mar 18, 2026
196d512
update: add heartbeat to keep connection alive and start/stop heartbe…
0xi4o Mar 18, 2026
70be39b
refactor: redis pub/sub methods, add unsubscribe, await subscribe
0xi4o Mar 18, 2026
85593ab
fix: merge conflicts
0xi4o Mar 18, 2026
6ccc7a1
fix: redundant unsubscribe
0xi4o Mar 18, 2026
171bb69
fix: use map to store clients list
0xi4o Mar 19, 2026
9ff0582
Merge branch 'main' of github.com:FlowiseAI/Flowise into fix/streaming
0xi4o Mar 24, 2026
7459efa
Merge branch 'main' of github.com:FlowiseAI/Flowise into fix/streaming
0xi4o Mar 24, 2026
2f6a108
Merge branch 'main' into fix/streaming
yau-wd Mar 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 6 additions & 2 deletions packages/server/src/controllers/internal-predictions/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@ 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)
Expand All @@ -42,8 +43,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 (process.env.MODE === MODE.QUEUE) {
getRunningExpressApp().redisSubscriber.subscribe(chatId)
if (isQueueMode) {
await getRunningExpressApp().redisSubscriber.subscribe(chatId)
}

const apiResponse = await utilBuildChatflow(req, true)
Expand All @@ -54,6 +55,9 @@ const createAndStreamInternalPrediction = async (req: Request, res: Response, ne
}
next(error)
} finally {
if (isQueueMode && chatId) {
await getRunningExpressApp().redisSubscriber.unsubscribe(chatId)
}
sseStreamer.removeClient(chatId)
}
}
Expand Down
8 changes: 6 additions & 2 deletions packages/server/src/controllers/predictions/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,7 @@ 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')
Expand All @@ -72,8 +73,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 (process.env.MODE === MODE.QUEUE) {
getRunningExpressApp().redisSubscriber.subscribe(chatId)
if (isQueueMode) {
await getRunningExpressApp().redisSubscriber.subscribe(chatId)
}

const apiResponse = await predictionsServices.buildChatflow(req)
Expand All @@ -84,6 +85,9 @@ const createPrediction = async (req: Request, res: Response, next: NextFunction)
}
next(error)
} finally {
if (isQueueMode && chatId) {
await getRunningExpressApp().redisSubscriber.unsubscribe(chatId)
}
sseStreamer.removeClient(chatId)
}
} else {
Expand Down
3 changes: 3 additions & 0 deletions packages/server/src/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -128,6 +128,7 @@ export class App {

// Initialize SSE Streamer
this.sseStreamer = new SSEStreamer()
this.sseStreamer.startHeartbeat()
logger.info('🌊 [server]: SSE Streamer initialized successfully')

// Init Queues
Expand All @@ -148,6 +149,7 @@ export class App {

this.redisSubscriber = new RedisEventSubscriber(this.sseStreamer)
await this.redisSubscriber.connect()
this.redisSubscriber.startPeriodicCleanup()
logger.info('🔗 [server]: Redis event subscriber connected successfully')
}

Expand Down Expand Up @@ -356,6 +358,7 @@ export class App {

async stopApp() {
try {
this.sseStreamer.stopHeartbeat()
const removePromises: any[] = []
removePromises.push(this.telemetry.flush())
if (this.queueManager) {
Expand Down
3 changes: 3 additions & 0 deletions packages/server/src/queue/PredictionQueue.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,9 @@ 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
Expand Down
Loading
Loading