/** * @fileoverview Enterprise event stream backed by the enterprise events API */// ------------------------------------------------------------------------------// Requirements// ------------------------------------------------------------------------------import { Readable } from 'stream';import BoxClient from './box-client';// ------------------------------------------------------------------------------// Typedefs// ------------------------------------------------------------------------------type Options = { streamPosition?: string; startDate?: string; endDate?: string; eventTypeFilter?: EventType[]; pollingInterval?: number; chunkSize?: number; streamType?: 'admin_logs' | 'admin_logs_streaming';};type EventType = string /* FIXME */;// ------------------------------------------------------------------------------// Private// ------------------------------------------------------------------------------const DEFAULT_OPTIONS = Object.freeze({ pollingInterval: 60, // seconds chunkSize: 500, streamType: 'admin_logs'});// ------------------------------------------------------------------------------// Public// ------------------------------------------------------------------------------/** * Stream of Box enterprise events. * * By default, the stream starts from the current time. * Pass 'startDate' to start from a specific time. * Pass 'streamPosition' to start from a previous stream position, or '0' for all available past events (~1 year). * Once the stream catches up to the current time, it will begin polling every 'pollingInterval' seconds. * If 'pollingInterval' = 0, then the stream will end when it catches up to the current time (no polling). * * @param {BoxClient} client - The client to use to get events * @param {Object} [options] - Options * @param {string} [options.streamPosition] - The stream position to start from (pass '0' for all past events) * @param {string} [options.startDate] - The date to start from * @param {string} [options.endDate] - The date to end at * @param {EventType[]} [options.eventTypeFilter] - Array of event types to return * @param {int} [options.pollingInterval=60] - Polling interval (in seconds). Pass 0 for no polling. * @param {int} [options.chunkSize=500] - Number of events to fetch per call (max = 500) * @constructor * @extends Readable */class EnterpriseEventStream extends Readable { _client: BoxClient; _options: Options & Required<Pick<Options, 'pollingInterval' | 'chunkSize'>>; _streamPosition?: string; constructor(client: BoxClient, options?: Options) { super({ objectMode: true, }); /** * @var {BoxClient} - The client for making API calls * @private */ this._client = client; /** * @var {Object} - Options * @private */ this._options = Object.assign({}, DEFAULT_OPTIONS, options); // Handle the case where the caller passes streamPosition = 0 instead of streamPosition = '0'. if ( this._options.streamType === 'admin_logs' && !this._options.startDate && !this._options.streamPosition && (this._options.streamPosition as any) !== 0 ) { // If neither startDate nor streamPosition is specified, start from the current time. this._options.startDate = new Date() .toISOString() .replace(/\.000Z$/, '-00:00'); } /** * @var {?string} - The current stream position * @private */ this._streamPosition = this._options.streamPosition; } /** * @returns {?number} - Returns null if no events have been fetched from Box yet. */ getStreamPosition() { return this._streamPosition; } /** * Get the stream state. * * @returns {Object} - The stream state */ getStreamState() { // We need to return both streamPosition and startDate, since streamPosition will be null until // the first set of events is returned from Box. return { streamPosition: this._streamPosition, startDate: this._options.startDate, endDate: this._options.endDate, eventTypeFilter: this._options.eventTypeFilter, }; } /** * Set the stream state. * * @param {Object} state - The stream state * @returns {void} */ setStreamState( state: Pick< Options, 'streamPosition' | 'startDate' | 'endDate' | 'eventTypeFilter' > ) { // We need to set both streamPosition and startDate, since streamPosition will be null until // the first set of events is returned from Box. this._streamPosition = state.streamPosition; this._options.startDate = state.startDate; this._options.endDate = state.endDate; this._options.eventTypeFilter = state.eventTypeFilter; } /** * Fetch the next chunk of events * * If there are no events, poll until events are available. * If an error occurs, emit the error but continuing polling as usual. * @param {Function} callback - Passed the array of events * @returns {void} * @private */ fetchEvents(callback: Function) { const self = this, params: { stream_type?: 'admin_logs' | 'admin_logs_streaming'; stream_position?: string; created_after?: string; created_before?: string; event_type?: string; limit?: number; } = { stream_type: this._options.streamType }; // Use the current stream position. // Handle the case where the caller passes streamPosition === 0 instead of streamPosition === '0'. if (this._streamPosition || (this._streamPosition as any) === 0) { params.stream_position = this._streamPosition; } if (this._options.streamType === 'admin_logs' && this._options.startDate) { params.created_after = this._options.startDate; } if (this._options.streamType === 'admin_logs' && this._options.endDate) { params.created_before = this._options.endDate; } if (this._options.eventTypeFilter) { params.event_type = this._options.eventTypeFilter.join(','); } if (this._options.chunkSize) { params.limit = this._options.chunkSize; } this._client.events.get(params, ( err: any /* FIXME */, result: any /* FIXME */ ) => { if (err) { self.emit('error', err); // If there was a "permanent" error, we would call the callback with it here. // But it's not clear which errors are truly permanent? // If Box is down or returning errors for an extended period, we still want to resume when it recovers. // So, continue polling at the regular frequency. // Don't use a shorter retry interval (to avoid DDOSing Box). } if (err || !result || !result.entries || result.entries.length === 0) { if (!self._options.pollingInterval) { // If polling is disabled, end the stream. callback(); return; } // There were no events returned (or an error occurred), so schedule another poll. const delay = self._options.pollingInterval * 1000; // Stream readers can use this to flush buffered events to a downstream system. self.emit('wait', delay); setTimeout(() => { self.fetchEvents(callback); }, delay); return; } // Only update the stream position if there were events returned. // The API currently returns next_stream_position = 0 if there are no events (may be a bug?). // But we don't want to start over at the beginning in that case, so ignore it. self._streamPosition = result.next_stream_position; // Notify the reader of the new stream position. // Stream readers can respond to the 'newStreamState' event to persist the stream state. self.emit('newStreamState', self.getStreamState()); callback(null, result.entries); }); } /** * Implementation of the stream-internal read function. This is called * by the stream whenever it needs more data, and will not be called again * until data is pushed into the stream. * @returns {void} * @private */ _read() { // Fetch the next chunk of events. const self = this; // This will poll forever until events are available. this.fetchEvents((err: any /* FIXME */, events: any /* FIXME */) => { if (err || !events || events.length === 0) { // Close the stream if there was a "permanent" failure or we reached the end of the events. self.push(null); return; } // Pause the stream to avoid race conditions while pushing in the new events. // Without this, _read() would be called again from inside each push(), // resulting in multiple parallel calls to fetchEvents(). // See https://github.com/nodejs/node/issues/3203 const wasPaused = self.isPaused(); self.pause(); // Push all of the events into the stream. events.forEach((event: any /* FIXME */) => { self.push(event); }); if (!wasPaused) { // This will deliver the events and trigger the next call to _read() once they have been consumed. self.resume(); } }); }}export = EnterpriseEventStream;