/*
 * Copyright 2015-2018 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
 *
 *      https://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.springframework.cloud.stream.binder.rabbit.config;

import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

import org.springframework.amqp.core.DeclarableCustomizer;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionNameStrategy;
import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPostProcessor;
import org.springframework.amqp.support.postprocessor.GZipPostProcessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.context.PropertyPlaceholderAutoConfiguration;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder;
import org.springframework.cloud.stream.binder.rabbit.properties.RabbitBinderConfigurationProperties;
import org.springframework.cloud.stream.binder.rabbit.properties.RabbitExtendedBindingProperties;
import org.springframework.cloud.stream.binder.rabbit.provisioning.RabbitExchangeQueueProvisioner;
import org.springframework.cloud.stream.config.ConsumerEndpointCustomizer;
import org.springframework.cloud.stream.config.ListenerContainerCustomizer;
import org.springframework.cloud.stream.config.MessageSourceCustomizer;
import org.springframework.cloud.stream.config.ProducerMessageHandlerCustomizer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter;
import org.springframework.integration.amqp.inbound.AmqpMessageSource;
import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
import org.springframework.lang.Nullable;

/**
 * Configuration class for RabbitMQ message channel binder.
 *
 * @author David Turanski
 * @author Vinicius Carvalho
 * @author Artem Bilan
 * @author Oleg Zhurakousky
 * @author Gary Russell
 */
@Configuration
@Import({ PropertyPlaceholderAutoConfiguration.class })
@EnableConfigurationProperties({ RabbitBinderConfigurationProperties.class,
		RabbitExtendedBindingProperties.class })
public class RabbitMessageChannelBinderConfiguration {

	@Autowired
	private ConnectionFactory rabbitConnectionFactory;

	@Autowired
	private RabbitProperties rabbitProperties;

	@Autowired
	private RabbitBinderConfigurationProperties rabbitBinderConfigurationProperties;

	@Autowired
	private RabbitExtendedBindingProperties rabbitExtendedBindingProperties;

	@Bean
	RabbitMessageChannelBinder rabbitMessageChannelBinder(
			@Nullable ListenerContainerCustomizer<AbstractMessageListenerContainer> listenerContainerCustomizer,
			@Nullable MessageSourceCustomizer<AmqpMessageSource> sourceCustomizer,
			@Nullable ProducerMessageHandlerCustomizer<AmqpOutboundEndpoint> producerMessageHandlerCustomizer,
			@Nullable ConsumerEndpointCustomizer<AmqpInboundChannelAdapter> consumerCustomizer,
			List<DeclarableCustomizer> declarableCustomizers) {

		RabbitMessageChannelBinder binder = new RabbitMessageChannelBinder(
				this.rabbitConnectionFactory, this.rabbitProperties,
				provisioningProvider(declarableCustomizers), listenerContainerCustomizer, sourceCustomizer);
		binder.setAdminAddresses(
				this.rabbitBinderConfigurationProperties.getAdminAddresses());
		binder.setCompressingPostProcessor(gZipPostProcessor());
		binder.setDecompressingPostProcessor(deCompressingPostProcessor());
		binder.setNodes(this.rabbitBinderConfigurationProperties.getNodes());
		binder.setExtendedBindingProperties(this.rabbitExtendedBindingProperties);
		binder.setProducerMessageHandlerCustomizer(producerMessageHandlerCustomizer);
		binder.setConsumerEndpointCustomizer(consumerCustomizer);
		return binder;
	}

	@Bean
	MessagePostProcessor deCompressingPostProcessor() {
		return new DelegatingDecompressingPostProcessor();
	}

	@Bean
	MessagePostProcessor gZipPostProcessor() {
		GZipPostProcessor gZipPostProcessor = new GZipPostProcessor();
		gZipPostProcessor
				.setLevel(this.rabbitBinderConfigurationProperties.getCompressionLevel());
		return gZipPostProcessor;
	}

	@Bean
	RabbitExchangeQueueProvisioner provisioningProvider(List<DeclarableCustomizer> customizers) {
		return new RabbitExchangeQueueProvisioner(this.rabbitConnectionFactory, customizers);
	}

	@Bean
	@ConditionalOnMissingBean(ConnectionNameStrategy.class)
	@ConditionalOnProperty("spring.cloud.stream.rabbit.binder.connection-name-prefix")
	public ConnectionNameStrategy connectionNamer(CachingConnectionFactory cf) {
		final AtomicInteger nameIncrementer = new AtomicInteger();
		ConnectionNameStrategy namer = f -> this.rabbitBinderConfigurationProperties
				.getConnectionNamePrefix() + "#" + nameIncrementer.getAndIncrement();
		cf.setConnectionNameStrategy(namer);
		return namer;
	}

}

@@ -926,84 +926,16 @@
 ion.
-CachingConnectionFactory;%0Aimport org.springframework.amqp.rabbit.connection.
+Abstract
 Conn
@@ -1406,169 +1406,8 @@
 es;%0A
-import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;%0Aimport org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;%0A
 impo
@@ -3164,16 +3164,24 @@
 private 
+Abstract
 Connecti
@@ -3936,20 +3936,452 @@
 tomizers
-) %7B%0A
+,%0A%09%09%09@Nullable ConnectionNameStrategy connectionNameStrategy) %7B%0A%0A%09%09String connectionNamePrefix = this.rabbitBinderConfigurationProperties.getConnectionNamePrefix();%0A%09%09if (connectionNamePrefix != null &&  connectionNameStrategy == null) %7B%0A%09%09%09final AtomicInteger nameIncrementer = new AtomicInteger();%0A%09%09%09this.rabbitConnectionFactory.setConnectionNameStrategy(f -%3E connectionNamePrefix%0A%09%09%09%09%09+ %22#%22 + nameIncrementer.getAndIncrement());%0A%09%09%7D
 %0A%09%09Rabbi
@@ -5692,506 +5692,6 @@
 %0A%09%7D%0A
-%0A%09@Bean%0A%09@ConditionalOnMissingBean(ConnectionNameStrategy.class)%0A%09@ConditionalOnProperty(%22spring.cloud.stream.rabbit.binder.connection-name-prefix%22)%0A%09public ConnectionNameStrategy connectionNamer(CachingConnectionFactory cf) %7B%0A%09%09final AtomicInteger nameIncrementer = new AtomicInteger();%0A%09%09ConnectionNameStrategy namer = f -%3E this.rabbitBinderConfigurationProperties%0A%09%09%09%09.getConnectionNamePrefix() + %22#%22 + nameIncrementer.getAndIncrement();%0A%09%09cf.setConnectionNameStrategy(namer);%0A%09%09return namer;%0A%09%7D%0A%0A
 %7D%0A

Fix circular bean reference