// Path: rahub-cross-cutting-concerns/src/main/java/net/jotorren/microservices/context/SpringContextProvider.java
// public class SpringContextProvider implements ApplicationContextAware {
// 	private static ApplicationContext ctx = null;
// 
// 	public static ApplicationContext getApplicationContext() {
// 		return ctx;
// 	}
// 
// 	public void setApplicationContext(ApplicationContext ctx) {
// 		SpringContextProvider.ctx = ctx;
// 	}
// }
// 
// Path: rahub-cross-cutting-concerns/src/main/java/net/jotorren/microservices/tx/CompositeTransactionManager.java
// public interface CompositeTransactionManager {
// 
// 	void open(String txId);
// 	void enlist(String txId, EntityCommand<?> ec);
// 	List<EntityCommand<?>> fetch(String txId);
// 	void close(String txId);
// }
// 
// Path: rahub-cross-cutting-concerns/src/main/java/net/jotorren/microservices/tx/impl/CompositeTransactionManagerKafkaImpl.java
// public class CompositeTransactionManagerKafkaImpl implements CompositeTransactionManager {
// 
// 	private static final Logger LOG = LoggerFactory.getLogger(CompositeTransactionManager.class);
// 
// 	@Value("${zookeeper.partitions:0}")
// 	private int zooPartitions;
// 	
// 	@Value("${zookeeper.replication:0}")
// 	private int zooReplication;
// 	
// 	@Value("${kafka.consumer.poll.timeout:0}")
// 	private long kafkaConsumerPollTimeout;
// 	
// 	
// 	@Autowired(required = false)
// 	private ZkUtils zkUtils;
// 	
// 	@Autowired(required = false)
// 	private KafkaTemplate<Integer, String> kafkaTemplate;
// 
// 	@Autowired(required = false)
// 	@Qualifier("kafkaConsumerConfiguration")
// 	private Map<String, Object> configuration;
// 	
// 	@Autowired(required = false)
// 	private Serializer<EntityCommand<?>> serializer;
// 	
// 	@Override
// 	public void open(String txId) {
// 		// Add topic configuration here
//         Properties topicConfig = new Properties();
// 
//         AdminUtils.createTopic(zkUtils, txId, zooPartitions, zooReplication, topicConfig, RackAwareMode.Enforced$.MODULE$);
// 	}
// 
// 	@Override
// 	public void enlist(String txId, EntityCommand<?> command) {
// 		kafkaTemplate.send(txId, serializer.writeToString(command));
// 	}
// 
// 	@SuppressWarnings("unchecked")
// 	@Override
// 	public List<EntityCommand<?>> fetch(String txId) {
// 		List<EntityCommand<?>> transactionOperations = new ArrayList<EntityCommand<?>>();
// 
// 		Map<String, Object> consumerConfigs = (Map<String, Object>)configuration.get("kafkaConsumerConfiguration");
// 		consumerConfigs.put(ConsumerConfig.GROUP_ID_CONFIG, UUID.randomUUID().toString());
// 		
// 		KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(consumerConfigs);
// 		kafkaConsumer.subscribe(Arrays.asList(txId));
// 		
// 		ConsumerRecords<String, String> records = kafkaConsumer.poll(kafkaConsumerPollTimeout);
// 		for (ConsumerRecord<String, String> record : records){
// 			LOG.info("offset = {}, key = {}, value = {}", record.offset(), record.key(), record.value());
// 			try {
// 				transactionOperations.add(serializer.readFromString(record.value()));
// 			} catch (SerializationFailedException e) {
// 				LOG.error("Unable to deserialize [{}] because of: {}", record.value(), e.getMessage());
// 			}
// 		}
// 		
// 		kafkaConsumer.close();
// 			
// 		return transactionOperations;
// 	}
// 	
// 	@Override
// 	public void close(String txId) {
// 		AdminUtils.deleteTopic(zkUtils, txId);
// 	}
// }


import java.util.HashMap;
import java.util.Map;
import net.jotorren.microservices.context.SpringContextProvider;
import net.jotorren.microservices.tx.CompositeTransactionManager;
import net.jotorren.microservices.tx.impl.CompositeTransactionManagerKafkaImpl;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
package net.jotorren.microservices.content.configuration;




@Configuration
public class CompositeTransactionConfiguration {
	
	@Value("${kafka.bootstrap-servers}")
	private String kafkaBootstrapServers;

	@Value("${kafka.consumer.enable.auto.commit}")
	private String kafkaEnableAutoCommit;

	@Value("${kafka.consumer.auto.commit.interval}")
	private String kafkaAutoCommitInterval;
	
	@Value("${kafka.consumer.session.timeout}")
	private String kafkaSessionTimeout;
	
	@Value("${kafka.consumer.auto.offset.reset}")
	private String kafkaAutoOffsetReset;
	
	@Bean
	public SpringContextProvider springContextProvider(){
		return new SpringContextProvider();
	}
	
