aws/SqsProvider.js
/**
		 * Makes repeated reads from a queue until canceled and returns messages
		 * using the callback provided.
		 *
		 * @public
		 * @param {string} queueName - The name of the queue to read.
		 * @param {Function} callback - Invoked with a messages as they become available.
		 * @param {Number=} pollInterval - The milliseconds to wait between polling the queue.
		 * @param {Number=} pollDuration - The maximum amount of time the server-side long-poll will wait for messages to become available.
		 * @param {Number=} maximumMessages - The maximum number of messages to read per request (cannot be more than 10).
		 * @param {Object=} createOptions - Options to use when queue does not exist and must be created.
		 * @returns {Disposable}
		 */
Makes repeated reads from a queue until canceled and returns messages
using the callback provided.

@public
@param {string} queueName - The name of the queue to read.
@param {Function} callback - Invoked with a messages as they become available.
@param {Number=} pollInterval - The milliseconds to wait between polling the queue.
@param {Number=} pollDuration - The maximum amount of time the server-side long-poll will wait for messages to become available.
@param {Number=} maximumMessages - The maximum number of messages to read per request (cannot be more than 10).
@param {Object=} createOptions - Options to use when queue does not exist and must be created.
@returns {Disposable}
Makes repeated reads from a queue until canceled and returns messages using the callback provided . @public @param { string } queueName - The name of the queue to read . @param { Function } callback - Invoked with a messages as they become available . @param { Number = } pollInterval - The milliseconds to wait between polling the queue . @param { Number = } pollDuration - The maximum amount of time the server - side long - poll will wait for messages to become available . @param { Number = } maximumMessages - The maximum number of messages to read per request ( cannot be more than 10 ) . @param { Object = } createOptions - Options to use when queue does not exist and must be created . @returns { Disposable }
observe(queueName, callback, pollInterval, pollDuration, batchSize, createOptions) {
			assert.argumentIsRequired(queueName, 'queueName', String);
			assert.argumentIsRequired(callback, 'callback', Function);
			assert.argumentIsOptional(pollInterval, 'pollInterval', Number);
			assert.argumentIsOptional(pollDuration, 'pollDuration', Number);
			assert.argumentIsOptional(batchSize, 'batchSize', Number);
			checkReady.call(this);
			const qualifiedQueueName = getQualifiedQueueName(this._configuration.prefix, queueName);
			if (this._queueObservers.hasOwnProperty(qualifiedQueueName)) {
				throw new Error('The queue is already being observed.');
			}
			logger.debug('Creating observer for queue [', qualifiedQueueName, ']');
			let disposed = false;
			this._queueObservers[qualifiedQueueName] = Disposable.fromAction(() => {
				logger.info('Disposing observer of queue [', qualifiedQueueName, ']');
				disposed = true;
				delete this._queueObservers[qualifiedQueueName];
			});
			const checkQueue = () => {
				if (disposed) {
					return;
				}
				let delay;
				receiveMessages.call(this, queueName, pollDuration, batchSize, false, createOptions)
					.then((messages) => {
						return Promise.all(messages.map((message) => {
							if (disposed) {
								return;
							}
							return callback(message);
						})).catch((error) => {
							logger.error('An error occurred while processing message(s) from queue [', qualifiedQueueName, ']');
							logger.error(error);
						}).then(() => {
							if (messages.length === 0) {
								delay = pollInterval || 2000;
							} else {
								delay = 0;
							}
						});
					}).catch((error) => {
						logger.error('An error occurred while receiving message(s) from queue [', qualifiedQueueName, ']');
						logger.error(error);
					}).then(() => {
						if (disposed) {
							return;
						}
						if (!is.number(delay)) {
							delay = 5000;
						}
						this._scheduler.schedule(checkQueue, delay, 'Check queue (' + qualifiedQueueName + ')');
					});
			};
			checkQueue();
			return this._queueObservers[qualifiedQueueName];
		}
observe ( queueName , callback , pollInterval , pollDuration , batchSize , createOptions ) { assert . argumentIsRequired ( queueName , 'queueName' , String ) ; assert . argumentIsRequired ( callback , 'callback' , Function ) ; assert . argumentIsOptional ( pollInterval , 'pollInterval' , Number ) ; assert . argumentIsOptional ( pollDuration , 'pollDuration' , Number ) ; assert . argumentIsOptional ( batchSize , 'batchSize' , Number ) ; checkReady . call ( this ) ; const qualifiedQueueName = getQualifiedQueueName ( this . _configuration . prefix , queueName ) ; if ( this . _queueObservers . hasOwnProperty ( qualifiedQueueName ) ) { throw new Error ( 'The queue is already being observed.' ) ; } logger . debug ( 'Creating observer for queue [' , qualifiedQueueName , ']' ) ; let disposed = false ; this . _queueObservers [ qualifiedQueueName ] = Disposable . fromAction ( ( ) => { logger . info ( 'Disposing observer of queue [' , qualifiedQueueName , ']' ) ; disposed = true ; delete this . _queueObservers [ qualifiedQueueName ] ; } ) ; const checkQueue = ( ) => { if ( disposed ) { return ; } let delay ; receiveMessages . call ( this , queueName , pollDuration , batchSize , false , createOptions ) . then ( ( messages ) => { return Promise . all ( messages . map ( ( message ) => { if ( disposed ) { return ; } return callback ( message ) ; } ) ) . catch ( ( error ) => { logger . error ( 'An error occurred while processing message(s) from queue [' , qualifiedQueueName , ']' ) ; logger . error ( error ) ; } ) . then ( ( ) => { if ( messages . length === 0 ) { delay = pollInterval || 2000 ; } else { delay = 0 ; } } ) ; } ) . catch ( ( error ) => { logger . error ( 'An error occurred while receiving message(s) from queue [' , qualifiedQueueName , ']' ) ; logger . error ( error ) ; } ) . then ( ( ) => { if ( disposed ) { return ; } if ( ! is . number ( delay ) ) { delay = 5000 ; } this . _scheduler . schedule ( checkQueue , delay , 'Check queue (' + qualifiedQueueName + ')' ) ; } ) ; } ; checkQueue ( ) ; return this . _queueObservers [ qualifiedQueueName ] ; }
Makes repeated reads from a queue until canceled and returns messages
 using the callback provided.
Makes repeated reads from a queue until canceled and returns messages using the callback provided .