added working RabbitMQ test cases

doc/RabbitMQ.README

@@ -0,0 +1,12 @@
+On Windows, setting up RabbitMQ requires downloading Erlang and and RabbitMQ installer.
+Although on 64bit, the 64bit Erlang wouldn't work for me, and I installed the 32bit Erlang
+and RabbitMQ versions. Those worked ok.
+
+I had to set the following global environment variables:
+
+  ERLANG_HOME=c:/Program Files/erl5.9.1/erts-5.9.1/
+  RABBITMQ_NODENAME=rabbit@localhost
+
+The RABBITMQ_NODENAME setting avoids a strange issue that otherwise happened in my case
+when Erlang is trying to lookup my hostname and can't resolve it although it's resolvable
+using, e.g., ping.

java/com.sap.sailing.server.replication.test/src/com/sap/sailing/server/replication/test/SimpleRabbitMQTest.java

@@ -42,29 +42,71 @@ public class SimpleRabbitMQTest {
     
     @Test
     public void testSendReceiveHelloWorld() throws IOException, InterruptedException {
-        final Consumer consumer = new Consumer();
-        new Thread(consumer).start();
         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
+        final Consumer consumer = new QueueConsumer(QUEUE_NAME);
+        new Thread(consumer).start();
         String message = "Hello World!";
         channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
-        Thread.sleep(500);
+        consumer.waitUntilReceived(1000);
         assertEquals(message, received.get(consumer));
     }
     
-    private class Consumer implements Runnable {
+    @Test
+    public void testSendReceiveHelloWorldToTwoNodes() throws IOException, InterruptedException {
+        final String EXCHANGE_NAME = "updates";
+        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
+        final Consumer consumer1 = new ExchangeConsumer(EXCHANGE_NAME);
+        final Consumer consumer2 = new ExchangeConsumer(EXCHANGE_NAME);
+        new Thread(consumer1).start();
+        new Thread(consumer2).start();
+        String message = "Hello World!";
+        channel.basicPublish(EXCHANGE_NAME, /* queue name */ "", null, message.getBytes());
+        consumer1.waitUntilReceived(1000);
+        consumer2.waitUntilReceived(1000);
+        assertEquals(message, received.get(consumer1));
+        assertEquals(message, received.get(consumer2));
+    }
+    
+    private abstract class Consumer implements Runnable {
+        private final Connection connection;
+        private final Channel channel;
+        private final QueueingConsumer consumer;
+        private boolean receivedFinished;
+        
+        protected Consumer() throws IOException {
+            ConnectionFactory factory = new ConnectionFactory();
+            factory.setHost("localhost");
+            connection = factory.newConnection();
+            channel = connection.createChannel();
+            consumer = new QueueingConsumer(channel);
+        }
+        
+        public synchronized void waitUntilReceived(long timeoutInMillis) throws InterruptedException {
+            while (!receivedFinished) {
+                wait(timeoutInMillis);
+            }
+        }
+        
+        protected abstract String getQueueName();
+        
+        protected Channel getChannel() {
+            return channel;
+        }
+        
         @Override
         public void run() {
             try {
-                ConnectionFactory factory = new ConnectionFactory();
-                factory.setHost("localhost");
-                Connection connection;
-                connection = factory.newConnection();
-                Channel channel = connection.createChannel();
-                channel.queueDeclare(QUEUE_NAME, false, false, false, null);
-                QueueingConsumer consumer = new QueueingConsumer(channel);
-                channel.basicConsume(QUEUE_NAME, true, consumer);
+                channel.basicConsume(getQueueName(), true, consumer);
                 QueueingConsumer.Delivery delivery = consumer.nextDelivery();
-                received.put(this, new String(delivery.getBody()));
+                final String s = new String(delivery.getBody());
+                received.put(this, s);
+                synchronized (this) {
+                    receivedFinished = true;
+                    notifyAll();
+                }
+                System.out.println("received "+s+" in "+this);
+                channel.close();
+                connection.close();
             } catch (IOException e) {
                 e.printStackTrace();
             } catch (ShutdownSignalException e) {
@@ -76,4 +118,35 @@ public class SimpleRabbitMQTest {
             }
         }
     }
+    
+    private class QueueConsumer extends Consumer {
+        private final String queueName;
+
+        public QueueConsumer(String queueName) throws IOException {
+            super();
+            this.queueName = queueName;
+            getChannel().queueDeclare(queueName, false, false, false, null);
+        }
+
+        @Override
+        protected String getQueueName() {
+            return queueName;
+        }
+    }
+    
+    private class ExchangeConsumer extends Consumer {
+        private final String queueName;
+
+        public ExchangeConsumer(String exchangeName) throws IOException {
+            super();
+            this.queueName = getChannel().queueDeclare().getQueue();
+            getChannel().exchangeDeclare(exchangeName, "fanout");
+            getChannel().queueBind(getQueueName(), exchangeName, "");
+        }
+
+        @Override
+        protected String getQueueName() {
+            return queueName;
+        }
+    }
 }