@Test
	public void testMultipleMessagesMultipleSubscribers() {
		int noOfSubscribers = 2;
		int noOfMessages = 5;
		final HashMap<String, AtomicInteger> messages = new HashMap<String, AtomicInteger>();

		try {

			final String message = "TestMessage-testGetMessage_MultipleSubscribers";
			final CountDownLatch gate = new CountDownLatch(noOfSubscribers
					* noOfMessages);
			for (int i = 0; i < noOfSubscribers; i++) {
				AsyncMessageConsumer testSubscriber = new MockTopicSubscriber() {
					public void onMessage(Message received) {

						assertNotNull("Received Null Message", received);
						AtomicInteger count = messages.get(received
								.getMessage());
						assertNotNull(
								"Unexpected message received:"
										+ received.getMessage(), count);

						count.incrementAndGet();
						gate.countDown();
					}
				};
				destination.addSubscriber(testSubscriber);
			}

			for (int i = 0; i < noOfMessages; i++) {
				messages.put(message + i, new AtomicInteger(0));
			}
			for (String m : messages.keySet()) {
				Message msg = new MockMessage(m);
				destination.put(msg);
			}
			boolean messageReceived = gate.await(800, TimeUnit.MILLISECONDS);
			assertTrue("Did not receive message in 500ms", messageReceived);
			for (Entry<String, AtomicInteger> m : messages.entrySet()) {

				assertEquals("Unexpected number of messages received",
						noOfSubscribers, m.getValue().get());

			}

		} catch (InterruptedException e) {
			fail("Wait InterruptedException" + e.getMessage());
		} catch (DestinationClosedException e) {
			fail("DestinationClosedException" + e.getMessage());
		}

	}