Fanout service for Sparkleshare clients

2 files changed, 174 insertions, 1 deletions

diff --git a/tests/com/gitblit/tests/FanoutServiceTest.java b/tests/com/gitblit/tests/FanoutServiceTest.java
new file mode 100644
index 00000000..28e5d82d
--- /dev/null
+++ b/tests/com/gitblit/tests/FanoutServiceTest.java
@@ -0,0 +1,172 @@
+/*

+ * 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 static org.junit.Assert.assertEquals;

+

+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.FanoutService;

+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 {

+	

+	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());

+	}

+}
\ No newline at end of file
diff --git a/tests/com/gitblit/tests/GitBlitSuite.java b/tests/com/gitblit/tests/GitBlitSuite.java
index bb734eb7..5220a6a3 100644
--- a/tests/com/gitblit/tests/GitBlitSuite.java
+++ b/tests/com/gitblit/tests/GitBlitSuite.java
@@ -59,7 +59,8 @@ import com.gitblit.utils.JGitUtils;
 		MarkdownUtilsTest.class, JGitUtilsTest.class, SyndicationUtilsTest.class,

 		DiffUtilsTest.class, MetricUtilsTest.class, TicgitUtilsTest.class, X509UtilsTest.class,

 		GitBlitTest.class, FederationTests.class, RpcTests.class, GitServletTest.class,

-		GroovyScriptTest.class, LuceneExecutorTest.class, IssuesTest.class, RepositoryModelTest.class })

+		GroovyScriptTest.class, LuceneExecutorTest.class, IssuesTest.class, RepositoryModelTest.class,

+		FanoutServiceTest.class })

 public class GitBlitSuite {

 

 	public static final File REPOSITORIES = new File("git");