aws/SnsProvider.js
/**
		 * Subscribes an SQS queue to an SNS topic. Once the subscription
		 * has been established the queue can be monitored (see
		 * {@link SqsProvider#receive} or {@link SqsProvider#observe}).
		 *
		 * The promise will return a Disposable instance. Call the
		 * dispose method to delete the subscription.
		 *
		 * @public
		 * @param {string} topicName - The name of the topic to subscribe to.
		 * @param {Object} queueArn - The ARN of the queue to receive notifications (see {@link SqsProvider#getQueueArn}).
		 * @returns {Promise<Disposable>}
		 */
Subscribes an SQS queue to an SNS topic. Once the subscription
has been established the queue can be monitored (see
SqsProvider#receive or  SqsProvider#observe).

The promise will return a Disposable instance. Call the
dispose method to delete the subscription.

@public
@param {string} topicName - The name of the topic to subscribe to.
@param {Object} queueArn - The ARN of the queue to receive notifications .
@returns {Promise}
Subscribes an SQS queue to an SNS topic . Once the subscription has been established the queue can be monitored ( see SqsProvider#receive or SqsProvider#observe ) . The promise will return a Disposable instance . Call the dispose method to delete the subscription . @public @param { string } topicName - The name of the topic to subscribe to . @param { Object } queueArn - The ARN of the queue to receive notifications . @returns { Promise }
subscribe(topicName, queueArn) {
			return Promise.resolve()
				.then(() => {
					assert.argumentIsRequired(topicName, 'topicName', String);
					assert.argumentIsRequired(queueArn, 'queueArn', String);
					checkReady.call(this);
					const qualifiedTopicName = getQualifiedTopicName(this._configuration.prefix, topicName);
					if (!this._subscriptionPromises.hasOwnProperty(qualifiedTopicName)) {
						this._subscriptionPromises[qualifiedTopicName] = this.getTopicArn(topicName)
							.then((topicArn) => {
								return promise.build(
									(resolveCallback, rejectCallback) => {
										logger.debug('Subscribing SQS queue to SNS topic [', qualifiedTopicName, ']');
										this._sns.subscribe({
											'TopicArn': topicArn,
											'Endpoint': queueArn,
											'Protocol': 'sqs'
										}, (error, data) => {
											if (error === null) {
												logger.info('SNS subscription to SQS topic complete [', qualifiedTopicName, ']');
												resolveCallback(Disposable.fromAction(() => {
													if (this.getIsDisposed()) {
														return;
													}
													logger.debug('Unsubscribing SQS queue from SNS topic [', qualifiedTopicName, ']');
													delete this._subscriptionPromises[qualifiedTopicName];
													this._sns.unsubscribe({
														SubscriptionArn: data.SubscriptionArn
													}, (error, data) => {
														if (error === null) {
															logger.info('SQS unsubscribe from SNS topic complete [', qualifiedTopicName, ']');
														} else {
															logger.error('SQS unsubscribe from SNS topic failed [', qualifiedTopicName, ']');
															logger.error(error);
														}
													});
												}));
											} else {
												logger.error('SNS subscription to SQS topic failed [', qualifiedTopicName, ']');
												logger.error(error);
												rejectCallback('Failed to subscribe to SNS topic.');
											}
										});
									}
								);
							});
					}
					return this._subscriptionPromises[qualifiedTopicName];
				});
		}
subscribe ( topicName , queueArn ) { return Promise . resolve ( ) . then ( ( ) => { assert . argumentIsRequired ( topicName , 'topicName' , String ) ; assert . argumentIsRequired ( queueArn , 'queueArn' , String ) ; checkReady . call ( this ) ; const qualifiedTopicName = getQualifiedTopicName ( this . _configuration . prefix , topicName ) ; if ( ! this . _subscriptionPromises . hasOwnProperty ( qualifiedTopicName ) ) { this . _subscriptionPromises [ qualifiedTopicName ] = this . getTopicArn ( topicName ) . then ( ( topicArn ) => { return promise . build ( ( resolveCallback , rejectCallback ) => { logger . debug ( 'Subscribing SQS queue to SNS topic [' , qualifiedTopicName , ']' ) ; this . _sns . subscribe ( { 'TopicArn' : topicArn , 'Endpoint' : queueArn , 'Protocol' : 'sqs' } , ( error , data ) => { if ( error === null ) { logger . info ( 'SNS subscription to SQS topic complete [' , qualifiedTopicName , ']' ) ; resolveCallback ( Disposable . fromAction ( ( ) => { if ( this . getIsDisposed ( ) ) { return ; } logger . debug ( 'Unsubscribing SQS queue from SNS topic [' , qualifiedTopicName , ']' ) ; delete this . _subscriptionPromises [ qualifiedTopicName ] ; this . _sns . unsubscribe ( { SubscriptionArn : data . SubscriptionArn } , ( error , data ) => { if ( error === null ) { logger . info ( 'SQS unsubscribe from SNS topic complete [' , qualifiedTopicName , ']' ) ; } else { logger . error ( 'SQS unsubscribe from SNS topic failed [' , qualifiedTopicName , ']' ) ; logger . error ( error ) ; } } ) ; } ) ) ; } else { logger . error ( 'SNS subscription to SQS topic failed [' , qualifiedTopicName , ']' ) ; logger . error ( error ) ; rejectCallback ( 'Failed to subscribe to SNS topic.' ) ; } } ) ; } ) ; } ) ; } return this . _subscriptionPromises [ qualifiedTopicName ] ; } ) ; }
Subscribes an SQS queue to an SNS topic.
Subscribes an SQS queue to an SNS topic .