/**
 * This operations ensures that required dependencies are not accidentally injected
 * early given that this bean is BPP.
 */
@SuppressWarnings({ "unchecked", "rawtypes" })
private void injectAndPostProcessDependencies() {
	Collection<StreamListenerParameterAdapter> streamListenerParameterAdapters = this.applicationContext
			.getBeansOfType(StreamListenerParameterAdapter.class).values();
	Collection<StreamListenerResultAdapter> streamListenerResultAdapters = this.applicationContext
			.getBeansOfType(StreamListenerResultAdapter.class).values();
	this.binderAwareChannelResolver = this.applicationContext
			.getBean("binderAwareChannelResolver", DestinationResolver.class);
	this.messageHandlerMethodFactory = this.applicationContext
			.getBean("integrationMessageHandlerMethodFactory", MessageHandlerMethodFactory.class);
	this.springIntegrationProperties = this.applicationContext
			.getBean(SpringIntegrationProperties.class);

	this.streamListenerSetupMethodOrchestrators.addAll(this.applicationContext
			.getBeansOfType(StreamListenerSetupMethodOrchestrator.class).values());

	// Default orchestrator for StreamListener method invocation is added last into
	// the LinkedHashSet.
	this.streamListenerSetupMethodOrchestrators.add(
			new DefaultStreamListenerSetupMethodOrchestrator(this.applicationContext,
					streamListenerParameterAdapters, streamListenerResultAdapters));

	this.streamListenerCallbacks.forEach(Runnable::run);
}
 

@Test
void send_validTextMessageWithCustomDestinationResolver_usesTopicChannel()
		throws Exception {
	// Arrange
	AmazonSNS amazonSns = mock(AmazonSNS.class);
	NotificationMessagingTemplate notificationMessagingTemplate = new NotificationMessagingTemplate(
			amazonSns,
			(DestinationResolver<String>) name -> name.toUpperCase(Locale.ENGLISH),
			null);

	// Act
	notificationMessagingTemplate.send("test",
			MessageBuilder.withPayload("Message content").build());

	// Assert
	verify(amazonSns).publish(new PublishRequest("TEST", "Message content", null)
			.withMessageAttributes(isNotNull()));
}
 

@Test
void send_withCustomDestinationResolveAndDestination_usesDestination() {
	AmazonSQSAsync amazonSqs = createAmazonSqs();
	QueueMessagingTemplate queueMessagingTemplate = new QueueMessagingTemplate(
			amazonSqs,
			(DestinationResolver<String>) name -> name.toUpperCase(Locale.ENGLISH),
			null);

	Message<String> stringMessage = MessageBuilder.withPayload("message content")
			.build();
	queueMessagingTemplate.send("myqueue", stringMessage);

	ArgumentCaptor<SendMessageRequest> sendMessageRequestArgumentCaptor = ArgumentCaptor
			.forClass(SendMessageRequest.class);
	verify(amazonSqs).sendMessage(sendMessageRequestArgumentCaptor.capture());
	assertThat(sendMessageRequestArgumentCaptor.getValue().getQueueUrl())
			.isEqualTo("MYQUEUE");
}
 

@Override
protected Map<String, Class<?>> getOptionTypeMappings() {
    Map<String, Class<?>> mappings = super.getOptionTypeMappings();
    mappings.put("messagingTemplate", MessagingTemplate.class);
    mappings.put("channelResolver", DestinationResolver.class);
    return mappings;
}
 

public BinderAwareRouter(AbstractMappingMessageRouter[] routers,
		DestinationResolver<MessageChannel> channelResolver) {
	if (routers != null) {
		for (AbstractMappingMessageRouter router : routers) {
			router.setChannelResolver(channelResolver);
		}
	}
}
 

@Bean
@ConditionalOnMissingBean
public BinderAwareRouter binderAwareRouterBeanPostProcessor(
		@Autowired(required = false) AbstractMappingMessageRouter[] routers,
		@Autowired(required = false) @Qualifier("binderAwareChannelResolver")
			DestinationResolver<MessageChannel> channelResolver) {

	return new BinderAwareRouter(routers, channelResolver);
}
 

