Blog

This post is about using Nord Pool push notifications to receive urgent market messages, UMMs, in a Java application. It is mainly of interest for people in the energy sector, but could also be useful to someone looking to receive notifications using SignalR in a Java environment.

Nord Pool

Nord Pool is a European market for trading power (electricity), owned by Nasdaq.

Urgent Market Messages

An urgent market message, or UMM, is used in the energy sector to inform the market about planned and unplanned events that affect the available power in production, consumption or transmission units. For example, information about a planned maintenance of a nuclear power plant would be sent as a UMM, as would information about a power line connecting two countries being cut by mistake.

Nord Pool aggregates UMMs from European power companies and provide a REST API to get UMMs. They also provide push notifications to asynchronously receive real-time UMM information using SignalR.

SignalR

SignalR is a Microsoft library for sending asynchronous notifications from servers to clients using standard web protocols.

There are two versions of SignalR that are not compatible: ASP.NET SignalR and ASP.NET Core SignalR. The Nord Pool push notifications use the older ASP.NET SignalR version, so it is important to use a client that supports that version.

In this case, we’re looking for a Java client, and luckily there is one available on GitHub. The readme for this project marks it as obsolete, and points to a version that supports the newer ASP.NET Core SignalR. However, since we want to connect to Nord Pool we will ignore the warning and use the old version.

There does not seem to be any version of the Java client uploaded to a Maven repository, so the first step is to clone the Git repository and build the code locally. There are Gradle build scripts available; I did not have Gradle available where I’m currently working, so I added simple POM files and built using Maven. The only module that is required for what we are doing here is signalr-client-sdk.

Listening for Notifications Using the Java Client

Let’s develop a Java domain service that listens for UMM notifications and calls registered handlers when UMM events occur. We start by defining the service interface that allows you to register new handlers, and start listening:

public interface UmmNotificationService {
    void registerNotificationHandler(UmmNotificationHandler notificationHandler);
    void startListeningForNotifications();
}

The Nord Pool push notification documentation describes three types of events:

• New message
• Update message
• Cancel / dismiss message

We therefore define a notification handler interface with three methods corresponding to the events:

public interface UmmNotificationHandler {
    void onNewMessage(String jsonUmm);
    void onUpdateMessage(String jsonUmm);
    void onDismissMessage(String jsonUmm);
}

Note that the methods are defined to take strings with the JSON representation of the UMMs. This is in order to make the example simpler; in a real system you would normally create a Umm entity and a factory class UmmFactory to create Umm objects from their JSON representation.

We can now create an empty implementation of the service interface, so that we can start writing an integration test (note that we anticipate that we will need to define which URL to connect to):

public class NordpoolUmmNotificationService implements UmmNotificationService {
    public NordpoolUmmNotificationService(String ummPushUrl) {
    }

    @Override
    public void registerNotificationHandler(UmmNotificationHandler notificationHandler) {
    }

    @Override
    public void startListeningForNotifications() {
    }
}

We are now in a position where we can create an integration test for our new service. Here I have to admit that I don’t really know how to create a solid, reliable and repeatable test. At the moment, the best I know how to do is to connect to Nord Pool, wait until a UMM push notification arrives and then verify that what we received was a correct UMM:

public class NordpoolUmmNotificationServiceIT {

    @Test
    public void receiveNotifications() throws Exception {
        NordpoolUmmNotificationService service =
                new NordpoolUmmNotificationService("https://ummws.nordpoolgroup.com");
        TestUmmNotificationHandler notificationHandler = new TestUmmNotificationHandler();
        service.registerNotificationHandler(notificationHandler);
        service.startListeningForNotifications();
        notificationHandler.latch.await();
        String umm = notificationHandler.umms.get(0);
        // Parse JSON and verify that it really is a UMM.
    }

    private static final class TestUmmNotificationHandler implements UmmNotificationHandler {
        private CountDownLatch latch = new CountDownLatch(1);
        private List<String> umms= new ArrayList<>();

