@Test
	public void testGetMessageNoUnexpectedMessages() {
		try {
			int noOfMessages = 2;
			final String message = "TestMessage-testGetMessage_OneSubscriber";
			final CountDownLatch gate = new CountDownLatch(1);
			final AtomicInteger msgCount = new AtomicInteger(0);
			final Message expected = new MockMessage(message);
			AsyncMessageConsumer testSubscriber = new MockTopicSubscriber() {
				public void onMessage(Message received) {

					assertNotNull("Received Null Message", received);
					assertEquals("Message values are not equal", message,
							received.getMessage());
					msgCount.incrementAndGet();
					gate.countDown();
				}
			};
			destination.addSubscriber(testSubscriber);
			for (int i = 0; i < noOfMessages; i++)
				destination.put(expected);
			boolean messageReceived = gate.await(100, TimeUnit.MILLISECONDS);
			assertTrue("Did not receive message in 100ms", messageReceived);
			Callable<Integer> wait = new Callable<Integer>() {

				public Integer call() throws Exception {
					Thread.sleep(500);
					return msgCount.get();
				}
			};
			FutureTask<Integer> future = new FutureTask<Integer>(wait);
			future.run();
			assertEquals("Unexpected number of messages received",
					noOfMessages, future.get().intValue());
		} catch (InterruptedException e) {
			fail("Wait InterruptedException " + e.getMessage());
		} catch (ExecutionException e) {
			fail("Wait ExecutionException " + e.getMessage());
			e.printStackTrace();
		} catch (DestinationClosedException e) {
			fail("DestinationClosedException" + e.getMessage());
		}
	}