	@Bean
	public CompositeTransactionManager compositeTransactionManager() {
// Path: rahub-cross-cutting-concerns/src/main/java/net/jotorren/microservices/context/SpringContextProvider.java
// public class SpringContextProvider implements ApplicationContextAware {
// 	private static ApplicationContext ctx = null;
// 
// 	public static ApplicationContext getApplicationContext() {
// 		return ctx;
// 	}
// 
// 	public void setApplicationContext(ApplicationContext ctx) {
// 		SpringContextProvider.ctx = ctx;
// 	}
// }
// 
// Path: rahub-cross-cutting-concerns/src/main/java/net/jotorren/microservices/tx/CompositeTransactionManager.java
// public interface CompositeTransactionManager {
// 
// 	void open(String txId);
// 	void enlist(String txId, EntityCommand<?> ec);
// 	List<EntityCommand<?>> fetch(String txId);
// 	void close(String txId);
// }
// 
// Path: rahub-cross-cutting-concerns/src/main/java/net/jotorren/microservices/tx/impl/CompositeTransactionManagerKafkaImpl.java
// public class CompositeTransactionManagerKafkaImpl implements CompositeTransactionManager {
// 
// 	private static final Logger LOG = LoggerFactory.getLogger(CompositeTransactionManager.class);
// 
// 	@Value("${zookeeper.partitions:0}")
// 	private int zooPartitions;
// 	
// 	@Value("${zookeeper.replication:0}")
// 	private int zooReplication;
// 	
// 	@Value("${kafka.consumer.poll.timeout:0}")
// 	private long kafkaConsumerPollTimeout;
// 	
// 	
// 	@Autowired(required = false)
// 	private ZkUtils zkUtils;
// 	
// 	@Autowired(required = false)
// 	private KafkaTemplate<Integer, String> kafkaTemplate;
// 
// 	@Autowired(required = false)
// 	@Qualifier("kafkaConsumerConfiguration")
// 	private Map<String, Object> configuration;
// 	
// 	@Autowired(required = false)
// 	private Serializer<EntityCommand<?>> serializer;
// 	
// 	@Override
// 	public void open(String txId) {
// 		// Add topic configuration here
//         Properties topicConfig = new Properties();
// 
//         AdminUtils.createTopic(zkUtils, txId, zooPartitions, zooReplication, topicConfig, RackAwareMode.Enforced$.MODULE$);
// 	}
// 
// 	@Override
// 	public void enlist(String txId, EntityCommand<?> command) {
// 		kafkaTemplate.send(txId, serializer.writeToString(command));
// 	}
// 
// 	@SuppressWarnings("unchecked")
// 	@Override
// 	public List<EntityCommand<?>> fetch(String txId) {
// 		List<EntityCommand<?>> transactionOperations = new ArrayList<EntityCommand<?>>();
// 
// 		Map<String, Object> consumerConfigs = (Map<String, Object>)configuration.get("kafkaConsumerConfiguration");
// 		consumerConfigs.put(ConsumerConfig.GROUP_ID_CONFIG, UUID.randomUUID().toString());
// 		
// 		KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(consumerConfigs);
// 		kafkaConsumer.subscribe(Arrays.asList(txId));
// 		
// 		ConsumerRecords<String, String> records = kafkaConsumer.poll(kafkaConsumerPollTimeout);
// 		for (ConsumerRecord<String, String> record : records){
// 			LOG.info("offset = {}, key = {}, value = {}", record.offset(), record.key(), record.value());
// 			try {
// 				transactionOperations.add(serializer.readFromString(record.value()));
// 			} catch (SerializationFailedException e) {
// 				LOG.error("Unable to deserialize [{}] because of: {}", record.value(), e.getMessage());
// 			}
// 		}
// 		
// 		kafkaConsumer.close();
// 			
// 		return transactionOperations;
// 	}
// 	
// 	@Override
// 	public void close(String txId) {
// 		AdminUtils.deleteTopic(zkUtils, txId);
// 	}
// }


// Path: rahub-source-code-service/src/main/java/net/jotorren/microservices/content/configuration/CompositeTransactionConfiguration.java
import java.util.HashMap;
import java.util.Map;
import net.jotorren.microservices.context.SpringContextProvider;
import net.jotorren.microservices.tx.CompositeTransactionManager;
import net.jotorren.microservices.tx.impl.CompositeTransactionManagerKafkaImpl;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

package net.jotorren.microservices.content.configuration;




@Configuration
public class CompositeTransactionConfiguration {
	
	@Value("${kafka.bootstrap-servers}")
	private String kafkaBootstrapServers;

	@Value("${kafka.consumer.enable.auto.commit}")
	private String kafkaEnableAutoCommit;

	@Value("${kafka.consumer.auto.commit.interval}")
	private String kafkaAutoCommitInterval;
	
	@Value("${kafka.consumer.session.timeout}")
	private String kafkaSessionTimeout;
	
	@Value("${kafka.consumer.auto.offset.reset}")
	private String kafkaAutoOffsetReset;
	
	@Bean
	public SpringContextProvider springContextProvider(){
		return new SpringContextProvider();
	}
	
	@Bean
	public CompositeTransactionManager compositeTransactionManager() {
		return new CompositeTransactionManagerKafkaImpl();