        @Override
        public void onNewMessage(String umm) {
            umms.add(umm);
            latch.countDown();
        }

        @Override
        public void onUpdateMessage(String umm) {
            umms.add(umm);
            latch.countDown();
        }

        @Override
        public void onDismissMessage(String umm) {
            umms.add(umm);
            latch.countDown();
        }
    }
}

Note that new UMMs are produced only a few per hour, so the above test may take a very long time to run.

After some trial and error, the following implementation turns out to make the test pass:

import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.gson.JsonElement;
import microsoft.aspnet.signalr.client.LogLevel;
import microsoft.aspnet.signalr.client.hubs.HubConnection;
import microsoft.aspnet.signalr.client.hubs.HubProxy;
import microsoft.aspnet.signalr.client.transport.ClientTransport;
import microsoft.aspnet.signalr.client.transport.ServerSentEventsTransport;

public class NordpoolUmmNotificationService implements UmmNotificationService {

    private static final Logger LOG = LoggerFactory.getLogger(NordpoolUmmNotificationService.class);

    private String ummPushUrl;

    private Set<UmmNotificationHandler> notificationHandlers = new CopyOnWriteArraySet<>();

    public NordpoolUmmNotificationService(String ummPushUrl) {
        LOG.info("Creating new NordpoolUmmNotificationService: ummPushUrl={}", ummPushUrl);
        this.ummPushUrl = ummPushUrl;
    }

    @Override
    public void registerNotificationHandler(UmmNotificationHandler notificationHandler) {
        LOG.info("Registering a new notification handler: {}", notificationHandler);
        notificationHandlers.add(notificationHandler);
    }

    @Override
    public void startListeningForNotifications() {
        LOG.info("Start listening for notifications");
        microsoft.aspnet.signalr.client.Logger logger = new Slf4jLogger(LOG);
        HubConnection connection = new HubConnection(ummPushUrl, "", true, logger);
        ClientTransport clientTransport = new ServerSentEventsTransport(logger);

        HubProxy proxy = connection.createHubProxy("MessageHub");
        proxy.on("newMessage", data -> {
            onNewMessage(data.toString());
        }, JsonElement.class);
        proxy.on("updateMessage", data -> {
            onUpdateMessage(data.toString());
        }, JsonElement.class);
        proxy.on("dismissMessage", data -> {
            onDismissMessage(data.toString());
        }, JsonElement.class);

        connection.start(clientTransport);
    }

    private void onNewMessage(String umm) {
        for (UmmNotificationHandler notificationHandler : notificationHandlers) {
            LOG.debug("Calling onNewMessage on notification handler: umm={}, notificationHandler={}", umm,
                    notificationHandler);
            notificationHandler.onNewMessage(umm);
        }
    }

    private void onUpdateMessage(String umm) {
        for (UmmNotificationHandler notificationHandler : notificationHandlers) {
            LOG.debug("Calling onUpdateMessage on notification handler: umm={}, notificationHandler={}", umm,
                    notificationHandler);
            notificationHandler.onUpdateMessage(umm);
        }
    }

    private void onDismissMessage(String umm) {
        for (UmmNotificationHandler notificationHandler : notificationHandlers) {
            LOG.debug("Calling onDismissMessage on notification handler: umm={}, notificationHandler={}", umm,
                    notificationHandler);
            notificationHandler.onDismissMessage(umm);
        }
    }

    /**
     * An adapter class that takes an Slf4j logger and turns it into a
     * <code>microsoft.aspnet.signalr.client.Logger</code>.
     */
    private static final class Slf4jLogger implements microsoft.aspnet.signalr.client.Logger {

        private final Logger slf4jLogger;

        Slf4jLogger(Logger slf4jLogger) {
            this.slf4jLogger = slf4jLogger;
        }

        @Override
        public void log(String message, LogLevel level) {
            switch (level) {
            case Critical:
                slf4jLogger.error(message);
                break;
            case Information:
                slf4jLogger.info(message);
                break;
            case Verbose:
                slf4jLogger.debug(message);
                break;
            default:
                throw new IllegalStateException("Unknonwn enum constant: " + level);
            }

        }
    }
}

