rusackas commented on code in PR #40866:
URL: https://github.com/apache/superset/pull/40866#discussion_r3375716361
##########
superset-websocket/src/index.ts:
##########
@@ -264,19 +264,31 @@ export const subscribeToGlobalStream = async (
};
/**
- * Callback function to process events received from a Redis Stream
+ * Callback function to process events received from a Redis Stream.
+ *
+ * For large batches the loop periodically yields to the Node.js event loop
+ * (via setImmediate) so that connection management, health checks and
+ * ping/pong handling are not starved while a burst of events is processed.
+ * The yield cadence is controlled by `eventYieldBatchSize` (0 disables it).
*/
-export const processStreamResults = (results: StreamResult[]): void => {
+export const processStreamResults = async (
+ results: StreamResult[],
+): Promise<void> => {
logger.debug(`events received: ${results}`);
- results.forEach(item => {
+ const { eventYieldBatchSize } = opts;
+ for (let i = 0; i < results.length; i += 1) {
+ if (eventYieldBatchSize > 0 && i > 0 && i % eventYieldBatchSize === 0) {
+ await new Promise(resolve => setImmediate(resolve));
Review Comment:
Good catch - this is a real regression introduced by the mid-batch yield.
Fixed by making `ListenerFunction` async-aware and awaiting the listener in
`subscribeToGlobalStream` (and `fetchRangeFromStream`) before advancing
`lastFirehoseId`, so batches are processed sequentially and sends stay in order
even when `processStreamResults` yields to the event loop under backlog.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
