|
15 | 15 | // limitations under the License. |
16 | 16 |
|
17 | 17 | import EventEmitter from "events"; |
18 | | -import { OutboundEvent } from "../lib/interfaces"; |
| 18 | +import { IAckEvent, IConfig, OutboundEvent } from "../lib/interfaces"; |
19 | 19 | import { Logger } from "../lib/logger"; |
20 | 20 | import * as utils from '../lib/utils'; |
21 | 21 |
|
22 | 22 | const log = new Logger("handlers/events.ts") |
23 | 23 |
|
24 | | -let eventQueue: OutboundEvent[] = []; |
25 | | -export const eventEmitter = new EventEmitter(); |
| 24 | +let maxInflight = utils.constants.DEFAULT_MAX_INFLIGHT; |
| 25 | +let maxEventQueueSize = utils.constants.MAX_EVENT_QUEUE_SIZE; |
| 26 | +let eventQueue: OutboundEvent[]; |
| 27 | +let inFlight: OutboundEvent[]; |
26 | 28 |
|
27 | | -export const queueEvent = (socketEvent: OutboundEvent) => { |
28 | | - if(eventQueue.length < utils.constants.MAX_EVENT_QUEUE_SIZE) { |
29 | | - eventQueue.push(socketEvent); |
30 | | - if(eventQueue.length === 1) { |
31 | | - eventEmitter.emit('event', eventQueue[0]); |
| 29 | +let eventEmitter: EventEmitter; |
| 30 | +let unblockPromise: Promise<void> | undefined; |
| 31 | +let unblock: (() => void) | undefined; |
| 32 | + |
| 33 | +export const init = async (config: IConfig) => { |
| 34 | + eventEmitter = new EventEmitter(); |
| 35 | + eventQueue = []; |
| 36 | + inFlight = []; |
| 37 | + unblockPromise = undefined; |
| 38 | + unblock = undefined; |
| 39 | + if (config.events?.maxInflight !== undefined) { |
| 40 | + maxInflight = config.events.maxInflight; |
| 41 | + } |
| 42 | + if (config.events?.queueSize !== undefined) { |
| 43 | + maxEventQueueSize = config.events.queueSize; |
| 44 | + } |
| 45 | +} |
| 46 | + |
| 47 | +const dispatchNext = () => { |
| 48 | + if (inFlight.length < maxInflight) { |
| 49 | + const event = eventQueue.shift(); |
| 50 | + if (event) { |
| 51 | + inFlight.push(event) |
| 52 | + log.debug(`${event.id}: dispatched`); |
| 53 | + eventEmitter.emit('event', event); |
32 | 54 | } |
33 | | - } else { |
34 | | - log.warn('Max queue size reached'); |
35 | 55 | } |
36 | | -}; |
37 | 56 |
|
38 | | -export const handleCommit = () => { |
39 | | - eventQueue.shift(); |
40 | | - if(eventQueue.length > 0) { |
41 | | - eventEmitter.emit('event', eventQueue[0]); |
| 57 | + if (eventQueue.length < maxEventQueueSize && unblock) { |
| 58 | + unblock(); |
| 59 | + unblockPromise = undefined; |
| 60 | + unblock = undefined; |
| 61 | + log.info(`Event queue unblocked (length=${eventQueue.length})`); |
42 | 62 | } |
43 | 63 | } |
44 | 64 |
|
45 | | -export const getCurrentEvent = () => { |
46 | | - if(eventQueue.length > 0) { |
47 | | - return eventQueue[0]; |
| 65 | +export const queueEvent = async (socketEvent: OutboundEvent) => { |
| 66 | + |
| 67 | + let currentUnblockPromise = unblockPromise; |
| 68 | + if (currentUnblockPromise) { |
| 69 | + let blockedTime = Date.now(); |
| 70 | + log.warn(`${socketEvent.id}: delaying receive due to full event queue (length=${eventQueue.length})`); |
| 71 | + await currentUnblockPromise; |
| 72 | + log.info(`${socketEvent.id}: unblocked receive after ${Date.now()-blockedTime}ms`); |
48 | 73 | } |
49 | | -}; |
50 | 74 |
|
51 | | -export const getQueueSize = () => { |
52 | | - return eventQueue.length; |
| 75 | + eventQueue.push(socketEvent); |
| 76 | + if (eventQueue.length >= maxEventQueueSize && !unblockPromise) { |
| 77 | + log.warn(`Event queue became full (length=${eventQueue.length})`); |
| 78 | + unblockPromise = new Promise(resolve => { |
| 79 | + unblock = resolve; |
| 80 | + }) |
| 81 | + } |
| 82 | + |
| 83 | + dispatchNext(); |
53 | 84 | }; |
| 85 | + |
| 86 | +export const reDispatchInFlight = () => { |
| 87 | + for (const event of inFlight) { |
| 88 | + eventEmitter.emit('event', event) |
| 89 | + } |
| 90 | +} |
| 91 | + |
| 92 | +export const handleAck = (ack: IAckEvent) => { |
| 93 | + |
| 94 | + // Check we have something in-flight |
| 95 | + if (inFlight.length <= 0) { |
| 96 | + log.error(`Ack for ${ack.id} while no events in-flight`); |
| 97 | + return |
| 98 | + } |
| 99 | + |
| 100 | + // If no ID supplied (back-level API) we |
| 101 | + if (ack.id === undefined) { |
| 102 | + log.warn(`FireFly core is back-level and did not supply an event ID`); |
| 103 | + ack.id = inFlight[0].id; |
| 104 | + } |
| 105 | + |
| 106 | + // Remove from our in-flight map |
| 107 | + let event; |
| 108 | + for (let i = 0; i < inFlight.length; i++) { |
| 109 | + const candidate = inFlight[i] |
| 110 | + if (ack.id === candidate.id) { |
| 111 | + event = candidate; |
| 112 | + inFlight.splice(i, 1); |
| 113 | + break; |
| 114 | + } |
| 115 | + } |
| 116 | + if (!event) { |
| 117 | + log.warn(`Ack received for ${ack.id} which is not in-flight`); |
| 118 | + return |
| 119 | + } |
| 120 | + log.debug(`${ack.id}: acknowledged`); |
| 121 | + |
| 122 | + dispatchNext(); |
| 123 | +} |
| 124 | + |
| 125 | +export const getEmitter = () => { |
| 126 | + return eventEmitter; |
| 127 | +} |
| 128 | + |
| 129 | +export const getStats = () => { |
| 130 | + return { |
| 131 | + messageQueueSize: eventQueue.length, |
| 132 | + inFlightCount: inFlight.length, |
| 133 | + } |
| 134 | +} |
0 commit comments