A few notes regarding the code above:

• The SignalR Java client adds /signalr to the URL you connect to, so in the case of Nord Pool you use the URL https://ummws.nordpoolgroup.com to connect.
• The name used in the call connection.createHubProxy must be "MessageHub", otherwise you receive HTTP 500, Interval Server Error, from Nord Pool.
• The default client transport will use a WebsocketTransport. I have not got this to work, possibly because we are behind an HTTP proxy. The ServerSentEventsTransport does work, however.

This post describes a simple way to create a test version of a service that reads JSON or XML from a REST service or similar. The purpose is to easily create a fake service that reads from files instead and that can be used for testing other code that use the service.

I believe in using production code as much as possible when running tests. Every time you use special test code, i.e., code that is only used for testing, you run the risk of the test code not behaving exactly the same as the production code. You also get a maintenance problem, where the test code must be kept up to date with respect to the production code.

Some types of code are inconvenient to use for testing, however. For example, database calls require setup and may be slow, and code calling REST services require the service to be available and again, may be slow. In a previous post, we saw a simple way to replace repositories calling a database with an in-memory version. In this post, we will see how to replace code calling a REST service with a version reading from file.

When creating a fake version of a piece of code, there are two things to keep in mind:

• The less of the production code you replace with test code, the easier it is to keep the two in sync.
• The test version should be tested using the same test suite as the production code, to verify that the two behave identically.

Getting Started

A sample project, rld-rest-sample, accompanying this post can be found on GitHub.

Use the following commands to download, build and run the sample project:

$ mkdir reallifedeveloper
$ cd reallifedeveloper
$ git clone https://github.com/reallifedeveloper/rld-rest-sample.git
$ cd rld-rest-sample
$ mvn -DcheckAll clean install # Should end with BUILD SUCCESS
$ java -Dserver.port=8081 -jar target/rld-rest-sample-1.0.jar

You can now try the following URLs to see that everything is working:

• http://localhost:8081/countries
• http://localhost:8081/countries/IND

Example Code

Assume that we want to create a REST service that can list the countries of the world, and also the states of a particular country. The main reason that this was chosen as example is that there are free online services that we can use for testing.

First of all, we define the CountryService interface:

package com.reallifedeveloper.sample.domain;

import java.io.IOException;
import java.util.List;

public interface CountryService {

    List<Country> allCountries() throws IOException;

    List<State> statesOfCountry(String alpha3Code) throws IOException;

}

We now want to create an implementation of the CountryService interface that uses the free services mentioned above, from a site called GroupKT. We call this implementation GroupKTCountryService. To get started, we create integration tests that connect to the REST services and define the behavior we expect:

package com.reallifedeveloper.sample.infrastructure;

// imports...

public class GroupKTCountryServiceIT {

    @Rule
    public ExpectedException expectedException = ExpectedException.none();

    private GroupKTCountryService service = new GroupKTCountryService("http://services.groupkt.com");

    @Test
    public void allCountries() throws Exception {
        List<Country> allCountries = service().allCountries();
        assertThat(allCountries, notNullValue());
        assertThat(allCountries.size(), is(249));
    }

    @Test
    public void indiaShouldHave36States() throws Exception {
        List<State> statesOfIndia = service().statesOfCountry("IND");
        assertThat(statesOfIndia, notNullValue());
        assertThat(statesOfIndia.size(), is(36));
    }

    @Test
    public void unknownCountryShouldGiveEmptyList() throws Exception {
        List<State> statesOfUnknownCountry = service().statesOfCountry("foo");
        assertThat(statesOfUnknownCountry, notNullValue());
        assertThat(statesOfUnknownCountry.isEmpty(), is(true));
    }

    @Test
    public void nullCountryShouldThrowException() throws Exception {
        expectedException.expect(IllegalArgumentException.class);
        expectedException.expectMessage("alpha3Code must not be null");
        service().statesOfCountry(null);
    }

