// 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 javax.ws.rs.client.Client;
import javax.ws.rs.client.ClientBuilder;
import javax.ws.rs.client.WebTarget;
import kafka.utils.ZKStringSerializer$;
import kafka.utils.ZkUtils;
import net.jotorren.microservices.tx.CompositeTransactionManager;
import net.jotorren.microservices.tx.impl.CompositeTransactionManagerKafkaImpl;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.ZkConnection;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import com.atomikos.icatch.tcc.rest.CoordinatorImp;
import com.atomikos.icatch.tcc.rest.TransactionProvider;
import com.fasterxml.jackson.jaxrs.json.JacksonJaxbJsonProvider;

package net.jotorren.microservices.configuration;





@Configuration
public class CompositeTransactionConfiguration {

	@Value("${tcc.service.base.url}")
	private String tccCoordinatorBaseUrl;

	
	@Value("${zookeeper.servers}")
	private String zooServers;

	@Value("${zookeeper.session.timeout}")
	private int zooSessionTimeout;
	
	@Value("${zookeeper.connection.timeout}")
	private int zooConnectionTimeout;

	
    @Bean
    public CoordinatorImp tccCoordinatorService() {
        return new CoordinatorImp();
    }
    
	@Bean
	public WebTarget tccCoordinatorClient() {
		Client client = ClientBuilder.newClient();
		client.register(new JacksonJaxbJsonProvider());
		client.register(new TransactionProvider());
		WebTarget target = client.target(tccCoordinatorBaseUrl);
		return target.path("/coordinator");
	}

	@Bean
	public CompositeTransactionManager compositeTransactionManager() {

// 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-composite-service/src/main/java/net/jotorren/microservices/configuration/CompositeTransactionConfiguration.java
import javax.ws.rs.client.Client;
import javax.ws.rs.client.ClientBuilder;
import javax.ws.rs.client.WebTarget;
import kafka.utils.ZKStringSerializer$;
import kafka.utils.ZkUtils;
import net.jotorren.microservices.tx.CompositeTransactionManager;
import net.jotorren.microservices.tx.impl.CompositeTransactionManagerKafkaImpl;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.ZkConnection;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import com.atomikos.icatch.tcc.rest.CoordinatorImp;
import com.atomikos.icatch.tcc.rest.TransactionProvider;
import com.fasterxml.jackson.jaxrs.json.JacksonJaxbJsonProvider;

package net.jotorren.microservices.configuration;





@Configuration
public class CompositeTransactionConfiguration {

	@Value("${tcc.service.base.url}")
	private String tccCoordinatorBaseUrl;

	
	@Value("${zookeeper.servers}")
	private String zooServers;

	@Value("${zookeeper.session.timeout}")
	private int zooSessionTimeout;
	
	@Value("${zookeeper.connection.timeout}")
	private int zooConnectionTimeout;

	
    @Bean
    public CoordinatorImp tccCoordinatorService() {
        return new CoordinatorImp();
    }
    
	@Bean
	public WebTarget tccCoordinatorClient() {
		Client client = ClientBuilder.newClient();
		client.register(new JacksonJaxbJsonProvider());
		client.register(new TransactionProvider());
		WebTarget target = client.target(tccCoordinatorBaseUrl);
		return target.path("/coordinator");
	}

	@Bean
	public CompositeTransactionManager compositeTransactionManager() {

		return new CompositeTransactionManagerKafkaImpl();
