package com.atomikos.hazelcast;

import static org.junit.Assert.*;

import java.util.Arrays;
import java.util.Map;

import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import com.atomikos.icatch.config.Configuration;
import com.atomikos.icatch.jta.UserTransactionManager;
import com.atomikos.icatch.provider.ConfigProperties;
import com.hazelcast.config.Config;
import com.hazelcast.config.GroupConfig;
import com.hazelcast.config.JoinConfig;
import com.hazelcast.config.MulticastConfig;
import com.hazelcast.config.NetworkConfig;
import com.hazelcast.config.TcpIpConfig;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.TransactionalMap;
import com.hazelcast.instance.HazelcastInstanceFactory;
import com.hazelcast.transaction.TransactionContext;
import com.hazelcast.transaction.TransactionNotActiveException;
import com.hazelcast.transaction.TransactionOptions;

public class HazelcastXaIntegrationTest {
	
	  static Config hazelcastConfig() {
		    MulticastConfig multicastConfig = new MulticastConfig();
		    
		    multicastConfig.setEnabled(false);
		    TcpIpConfig tcpIpConfig = new TcpIpConfig();
		    tcpIpConfig.setMembers(Arrays.asList("127.0.0.1"));
		    tcpIpConfig.setEnabled(true);
		    JoinConfig joinConfig = new JoinConfig();
		    joinConfig.setMulticastConfig(multicastConfig);
		    joinConfig.setTcpIpConfig(tcpIpConfig);
		    NetworkConfig networkConfig = new NetworkConfig();
		    networkConfig.setJoin(joinConfig);
		    GroupConfig groupConfig = new GroupConfig();
		    groupConfig.setName("stp-nonosgi");
		    groupConfig.setPassword("pass");
		    Config config = new Config();
		    config.setGroupConfig(groupConfig);
		    config.setNetworkConfig(networkConfig);
	        
		    return config;
		  }
	
	private HazelcastInstance instance;
	private UserTransactionManager utm;
	
	@Before
	public void setUp() throws Exception {
		Config config = hazelcastConfig();
		HazelcastInstance nativeInstance = HazelcastInstanceFactory.newHazelcastInstance(config);
		instance = AtomikosHazelcastInstanceFactory.createAtomikosInstance("hazelcast", nativeInstance);
		Configuration.getConfigProperties().setProperty(ConfigProperties.MAX_TIMEOUT_PROPERTY_NAME, "6000");
		utm = new UserTransactionManager();
		utm.setTransactionTimeout(5);
		utm.init();
	}
	
	@After
	public void teardown() {
		if (instance != null) instance.shutdown();
		utm.close();
	}

	@Test
	public void testCommit() throws Exception {
		utm.begin();
		TransactionContext ctx = instance.newTransactionContext();
		final TransactionalMap<Object, Object> map = ctx.getMap("map");
        map.put("key", "value");
		utm.commit();
		assertInMap("key", "value");
	}
	
	@Test(expected=TransactionNotActiveException.class)
	public void testUsingTransactionalMapAfterCommitThrows() throws Exception {
		utm.begin();
		TransactionContext ctx = instance.newTransactionContext();
		final TransactionalMap<Object, Object> map = ctx.getMap("map");
        map.put("key", "value");
		utm.commit();
		map.put("key", "value"); //forbidden
	}
	
	@Test
	public void testCallingNewTransactionContextTwiceIsAllowed() throws Exception {
		utm.begin();
		TransactionContext ctx1 = instance.newTransactionContext();
		TransactionContext ctx2 = instance.newTransactionContext();
		assertSame(ctx1,ctx2);
		utm.rollback();
	}
	
	@Test
	public void testRollback() throws Exception {
		utm.begin();
		TransactionContext ctx = instance.newTransactionContext();
		final TransactionalMap<Object, Object> map = ctx.getMap("map");
        map.put("key", "value");
		utm.rollback();
		assertInMap("key", null);
	}
	
	@Test(expected=UnsupportedOperationException.class)
	public void testCallingNewTransactionContextWithOptionsThrows() throws Exception {
		instance.newTransactionContext(new TransactionOptions());
	}
	
	@SuppressWarnings("rawtypes")
	private void assertInMap(String key, String value) throws Exception {
		Map map = instance.getMap("map");
		assertEquals(value, map.get(key));
	}

}