    @Test
    public void constructorNullBaseUrlShouldThrowException() {
        expectedException.expect(IllegalArgumentException.class);
        expectedException.expectMessage("baseUrl must not be null");
        new GroupKTCountryService(null);
    }

    protected CountryService service() {
        return service;
    }
}

Note the protected service method that will be used later when we test the file version of the service.

The GroupKTCountryService that is created together with the integration test is as follows:

package com.reallifedeveloper.sample.infrastructure;

// imports...

public class GroupKTCountryService implements CountryService {

    private final String baseUrl;

    public GroupKTCountryService(String baseUrl) {
        if (baseUrl == null) {
            throw new IllegalArgumentException("baseUrl must not be null");
        }
        this.baseUrl = baseUrl;
    }

    @Override
    public List<Country> allCountries() throws IOException {
        String jsonCountries = jsonAllCountries();
        ObjectMapper objectMapper = new ObjectMapper();
        RestResponseWrapper<Country> countriesResponse =
                objectMapper.readValue(jsonCountries, new TypeReference<RestResponseWrapper<Country>>() {});
        return countriesResponse.restResponse.result;
    }

    @Override
    public List<State> statesOfCountry(String alpha3Code) throws IOException {
        if (alpha3Code == null) {
            throw new IllegalArgumentException("alpha3Code must not be null");
        }
        String jsonStates = jsonStatesOfCountry(alpha3Code);
        ObjectMapper objectMapper = new ObjectMapper();
        RestResponseWrapper<State> statesResponse =
                objectMapper.readValue(jsonStates, new TypeReference<RestResponseWrapper<State>>() {});
        return statesResponse.restResponse.result;
    }

    protected String jsonAllCountries() throws IOException {
        RestTemplate restTemplate = new RestTemplate();
        return restTemplate.getForObject(baseUrl() + "/country/get/all", String.class);
    }

    protected String jsonStatesOfCountry(String alpha3Code) throws IOException {
        RestTemplate restTemplate2 = new RestTemplate();
        String stateUrl = baseUrl() + "/state/get/" + alpha3Code + "/all";
        return restTemplate2.getForObject(stateUrl, String.class);
    }

    protected String baseUrl() {
        return baseUrl;
    }

    private static final class RestResponseWrapper<T> {
        private final RestResponse<T> restResponse;

        @JsonCreator
        RestResponseWrapper(@JsonProperty("RestResponse") RestResponse<T> restResponse) {
            this.restResponse = restResponse;
        }

        private static final class RestResponse<T> {
            private final List<String> messages;
            private final List<T> result;

            @JsonCreator
            RestResponse(@JsonProperty("messages") List<String> messages,
                    @JsonProperty("result") List<T> result) {
                this.messages = messages;
                this.result = result;
            }
        }
    }
}

Note the protected jsonAllCountries and jsonStatesOfCountry methods that return a JSON string representing the different types of information. These methods are overridden in the FileCountryService that reads JSON from files instead of over HTTP:

package com.reallifedeveloper.sample.infrastructure;

import java.io.IOException;

import com.reallifedeveloper.tools.test.TestUtil;

public class FileCountryService extends GroupKTCountryService {

    public FileCountryService(String baseUrl) {
        super(baseUrl);
    }

    @Override
    protected String jsonAllCountries() throws IOException {
        return TestUtil.readResource(baseUrl() + "/all_countries.json");
    }

    @Override
    protected String jsonStatesOfCountry(String alpha3Code) throws IOException {
        return TestUtil.readResource(baseUrl() + "/states_" + alpha3Code + ".json");
    }

}

The TestUtil.readResource method comes from rld-build-tools that is available from the central Maven repository and from GitHub. The method simply reads a file from classpath and returns its contents as a string.

We also need to add a few JSON files under src/test/resources/json:

• all_countries.json that contains a list of all countries
• states_IND.json that contains the states of India
• states_foo.json that contains an empty response

