123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169 |
- /*
- * 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());
- }
- }
|