/*
 * Copyright 2013 gitblit.com.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.gitblit.tests;

import java.text.MessageFormat;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

import org.junit.Test;

import com.gitblit.fanout.FanoutClient;
import com.gitblit.fanout.FanoutClient.FanoutAdapter;
import com.gitblit.fanout.FanoutNioService;
import com.gitblit.fanout.FanoutService;
import com.gitblit.fanout.FanoutSocketService;

public class FanoutServiceTest extends GitblitUnitTest {

	int fanoutPort = FanoutService.DEFAULT_PORT;

	@Test
	public void testNioPubSub() throws Exception {
		testPubSub(new FanoutNioService(fanoutPort));
	}

	@Test
	public void testSocketPubSub() throws Exception {
		testPubSub(new FanoutSocketService(fanoutPort));
	}

	@Test
	public void testNioDisruptionAndRecovery() throws Exception {
		testDisruption(new FanoutNioService(fanoutPort));
	}

	@Test
	public void testSocketDisruptionAndRecovery() throws Exception {
		testDisruption(new FanoutSocketService(fanoutPort));
	}

	protected void testPubSub(FanoutService service) throws Exception {
		System.out.println(MessageFormat.format("\n\n========================================\nPUBSUB TEST {0}\n========================================\n\n", service.toString()));
		service.startSynchronously();

		final Map<String, String> announcementsA = new ConcurrentHashMap<String, String>();
		FanoutClient clientA = new FanoutClient("localhost", fanoutPort);
		clientA.addListener(new FanoutAdapter() {

			@Override
			public void announcement(String channel, String message) {
				announcementsA.put(channel, message);
			}
		});

		clientA.startSynchronously();

		final Map<String, String> announcementsB = new ConcurrentHashMap<String, String>();
		FanoutClient clientB = new FanoutClient("localhost", fanoutPort);
		clientB.addListener(new FanoutAdapter() {
			@Override
			public void announcement(String channel, String message) {
				announcementsB.put(channel, message);
			}
		});
		clientB.startSynchronously();


		// subscribe clients A and B to the channels
		clientA.subscribe("a");
		clientA.subscribe("b");
		clientA.subscribe("c");

		clientB.subscribe("a");
		clientB.subscribe("b");
		clientB.subscribe("c");

		// give async messages a chance to be delivered
		Thread.sleep(1000);

		clientA.announce("a", "apple");
		clientA.announce("b", "banana");
		clientA.announce("c", "cantelope");

		clientB.announce("a", "avocado");
		clientB.announce("b", "beet");
		clientB.announce("c", "carrot");

		// give async messages a chance to be delivered
		Thread.sleep(2000);

		// confirm that client B received client A's announcements
		assertEquals("apple", announcementsB.get("a"));
		assertEquals("banana", announcementsB.get("b"));
		assertEquals("cantelope", announcementsB.get("c"));

		// confirm that client A received client B's announcements
		assertEquals("avocado", announcementsA.get("a"));
		assertEquals("beet", announcementsA.get("b"));
		assertEquals("carrot", announcementsA.get("c"));

		clientA.stop();
		clientB.stop();
		service.stop();
	}

	protected void testDisruption(FanoutService service) throws Exception  {
		System.out.println(MessageFormat.format("\n\n========================================\nDISRUPTION TEST {0}\n========================================\n\n", service.toString()));
		service.startSynchronously();

		final AtomicInteger pongCount = new AtomicInteger(0);
		FanoutClient client = new FanoutClient("localhost", fanoutPort);
		client.addListener(new FanoutAdapter() {
			@Override
			public void pong(Date timestamp) {
				pongCount.incrementAndGet();
			}
		});
		client.startSynchronously();

		// ping and wait for pong
		client.ping();
		Thread.sleep(500);

		// restart client
		client.stop();
		Thread.sleep(1000);
		client.startSynchronously();

		// ping and wait for pong
		client.ping();
		Thread.sleep(500);

		assertEquals(2, pongCount.get());

		// now disrupt service
		service.stop();
		Thread.sleep(2000);
		service.startSynchronously();

		// wait for reconnect
		Thread.sleep(2000);

		// ping and wait for pong
		client.ping();
		Thread.sleep(500);

		// kill all
		client.stop();
		service.stop();

		// confirm expected pong count
		assertEquals(3, pongCount.get());
	}
}