To test the FileCountryService, we use the same test cases as for the GroupKTCountryService, so we create a FileCountryServiceTest that inherits from GroupKTCountryServiceIT but plugs in a FileCountryService to test instead of a GroupKTCountryService:

package com.reallifedeveloper.sample.infrastructure;

import com.reallifedeveloper.sample.domain.CountryService;

public class FileCountryServiceTest extends GroupKTCountryServiceIT {

    private FileCountryService service = new FileCountryService("json");

    @Override
    protected CountryService service() {
        return service;
    }
}

We can now use the FileCountryService when testing other code, for example application services or REST resources that use the service. We can be sure that it behaves like the real service since we run the same test suite on the two.

Packaging the Code

The test versions of your services, and the JSON or XML response files that you provide, should normally be under src/test and will therefore not be available in the jar file created. If you need to use the test versions of services in other projects, you can easily configure Maven to create a jar file containing your test code:

                <!-- Always generate a *-tests.jar with all test code -->
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-jar-plugin</artifactId>
                    <version>3.0.2</version>
                    <executions>
                        <execution>
                            <goals>
                                <goal>test-jar</goal>
                            </goals>
                        </execution>
                    </executions>
                </plugin>

This will create a file called something like rld-rest-sample-1.0-tests.jar containing the test code.

In other projects where you want to use the test versions of the services, add a dependency of type test-jar:

        <dependency>
            <groupId>com.reallifedeveloper</groupId>
            <artifactId>rld-rest-sample</artifactId>
            <version>1.0</version>
            <type>test-jar</type>
            <scope>test</scope>
        </dependency>

You can now use the test services and the packaged JSON or XML files when testing your other projects.

Summary

When you create a service that reads JSON or XML, isolate the methods that read over the network. Create a test version of the service that substitutes the methods with methods that read from local files instead. Also provide a few JSON or XML files that contain the responses you currently need during testing. It is easy to add more response files later if you need to add new test cases. Make sure that you run the same test cases on the file version of the service that you run on the real version.

Following these simple recommendations gives you a test version of the service that runs quickly and reliably, and that is guaranteed to be kept up to date with respect to the real version.

In a previous post, we saw some Java code for redelivering messages from a queue to an exchange in RabbitMQ. Obviously, a test was written before writing the actual code.

What kind of test is appropriate in this situation? What we want to test is that messages that are in a RabbitMQ queue are removed from that queue and instead available in the queues that are bound to the exchange we move the messages to.

In this case, the important thing is how RabbitMQ behaves as a result of executing our code, so an integration test that connects to a test instance of RabbitMQ running on the developer machine is the right solution. I strongly believe that every developer should have access to a personal instance of the systems that they need to integrate with, as far as possible. This means that the developers should have their own databases, message queues, web containers, and so on, that they can use for testing without disturbing or being disturbed by anyone else. The easiest way to achieve this, in my experience, is to install the systems on each developer machine.

The test declares two exchanges, foo.domain and foo.domain.dlx, and three queues, foo.domain.queue1, foo.domain.queue2 and foo.comain.dlq. The queue foo.domain.queue1 is bound to exchange foo.domain with routing key rk1, and foo.domain.queue2 is bound to the same exchange with routing key rk2. The exchange foo.domain.dlx is set a dead letter exchange for both queues. We then put three messages, foo, bar and baz in the dead letter queue with different routing keys:

public class MoveMessagesIT {

    private static final String EXCHANGE = "foo.domain";
    private static final String EXCHANGE_DLX = "foo.domain.dlx";
    private static final String QUEUE1 = "foo.domain.queue1";
    private static final String QUEUE2 = "foo.domain.queue2";
    private static final String QUEUE_DLX = "foo.domain.dlq";
    private static final String[] TEST_MESSAGES = { "foo", "bar", "baz" };
    private static final String ROUTING_KEY1 = "rk1";
    private static final String ROUTING_KEY2 = "rk2";
    private static final String[] ROUTING_KEYS = { ROUTING_KEY1, ROUTING_KEY2, ROUTING_KEY1 };

