You've settled on Mockito (http://mockito.org) as your mocking framework of choice, because it's able to mock both interfaces and concrete classes (the RabbitMQ client contains both), ties perfectly with JUnit, has an awesome syntax, and "Does The Right Thing™" by default! Because Mockito works great with JUnit, you'll be able to run these unit tests as part of your Maven build using the standard Surefire plug-in (http://maven.apache.org/surefire/maven-surefire-plugin/).

Let's focus on unit testing the RabbitMqManager class, as it's the foremost class of the Java application. You first need to create the test class and initialize the SUT (System Under Test). This is how you do it:

@RunWith(MockitoJUnitRunner.class)
public class RabbitMqManagerTest
{
    private static final Address[] TEST_ADDRESSES = {new Address("fake")};

    @Mock
    private ConnectionFactory connectionFactory;

    @Mock
    private Connection connection;

    @Mock
    private Channel channel;

    private RabbitMqManager rabbitMqManager;

    @Before
    public void initialize() throws Exception
    {
        rabbitMqManager = new RabbitMqManager(connectionFactory, TEST_ADDRESSES);
        when(connection.getAddress()).thenReturn(InetAddress.getLocalHost());
    }

As you can see, you declared mocks for the principal classes that are involved when dealing with RabbitMQ: connectionFactory, connection, and channel. Then you initialize the RabbitMQManager class with the array that contains a fake address. It's fine, because no actual connection attempt will be made since we're mocking the RabbitMQ classes. You may wonder whether you should have several addresses there in order to test the connection fall-back mechanism onto which the cluster client relies. The answer is no; this is a behavior provided by the actual RabbitMQ client library and not by any of your code.

You've also attached a global behavior to the getAddress() method of the connection mock, so it returns a valid InetAddress instance, which spares you from doing it again and again in all the tests. Let's now detail a few tests that you've written to exercise the start() method:

@Test
public void startFailure() throws Exception
{
    when(connectionFactory.newConnection(TEST_ADDRESSES))
        .thenThrow( new RuntimeException("simulated failure"));

    rabbitMqManager.start();

    final List<Runnable> scheduleReconnection = rabbitMqManager.getExecutor().shutdownNow();

    assertThat(scheduleReconnection.size(), is(1));
}

In this test, you've first configured the connectionFactory mock to throw an exception when asked to create a new connection. This will simulate an issue when communicating with RabbitMQ. Note that you're throwing RuntimeException and not IOException, which is the checked exception that is thrown by newConnection(). This is because you want to ensure that your code can actually handle any exception that could bubble up through this method call, which is its intended behavior. You've also made it clear in the exception message that it is an intentional one.

After that, you actually call the start() method, which is the main purpose of this test. If you remember its behavior from Chapter 2, Creating an Application Inbox, the start() method should have scheduled a reconnection task in case of a connection failure. That's why you shut down the executor encapsulated by the RabbitMqManager class and assert that it actually contained a scheduled Runnable instance. At this point, you're happy with this first test and move on to testing a successful start attempt as follows:

@Test
public void startSuccess() throws Exception
{
    when(connectionFactory.newConnection(TEST_ADDRESSES))
        .thenReturn(connection);

    rabbitMqManager.start();

    verify(connection).addShutdownListener(rabbitMqManager);
}

This test is short and sweet: the connectionFactory mock is configured to return the connection mock. The only assertion you've added is a verification of the fact that RabbitMqManager has registered itself as a shutdown listener on the connection mock. This is enough to ensure that the expected behavior kicks in when a connection is successfully created. Finally, you add another test to verify that the actual reconnection mechanism works as follows:

@Test
public void startFailureThenSuccess() throws Exception
{

    when(connectionFactory.newConnection(TEST_ADDRESSES))
      .thenThrow( new RuntimeException("simulated connection failure"))
      .thenReturn(connection);

    rabbitMqManager.setReconnectDelaySeconds(0);
 
    rabbitMqManager.start();

    Thread.sleep(100L);

    verify(connectionFactory, times(2)).newConnection(TEST_ADDRESSES);
    verify(connection).addShutdownListener(rabbitMqManager);
    verifyNoMoreInteractions(connectionFactory);
}

Let's detail the notable bits in this test. First, you configured the connectionFactory mock to initially fail then succeed when asked to create a new connection. Then you set the reconnection delay to be 0 so that RabbitMqManager retries right away. After starting it, you ponder for a few milliseconds before asserting that everything went fine. It's a little unfortunate that Thread.sleep has to be used, but there is no testing seam that we could use to register a synchronization primitive to block the testing thread just at the needed time.

The verifications in this test ensure that the newConnection method has been called twice, that the RabbitMqManager class has registered itself as a shutdown listener (as it should in case of connection success), and that no other interaction has occurred with connectionFactory (as expected after a connection success).

Many other tests are needed in order to exercise the channel creation and closure and the subscription management features of RabbitMqManager, but we will just detail one extra test. The following is what you've written to test a successful execution of a ChannelCallable instance via the call method:

@Test
public void callSuccess() throws Exception
{
    when(channel.isOpen()).thenReturn(true);
    when(connection.createChannel()).thenReturn(channel);
   rabbitMqManager.setConnection(connection);

    final Channel expectedChannel = channel;

    final String result = rabbitMqManager.call(new ChannelCallable<String>()
    {
        @Override
        public String getDescription()
        {
            return "success";
        }

        @Override
        public String call(final Channel channel) throws IOException
        {
            return channel == expectedChannel ? "ok" : "bad";
        }
    });

    assertThat(result, is("ok"));

    verify(channel).close();
}

Let's detail this test bit by bit as follows:

With this approach, you've been able to achieve almost 100 percent test coverage for your classes using the RabbitMQ client SDK. This is a great position to be in as it will ensure that any breaking change will be caught at development time. This said, you also want to add tests that will exercise your code against a real broker instead of using mocks. It's time to write some integration tests.

Unlike what you've just done with unit tests, which are aware of all the internals of the tested application, integration tests focus on testing a system as a whole. They are sometimes referred to as black-box testing, which is opposite to white-box or clear-box testing. Your goal with integration testing your RabbitMQ application is to gain confidence that not only are things wired up internally as they should be, but also that they really work as intended.

You will use JUnit to run the integration tests so that they're developed in a familiar fashion, are automated, and can easily be run from a programming environment. You will also use the Failsafe Maven plugin (https://maven.apache.org/surefire/maven-failsafe-plugin/) to run these tests as part of your Maven build, but only as part of a specific profile. Indeed, you do not want these tests to run by default; otherwise, the build may fail when run on a machine where RabbitMQ is not running (like your continuous integration server).

Let's look in detail at the test that validates that the subscription mechanism works well. First, you need to create and configure the SUT, which is again an instance of RabbitMqManager, but this time configured to connect to a live RabbitMQ broker as follows:

public class RabbitMqManagerIT
{
    private RabbitMqManager rabbitMqManager;

