import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.is;
import static org.junit.Assert.assertThat;

import java.util.LinkedList;
import java.util.List;

import org.infinispan.Cache;
import org.infinispan.configuration.cache.CacheMode;
import org.infinispan.configuration.cache.ConfigurationBuilder;
import org.infinispan.configuration.global.GlobalConfigurationBuilder;
import org.infinispan.manager.DefaultCacheManager;
import org.infinispan.notifications.Listener;
import org.infinispan.notifications.cachelistener.annotation.CacheEntryCreated;
import org.infinispan.notifications.cachelistener.annotation.CacheEntryModified;
import org.infinispan.notifications.cachelistener.annotation.CacheEntryRemoved;
import org.infinispan.notifications.cachelistener.event.CacheEntryCreatedEvent;
import org.infinispan.notifications.cachelistener.event.CacheEntryEvent;
import org.infinispan.notifications.cachelistener.event.CacheEntryModifiedEvent;
import org.infinispan.notifications.cachelistener.event.CacheEntryRemovedEvent;
import org.infinispan.transaction.TransactionMode;
import org.junit.Test;

public class TestInfinispanDuplicatedEvents {

	static {
		System.setProperty("java.net.preferIPv4Stack", "true");
	}
	
	@Listener
	public class MyCacheListener {
		
		private List<CacheEntryEvent<String, String>> events = new LinkedList<CacheEntryEvent<String,String>>();
		
		@CacheEntryCreated
		public void created(CacheEntryCreatedEvent<String, String> event) {
			events.add(event);
		}
		
		@CacheEntryModified
		public void modified(CacheEntryModifiedEvent<String, String> event) {
			events.add(event);
		}
		
		@CacheEntryRemoved
		public void removed(CacheEntryRemovedEvent<String, String> event) {
			events.add(event);
		}
	}
	
	@Test
	public void must_raise_create_event_once() {
		DefaultCacheManager nodeA = configureNode();
		
		Cache<String, String> cacheA = nodeA.getCache();
		
		MyCacheListener listenerA = new MyCacheListener();
		cacheA.addListener(listenerA);
		
		DefaultCacheManager nodeB = configureNode();
		
		Cache<String, String> cacheB = nodeB.getCache();
		
		MyCacheListener listenerB = new MyCacheListener();
		cacheB.addListener(listenerB);
		
		cacheA.put("a", "a");
		
		/*
		 * We expect 4 events on both nodes: pre-create, pre-modified, post-modified, post-create
		 */
		assertThat(listenerA.events.size(), is(4));
		assertThat(listenerB.events.size(), is(4));
		
		checkEvents(listenerA, "a");
		checkEvents(listenerB, "a");

		/*
		 * So far so good, let's try another key, say "b"
		 */
		
		listenerA.events.clear();
		listenerB.events.clear();
		
		cacheA.put("b", "b");
		
		/*
		 * We expect 4 events again
		 */
		assertThat(listenerA.events.size(), is(4));
		assertThat(listenerB.events.size(), is(4));
		
		checkEvents(listenerA, "b");
		checkEvents(listenerB, "b");
		
		/*
		 * Let's try another one, say "a0"
		 */
		
		listenerA.events.clear();
		listenerB.events.clear();
		
		cacheA.put("a0", "a0");
		
		/*
		 * We expect another 4 events, but on the local node (A in this case) we get 8
		 */
		assertThat(listenerA.events.size(), is(4));
		assertThat(listenerB.events.size(), is(4));
		
		checkEvents(listenerA, "a0");
		checkEvents(listenerB, "a0");
	}

	private DefaultCacheManager configureNode() {
		return new DefaultCacheManager(GlobalConfigurationBuilder.defaultClusteredBuilder().transport().globalJmxStatistics().allowDuplicateDomains(true)
				.build(), new ConfigurationBuilder().clustering().cacheMode(CacheMode.REPL_SYNC).transaction().transactionMode(TransactionMode.TRANSACTIONAL).build());
	}

	private void checkEvents(MyCacheListener listenerA, String expectedKey) {
		assertThat(listenerA.events.get(0), is(instanceOf(CacheEntryCreatedEvent.class)));
		assertThat(listenerA.events.get(0).getKey(), is(expectedKey));
		assertThat(listenerA.events.get(0).isPre(), is(true));
		
		assertThat(listenerA.events.get(1), is(instanceOf(CacheEntryModifiedEvent.class)));
		assertThat(listenerA.events.get(1).getKey(), is(expectedKey));
		assertThat(listenerA.events.get(1).isPre(), is(true));
		
		assertThat(listenerA.events.get(2), is(instanceOf(CacheEntryModifiedEvent.class)));
		assertThat(listenerA.events.get(2).getKey(), is(expectedKey));
		assertThat(listenerA.events.get(2).isPre(), is(false));
		
		assertThat(listenerA.events.get(3), is(instanceOf(CacheEntryCreatedEvent.class)));
		assertThat(listenerA.events.get(3).getKey(), is(expectedKey));
		assertThat(listenerA.events.get(3).isPre(), is(false));
	}
}