    @Before
    public void init() throws IOException {
        Connection connection = connectionFactory().newConnection();
        Channel channel = connection.createChannel();
        // Cleanup from previous test
        channel.queueDelete(QUEUE_DLX);
        channel.queueDelete(QUEUE2);
        channel.queueDelete(QUEUE1);
        channel.exchangeDelete(EXCHANGE_DLX);
        channel.exchangeDelete(EXCHANGE);
        // EXCHANGE/QUEUEs
        channel.exchangeDeclare(EXCHANGE, "topic");
        Map<String, Object> queueArgs = new HashMap<>();
        queueArgs.put("x-message-ttl", 10 * 1000);
        queueArgs.put("x-dead-letter-exchange", EXCHANGE_DLX);
        channel.queueDeclare(QUEUE1, true, false, false, queueArgs);
        channel.queueBind(QUEUE1, EXCHANGE, ROUTING_KEY1);
        channel.queueDeclare(QUEUE2, true, false, false, queueArgs);
        channel.queueBind(QUEUE2, EXCHANGE, ROUTING_KEY2);
        // DLX/DLQ
        channel.exchangeDeclare(EXCHANGE_DLX, "topic");
        channel.queueDeclare(QUEUE_DLX, true, false, false, null);
        channel.queueBind(QUEUE_DLX, EXCHANGE_DLX, "#");
        // Send test messages to DLQ
        for (int i = 0; i < TEST_MESSAGES.length; i++) {
            channel.basicPublish(EXCHANGE_DLX, ROUTING_KEYS[i], null, TEST_MESSAGES[i].getBytes());
        }
        channel.close();
        connection.close();
    }

We also want to verify that the contents of the three queues after moving the messages are as expected, so we create a helper method, verifyMessages that reads messages from a queue, verifying that the message content and routing key are correct:

    private static void verifyMessages(String queue, String routingKey, String... messages) throws IOException {
        Connection connection = connectionFactory().newConnection();
        Channel channel = connection.createChannel();

        List<String> messagesRead = new ArrayList<>();
        while (true) {
            GetResponse response = channel.basicGet(queue, true);
            if (response == null) {
                break;
            }
            Envelope envelope = response.getEnvelope();
            assertThat(envelope.getRoutingKey(), is(routingKey));
            messagesRead.add(new String(response.getBody()));
        }
        channel.close();
        connection.close();
        assertThat(messagesRead, is(Arrays.asList(messages)));
    }

    private static ConnectionFactory connectionFactory() {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        return factory;

    }

We are now ready to add the test method, which is very simple: move all messages from foo.domain.dlq to exchange foo.domain and then verify that the contents of the queues are as expected:

    @Test
    public void moveAllMessagesToExchange() throws Exception {
        MoveMessages moveMessages = new MoveMessages("localhost", "guest", "guest", "/");
        moveMessages.moveAllMessagesToExchange(QUEUE_DLX, EXCHANGE);
        verifyMessages(QUEUE1, ROUTING_KEY1, "foo", "baz");
        verifyMessages(QUEUE2, ROUTING_KEY2, "bar");
        verifyMessages(QUEUE_DLX, null);
    }

Conclusion

We have seen an example of how to write an integration test for RabbitMQ. A few things to note:

• In this case, an integration test is exactly what is needed since we want to verify how an external system, RabbitMQ, behaves as an effect of running our code. No unit testing is necessary, and mocking the behavior of RabbitMQ in this case would only verify that our mock setup behaves the way that we believe RabbitMQ to behave.
• Having easy access to your own instance of the systems you integrate with, for example by having them locally installed, makes integration testing much simpler.
• You often learn useful things about the system you integrate with while you write the integration tests. In this case, the mechanics of getting a connection and channel, and for getting and publishing messages, were already in place before writing any production code.
• Sometimes the test code is larger and more complex than the resulting code under test. This is OK, but remember to write the test code as cleanly as possible and refactor when necessary.