    @Before
    public void configureAndStart() throws Exception
    {
        final ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setUsername(System.getProperty("test.rmq.username", "ccm-dev"));
        connectionFactory.setPassword(System.getProperty("test.rmq.password", "coney123"));
        connectionFactory.setVirtualHost(System.getProperty("test.rmq.vhost", "ccm-dev-vhost"));

        final String addresses = System.getProperty("test.rmq.addresses", "localhost:5672");

        rabbitMqManager = new RabbitMqManager(connectionFactory, Address.parseAddresses(addresses));

        System.out.printf("%nRunning integration tests on %s%n%n", addresses);

        rabbitMqManager.start();
    }

    @After
    public void stop() throws Exception
    {
        rabbitMqManager.stop();
    }

As you can see, before reaching the point where RabbitMqManager gets instantiated, we extract configuration values from system properties, falling back to default values that point to a locally running RabbitMQ broker. This approach allows you to point the tests at any broker by simply providing different connection parameters to the test.

Now let's look at the test itself. It's pretty long, so before delving into the code, let's talk about what it will do. You want to test if the subscription mechanism works. For this, you will: create a test queue, subscribe to it, send a message to it, and finally assert that the message has been consumed by the subscriber. With this said, let's look at the first part of the test, which takes care of setting the test queue as follows:

@Test
public void subscriptionTest() throws Exception
{
    final String queue = rabbitMqManager.call(new ChannelCallable<String>()
    {
        @Override
        public String getDescription()
        {
            return "subscription test setup";
        }

        @Override
        public String call(final Channel channel) throws IOException
        {
            final DeclareOk declareOk = channel.queueDeclare("", false, true, true, null);
            return declareOk.getQueue();
        }
    });

The important bit here is that you create an automatically named, exclusive, nondurable, and auto-delete queue. Why these options? You want the queue name to be unique so that there is no possible collision if another developer runs the test on the same broker. For further protection from any risk of having another consumer interacting with this queue, you made it exclusive. Finally, you do not want either the messages or the queue itself to be retained when you're done with the test, hence the nondurability and autodeletion attributes. Next, we create the subscription as follows:

final AtomicReference<byte[]> delivered = new AtomicReference<byte[]>();
final CountDownLatch latch = new CountDownLatch(1);
final Subscription subscription = rabbitMqManager.createSubscription(queue,
    new SubscriptionDeliverHandler()
    {
        @Override
        public void handleDelivery(final Channel channel,
          final Envelope envelope,
          final BasicProperties properties,
          final byte[] body)
        {
          delivered.set(body);
          latch.countDown();
        }
    });
 
assertThat(subscription.getChannel().isOpen(), is(true));

This bit is very interesting because here you can use synchronization primitives instead of having to resort to a sleep statement as you did before. Indeed, the latch will allow you to block the test thread until the handleDelivery method has been called and the message body value set on the delivered atomic reference. Without this mechanism, there would be no way to check what message was delivered or when it was delivered, since it is done by a thread other than the testing thread. That said, you can right away assert that the subscription encapsulates an open channel. With this in place, it's now time to send a test message to the queue as follows:

final byte[] body = rabbitMqManager.call(new ChannelCallable<byte[]>()
{
    @Override
    public String getDescription()
    {
        return "publish test message";
    }

    @Override
    public byte[] call(final Channel channel) throws IOException
    {
        final byte[] body = UUID.randomUUID().toString().getBytes();
        channel.basicPublish("", queue, null, body);
        return body;
    }
});

Nothing particularly novel here. You may wonder why we are creating a random body payload for the test message. This is to ensure that the test will be consuming the right message and not a message that could be lingering from a previous test. Notice that you have to target the default exchange to be able to send to the test queue directly, which frees you from the need to declare a test exchange and bind the queue to it.

The following final code fragment is where the assertions happen:

if (!latch.await(1, TimeUnit.MINUTES))
{
    fail("Handler not called on time");
}

assertThat(delivered.get(), is(body));

subscription.stop();
assertThat(subscription.getChannel(), is(nullValue()));

Did you see how you've leveraged latch to block the test thread until the message gets delivered? Nothing should ever be blocked forever, so you've been wise enough to cap the waiting period to one minute and fail the test if no message has been received after that.

At this point, you've covered your bases in term of testing. You're now pretty confident that any regression will be caught at an early stage thanks to the barrage of unit and integration tests you've created.

When adding new features, it's sometimes convenient to step-debug into an application to trace its execution. The next section will detail how to achieve this with RabbitMQ.