package com.atomikos.demo;

import javax.jms.ConnectionFactory;
import javax.jms.XAConnectionFactory;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.ActiveMQXAConnectionFactory;
import org.apache.activemq.broker.BrokerService;
import org.apache.camel.CamelContext;
import org.apache.camel.Component;
import org.apache.camel.component.jms.ConsumerType;
import org.apache.camel.component.jms.JmsComponent;
import org.apache.camel.component.jms.JmsConfiguration;
import org.apache.camel.component.jms.JmsEndpoint;
import org.apache.camel.component.jms.MessageListenerContainerFactory;
import org.apache.camel.spi.Policy;
import org.apache.camel.spring.boot.CamelContextConfiguration;
import org.apache.camel.spring.spi.SpringTransactionPolicy;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.DependsOn;
import org.springframework.jms.listener.AbstractMessageListenerContainer;
import org.springframework.transaction.PlatformTransactionManager;

import com.atomikos.icatch.config.UserTransactionService;
import com.atomikos.icatch.config.UserTransactionServiceImp;
import com.atomikos.jms.AtomikosConnectionFactoryBean;
import com.atomikos.spring.BatchingMessageListenerContainer;

@Configuration
public class CamelConfig {

	private static final String BROKER_URL = "vm://localhost";

	/**
	 * Embedded broker for local testing. 
	 * Delete if you have a centralized enterprise broker.
	 */

	@Bean(initMethod="start", destroyMethod="stop")
	public BrokerService embeddedBroker() throws Exception {
		BrokerService broker = new BrokerService();
		broker.setUseShutdownHook(false); // needed to avoid issues on shutdown
		broker.setPersistent(false); //in-memory for testing embedded
		broker.addConnector(BROKER_URL);
		return broker;
	}

	@Bean
	@DependsOn("embeddedBroker")
	public XAConnectionFactory xaConnectionFactory() {
		return new ActiveMQXAConnectionFactory(BROKER_URL);	}

	@Bean
	public ConnectionFactory connectionFactory() {
		return new ActiveMQConnectionFactory(BROKER_URL);	}

	@Autowired
	private PlatformTransactionManager transactionManager;

	/**
	 * Redeclared to allow DependsOn ordering for clean start/stop
	 * so we don't get stack traces on shutdown.
	 */

	@Bean(initMethod="init", destroyMethod="shutdownWait")
	@DependsOn("xaConnectionFactory")
	UserTransactionService userTransactionService() {
		return new UserTransactionServiceImp();
	}


	@Bean
	@DependsOn("userTransactionService")
	CamelContextConfiguration contextConfiguration() {
		return new CamelContextConfiguration() {
			@Override
			public void beforeApplicationStart(CamelContext camelContext) {
				camelContext.getRegistry().bind("PROPAGATION_REQUIRED", policy());
				camelContext.addComponent("jms", jmsComponent());
				camelContext.addComponent("noXaJms", noXaJmsComponent());
			}

			@Override
			public void afterApplicationStart(CamelContext camelContext) {
			}
		};
	}

	/**
	 * Non-XA connection factory for sending a test message to the in-queue
	 * because ActiveMQ does not allow using the XAConnectionFactory for non-XA JMS.
	 * 
	 * @return
	 */
	@Bean
	public Component noXaJmsComponent() {
		JmsComponent jmsComponent = new JmsComponent();
		jmsComponent.setConnectionFactory(connectionFactory());
		return jmsComponent;
	}

	@Bean
	public JmsComponent jmsComponent() {
		JmsComponent jmsComponent = new JmsComponent();
		jmsComponent.setConnectionFactory(atomikosConnectionFactoryBean());
		jmsComponent.setTransactionManager(transactionManager);
		jmsComponent.setConfiguration(jmsConfig());
		return jmsComponent;
	}

	@Bean
	public Policy policy() {
		SpringTransactionPolicy springTransactionPolicy = new SpringTransactionPolicy();
		springTransactionPolicy.setTransactionManager(transactionManager);
		springTransactionPolicy.setPropagationBehaviorName("PROPAGATION_REQUIRED");
		return springTransactionPolicy;
	}

	@Bean
	public JmsConfiguration jmsConfig() {
		JmsConfiguration jmsConfiguration = new JmsConfiguration();
		jmsConfiguration.setConsumerType(ConsumerType.Custom);
		jmsConfiguration.setConnectionFactory(atomikosConnectionFactoryBean());
		jmsConfiguration.setTransactionManager(transactionManager);
		jmsConfiguration.setTransacted(true);
		jmsConfiguration.setCacheLevelName("CACHE_NONE");
		jmsConfiguration.setMessageListenerContainerFactory(messageListenerContainerFactory());
		return jmsConfiguration;
	}

	@Bean
	public AtomikosConnectionFactoryBean atomikosConnectionFactoryBean() {
		AtomikosConnectionFactoryBean atomikosConnectionFactoryBean = new AtomikosConnectionFactoryBean();
		atomikosConnectionFactoryBean.setXaConnectionFactory(xaConnectionFactory());
		atomikosConnectionFactoryBean.setUniqueResourceName("jms");
		return atomikosConnectionFactoryBean;
	}


	@Bean
	@DependsOn({"userTransactionService"})
	public MessageListenerContainerFactory messageListenerContainerFactory() {
		MessageListenerContainerFactory batchingMessageListenerContainerFactory = new MessageListenerContainerFactory() {

			@Override
			public AbstractMessageListenerContainer createMessageListenerContainer(JmsEndpoint endpoint) {
				BatchingMessageListenerContainer batchingMessageListenerContainer = new BatchingMessageListenerContainer();
				batchingMessageListenerContainer.setSessionTransacted(true);
				batchingMessageListenerContainer.setTransactionManager(transactionManager);
				batchingMessageListenerContainer.setConnectionFactory(atomikosConnectionFactoryBean());
				return batchingMessageListenerContainer;
			}

		};
		return batchingMessageListenerContainerFactory;
	}

}