public NotificationMessagingTemplate(AmazonSNS amazonSns,
		DestinationResolver<String> destinationResolver,
		MessageConverter messageConverter) {
	super(destinationResolver);
	this.amazonSns = amazonSns;
	initMessageConverter(messageConverter);
}
 

@Test
void testDestinationResolverIsCreatedIfNull() throws Exception {
	AbstractMessageListenerContainer container = new StubAbstractMessageListenerContainer();

	container.setAmazonSqs(mock(AmazonSQSAsync.class, withSettings().stubOnly()));
	container.setMessageHandler(mock(QueueMessageHandler.class));
	container.afterPropertiesSet();

	DestinationResolver<String> destinationResolver = container
			.getDestinationResolver();
	assertThat(destinationResolver).isNotNull();
	assertThat(CachingDestinationResolverProxy.class.isInstance(destinationResolver))
			.isTrue();
}
 

@Test
void testCustomDestinationResolverSet() throws Exception {
	AbstractMessageListenerContainer container = new StubAbstractMessageListenerContainer();

	container.setAmazonSqs(mock(AmazonSQSAsync.class, withSettings().stubOnly()));
	container.setMessageHandler(mock(QueueMessageHandler.class));

	DestinationResolver<String> destinationResolver = mock(
			DynamicQueueUrlDestinationResolver.class);
	container.setDestinationResolver(destinationResolver);

	container.afterPropertiesSet();

	assertThat(container.getDestinationResolver()).isEqualTo(destinationResolver);
}
 

public void setChannelResolver(DestinationResolver<MessageChannel> channelResolver) {
	this.channelResolver = channelResolver;
}
 

protected DestinationResolver<String> getDestinationResolver() {
	return this.destinationResolver;
}
 

protected AbstractMessageChannelMessagingSendingTemplate(
		DestinationResolver<String> destinationResolver) {
	this.destinationResolver = new CachingDestinationResolverProxy<>(
			destinationResolver);
}
 

protected MessageSendingTemplateTest(
		DestinationResolver<String> destinationResolver) {
	super(destinationResolver);
}
 

/**
 * Initializes the messaging template by configuring the destination resolver as well
 * as the message converter. Uses the {@link DynamicQueueUrlDestinationResolver} with
 * the default configuration to resolve destination names.
 * @param amazonSqs The {@link AmazonSQS} client, cannot be {@code null}.
 * @param destinationResolver A destination resolver implementation to resolve queue
 * names into queue urls. The destination resolver will be wrapped into a
 * {@link org.springframework.messaging.core.CachingDestinationResolverProxy} to avoid
 * duplicate queue url resolutions.
 * @param messageConverter A {@link MessageConverter} that is going to be added to the
 * composite converter.
 */
public QueueMessagingTemplate(AmazonSQSAsync amazonSqs,
		DestinationResolver<String> destinationResolver,
		MessageConverter messageConverter) {
	super(destinationResolver);
	this.amazonSqs = amazonSqs;
	initMessageConverter(messageConverter);
}
 

/**
 * Configures the destination resolver used to retrieve the queue url based on the
 * destination name configured for this instance. <br>
 * This setter can be used when a custom configured {@link DestinationResolver} must
 * be provided. (For example if one want to have the
 * {@link DynamicQueueUrlDestinationResolver} with the auto creation of queues set to
 * {@code true}.
 * @param destinationResolver - the destination resolver. Must not be null
 */
public void setDestinationResolver(DestinationResolver<String> destinationResolver) {
	this.destinationResolver = destinationResolver;
}
 

/**
 * Configures the destination resolver used to retrieve the queue url based on the
 * destination name configured for this instance. <br>
 * This setter can be used when a custom configured {@link DestinationResolver} must
 * be provided. (For example if one want to have the
 * {@link org.springframework.cloud.aws.messaging.support.destination.DynamicQueueUrlDestinationResolver}
 * with the auto creation of queues set to {@code true}.
 * @param destinationResolver another or customized {@link DestinationResolver}
 */
public void setDestinationResolver(DestinationResolver<String> destinationResolver) {
	this.destinationResolver = destinationResolver;
}