/*
 * Copyright 2013 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.elasticspring.messaging.endpoint;

import com.amazonaws.services.sns.AmazonSNS;
import com.amazonaws.services.sns.model.Subscription;
import com.amazonaws.services.sqs.AmazonSQSAsync;
import com.amazonaws.services.sqs.model.GetQueueAttributesRequest;
import com.amazonaws.services.sqs.model.GetQueueUrlRequest;
import org.elasticspring.core.naming.AmazonResourceName;
import org.elasticspring.messaging.config.annotation.TopicListener;
import org.elasticspring.messaging.listener.MessageListenerAdapter;
import org.elasticspring.messaging.listener.SimpleMessageListenerContainer;
import org.elasticspring.messaging.support.converter.NotificationMessageConverter;
import org.springframework.util.Assert;

/**
 * @author Agim Emruli
 * @since 1.0
 */
public class QueueingNotificationEndpointFactoryBean extends AbstractNotificationEndpointFactoryBean<SimpleMessageListenerContainer> {

	private final AmazonSQSAsync amazonSqs;
	private final NotificationMessageConverter messageConverter = new NotificationMessageConverter();

	public QueueingNotificationEndpointFactoryBean(AmazonSNS amazonSns, AmazonSQSAsync amazonSqs, String topicName,
												   TopicListener.NotificationProtocol protocol, String endpoint, Object target, String method) {
		super(amazonSns, topicName, protocol, endpoint, target, method);
		this.amazonSqs = amazonSqs;
		Assert.notNull(amazonSqs, "amazonSqs must not be null");
		Assert.isTrue(protocol == TopicListener.NotificationProtocol.SQS, "This endpoint only support sqs endpoints");
	}

	@Override
	protected SimpleMessageListenerContainer doCreateEndpointInstance(Subscription subscription) {
		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
		container.setAmazonSqs(this.amazonSqs);

		MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(this.messageConverter, getTarget(), getMethod());
		container.setMessageListener(messageListenerAdapter);
		container.setDestinationName(AmazonResourceName.fromString(subscription.getEndpoint()).getResourceType());
		return container;
	}

	@Override
	public Class<SimpleMessageListenerContainer> getObjectType() {
		return SimpleMessageListenerContainer.class;
	}

	@Override
	public String getEndpoint() {
		String queueUrl = this.amazonSqs.getQueueUrl(new GetQueueUrlRequest(super.getEndpoint())).getQueueUrl();
		return this.amazonSqs.getQueueAttributes(new GetQueueAttributesRequest(queueUrl).withAttributeNames("QueueArn")).getAttributes().get("QueueArn");
	}
}
@@ -877,68 +877,8 @@
 st;%0A
-import com.amazonaws.services.sqs.model.GetQueueUrlRequest;%0A
 impo
@@ -1221,24 +1221,322 @@
 eConverter;%0A
+import org.elasticspring.messaging.support.destination.CachingDestinationResolver;%0Aimport org.elasticspring.messaging.support.destination.DestinationResolver;%0Aimport org.elasticspring.messaging.support.destination.DynamicQueueDestinationResolver;%0Aimport org.springframework.context.SmartLifecycle;%0A
 import org.s
@@ -1741,16 +1741,42 @@
 ntainer%3E
+ implements SmartLifecycle
  %7B%0A%0A%09pri
@@ -1910,16 +1910,123 @@
 erter();
+%0A%09private final DestinationResolver destinationResolver;%0A%09private SimpleMessageListenerContainer container;
 %0A%0A%09publi
@@ -2312,38 +2312,8 @@
 d);%0A
-%09%09this.amazonSqs = amazonSqs;%0A
 %09%09As
@@ -2480,16 +2480,155 @@
 ints%22);%0A
+%09%09this.amazonSqs = amazonSqs;%0A%09%09this.destinationResolver = new CachingDestinationResolver(new DynamicQueueDestinationResolver(amazonSqs));%0A
 %09%7D%0A%0A%09@Ov
@@ -2736,39 +2736,13 @@
 %7B%0A%09%09
-SimpleMessageListenerContainer 
+this.
 cont
@@ -2785,24 +2785,29 @@
 tainer();%0A%09%09
+this.
 container.se
@@ -2960,24 +2960,29 @@
 ethod());%0A%09%09
+this.
 container.se
@@ -3025,16 +3025,21 @@
 ter);%0A%09%09
+this.
 containe
@@ -3146,16 +3146,21 @@
 %09return 
+this.
 containe
@@ -3363,52 +3363,50 @@
 his.
-amazonSqs.getQueueUrl(new GetQueueUrlRequest
+destinationResolver.resolveDestinationName
 (sup
@@ -3426,23 +3426,8 @@
 t())
-).getQueueUrl()
 ;%0A%09%09
@@ -3574,10 +3574,471 @@
 rn%22);%0A%09%7D
+%0A%0A%09@Override%0A%09public boolean isAutoStartup() %7B%0A%09%09return this.container.isAutoStartup();%0A%09%7D%0A%0A%09@Override%0A%09public void stop(Runnable callback) %7B%0A%09%09this.container.stop(callback);%0A%09%7D%0A%0A%09@Override%0A%09public void start() %7B%0A%09%09this.container.start();%0A%09%7D%0A%0A%09@Override%0A%09public void stop() %7B%0A%09%09this.container.stop();%0A%09%7D%0A%0A%09@Override%0A%09public boolean isRunning() %7B%0A%09%09return this.container.isRunning();%0A%09%7D%0A%0A%09@Override%0A%09public int getPhase() %7B%0A%09%09return this.container.getPhase();%0A%09%7D
 %0A%7D

Add support to respect Lifecycle interface