import { Configuration } from '../configuration/Configuration'; import { ILog } from '../logging/ILog'; import { IEvent } from '../models/IEvent'; import { IEventQueue } from '../queue/IEventQueue'; import { IStorageItem } from '../storage/IStorageItem'; import { SubmissionResponse } from '../submission/SubmissionResponse'; export class DefaultEventQueue implements IEventQueue { /** * The configuration object. * @type {Configuration} * @private */ private _config: Configuration; /** * A list of handlers that will be fired when events are submitted. * @type {Array} * @private */ private _handlers: Array<(events: IEvent[], response: SubmissionResponse) => void> = []; /** * Suspends processing until the specified time. * @type {Date} * @private */ private _suspendProcessingUntil: Date; /** * Discards queued items until the specified time. * @type {Date} * @private */ private _discardQueuedItemsUntil: Date; /** * Returns true if the queue is processing. * @type {boolean} * @private */ private _processingQueue: boolean = false; /** * Processes the queue every xx seconds. * @type {Timer} * @private */ private _queueTimer: any; constructor(config: Configuration) { this._config = config; } public enqueue(event: IEvent): void { const eventWillNotBeQueued: string = 'The event will not be queued.'; // optimization for minifier. const config: Configuration = this._config; // Optimization for minifier. const log: ILog = config.log; // Optimization for minifier. if (!config.enabled) { log.info(`Configuration is disabled. ${eventWillNotBeQueued}`); return; } if (!config.isValid) { log.info(`Invalid Api Key. ${eventWillNotBeQueued}`); return; } if (this.areQueuedItemsDiscarded()) { log.info(`Queue items are currently being discarded. ${eventWillNotBeQueued}`); return; } this.ensureQueueTimer(); const timestamp = config.storage.queue.save(event); const logText = `type=${event.type} ${event.reference_id ? 'refid=' + event.reference_id : ''}`; if (timestamp) { log.info(`Enqueuing event: ${timestamp} ${logText}`); } else { log.error(`Could not enqueue event ${logText}`); } } public process(isAppExiting?: boolean): void { const queueNotProcessed: string = 'The queue will not be processed.'; // optimization for minifier. const config: Configuration = this._config; // Optimization for minifier. const log: ILog = config.log; // Optimization for minifier. if (this._processingQueue) { return; } log.info('Processing queue...'); if (!config.enabled) { log.info(`Configuration is disabled. ${queueNotProcessed}`); return; } if (!config.isValid) { log.info(`Invalid Api Key. ${queueNotProcessed}`); return; } this._processingQueue = true; this.ensureQueueTimer(); try { const events = config.storage.queue.get(config.submissionBatchSize); if (!events || events.length === 0) { this._processingQueue = false; return; } log.info(`Sending ${events.length} events to ${config.serverUrl}.`); config.submissionClient.postEvents(events.map((e) => e.value), config, (response: SubmissionResponse) => { this.processSubmissionResponse(response, events); this.eventsPosted(events.map((e) => e.value), response); log.info('Finished processing queue.'); this._processingQueue = false; }, isAppExiting); } catch (ex) { log.error(`Error processing queue: ${ex}`); this.suspendProcessing(); this._processingQueue = false; } } public suspendProcessing(durationInMinutes?: number, discardFutureQueuedItems?: boolean, clearQueue?: boolean): void { const config: Configuration = this._config; // Optimization for minifier. if (!durationInMinutes || durationInMinutes <= 0) { durationInMinutes = 5; } config.log.info(`Suspending processing for ${durationInMinutes} minutes.`); this._suspendProcessingUntil = new Date(new Date().getTime() + (durationInMinutes * 60000)); if (discardFutureQueuedItems) { this._discardQueuedItemsUntil = this._suspendProcessingUntil; } if (clearQueue) { // Account is over the limit and we want to ensure that the sample size being sent in will contain newer errors. config.storage.queue.clear(); } } public onEventsPosted(handler: (events: IEvent[], response: SubmissionResponse) => void): void { handler && this._handlers.push(handler); } private eventsPosted(events: IEvent[], response: SubmissionResponse) { const handlers = this._handlers; // optimization for minifier. for (const handler of handlers) { try { handler(events, response); } catch (ex) { this._config.log.error(`Error calling onEventsPosted handler: ${ex}`); } } } private areQueuedItemsDiscarded(): boolean { return this._discardQueuedItemsUntil && this._discardQueuedItemsUntil > new Date(); } private ensureQueueTimer(): void { if (!this._queueTimer) { this._queueTimer = setInterval(() => this.onProcessQueue(), 10000); } } private isQueueProcessingSuspended(): boolean { return this._suspendProcessingUntil && this._suspendProcessingUntil > new Date(); } private onProcessQueue(): void { if (!this.isQueueProcessingSuspended() && !this._processingQueue) { this.process(); } } private processSubmissionResponse(response: SubmissionResponse, events: IStorageItem[]): void { const noSubmission: string = 'The event will not be submitted.'; // Optimization for minifier. const config: Configuration = this._config; // Optimization for minifier. const log: ILog = config.log; // Optimization for minifier. if (response.success) { log.info(`Sent ${events.length} events.`); this.removeEvents(events); return; } if (response.serviceUnavailable) { // You are currently over your rate limit or the servers are under stress. log.error('Server returned service unavailable.'); this.suspendProcessing(); return; } if (response.paymentRequired) { // If the organization over the rate limit then discard the event. log.info('Too many events have been submitted, please upgrade your plan.'); this.suspendProcessing(null, true, true); return; } if (response.unableToAuthenticate) { // The api key was suspended or could not be authorized. log.info(`Unable to authenticate, please check your configuration. ${noSubmission}`); this.suspendProcessing(15); this.removeEvents(events); return; } if (response.notFound || response.badRequest) { // The service end point could not be found. log.error(`Error while trying to submit data: ${response.message}`); this.suspendProcessing(60 * 4); this.removeEvents(events); return; } if (response.requestEntityTooLarge) { const message = 'Event submission discarded for being too large.'; if (config.submissionBatchSize > 1) { log.error(`${message} Retrying with smaller batch size.`); config.submissionBatchSize = Math.max(1, Math.round(config.submissionBatchSize / 1.5)); } else { log.error(`${message} ${noSubmission}`); this.removeEvents(events); } return; } if (!response.success) { log.error(`Error submitting events: ${response.message || 'Please check the network tab for more info.'}`); this.suspendProcessing(); } } private removeEvents(events: IStorageItem[]) { for (let index = 0; index < (events || []).length; index++) { this._config.storage.queue.remove(events[index].timestamp); } } }