Pricing Integration Adapter (2)

Pricing adapters integrate your bank's pricing systems with the Caplin Platform. It is the responsibility of the adapter to connect to the bank's pricing server, retrieve the requested data from the server, and convert it into the format expected by the Caplin Platform.

Objectives

In this tutorial you will make the following changes to the pricing adapter you created in the tutorial Pricing Integration Adapter 1:

  1. Handle request for a currency pair and return a specific price
  2. Handle requests for a container of currency pairs and return the subjects in the container
  3. Connect to a backend pricing server and return regular price updates to subscribed peers

Handling requests for individual currency pairs

In this section you will create code to respond to requests for individual currency pairs.

Take a look at the src directory in your project. Here you will find the following files:

  • A class with your application entry point (named <blade name>). When the adapter is run, this class creates a new DataSource object that connects to Liberator or Transformer (in our case this is Transformer) and it creates a new PricingProvider instance.
  • A PricingProvider class (named <blade name>PricingProvider). This class implements the DataProvider interface's onRequest and onDiscard methods, which are called when Liberator subscribes and unsubscribes to a subject respectively. The PricingProvider uses a Publisher to create new messages (e.g. records) and send these to the Platform.

Handling a subject request

The PricingProvider class implements the DataProvider interface to receive requests from Liberator. The method onRequest is called every time the DataProvider receives a subject request. Which requests are forwarded to your adapter is determined by the add-data-service block in the file blade/Liberator/etc/rttpd.conf.

Follow the steps below to make your adapter write to the console each time it receives a message from Liberator:

  1. Edit the onRequest and onDiscard methods so that they each log the subject name associated with the event:
    @Override
    public void onRequest(RequestEvent requestEvent)
    {
         System.out.println("Received request for " + requestEvent.getSubject());
    }
    
    @Override
    public void onDiscard(DiscardEvent discardEvent)
    {
         System.out.println("Received discard for " + discardEvent.getSubject());
    }
  2. Restart the adapter in your IDE.
  3. Open you adapter's console output. In Eclipse, click Window > Show View > Console.

Follow the steps below to send a request to your adapter from a web client:

  1. Open a web browser and navigate to your Liberator's homepage (http://localhost:18080).
  2. On the Liberator home page, click Diagnostics > Liberator Explorer. Login with the default credentials: username 'admin' and password 'admin'.
  3. In the Subject box at the top of Liberator Explorer, type /FX/GBPUSD and press Return.

After you submit the request for /FX/GBPUSD, the follow events occur:

  1. Liberator creates an object for /FX/GBPUSD and sends a request message to your adapter
  2. Your adapter receives the subscription request and writes the following line to its console window: Received request for /FX/GBPUSD
  3. Your adapter does not publish a response to Liberator. After 10 seconds, Liberator Explorer times out the request and writes the following entry in the Callbacks panel: In - ActionFailEvent [subject=/FX/GBPUSD]
  4. Liberator discards the object it created for /FX/GBPUSD and sends a discard message to your adapter, indicating that Liberator is no longer interested in the subject.
  5. Your adapter receives the discard message and writes the following line to its console window: Received discard for /FX/GBPUSD

Publishing a response to a subject request

In this section, you add new code to the onRequest method to publish a response to Liberator's request.

The Publisher class has two methods for publishing messages to DataSource peers:

  • publishInitialMessage: publishes a message to all peers that have not yet received an initial message. This method sets the image flag on the message, which indicates to the subscribing peer that the message contains a complete set of fields for the subject.
  • publishToSubscribedPeers: publishes a message to all peers that have received an initial message. This method does not set the image flag on the message. The message may contain a partial set of fields for the subject.

The first response to a peer is sent by publishInitialMessage; subsequent messages to the peer are sent by publishToSubscribedPeers. The distinction between the methods publishInitialMessage and publishToSubscribedPeers provides compatiblity with backend servers that send full and partial images of data. While the order in which methods are called must be observed, the requirement to send a full set of fields in the initial message is not applicable in all cases. For example, if an adapter retrieves backend data asynchronously, then only the callback method has access to backend data and the onRequest method can only send an empty initial message.

Follow the steps below to publish an initial message in response to a request:

  1. Edit the onRequest method so that it creates and publishes an empty record:
    @Override
    public void onRequest(RequestEvent requestEvent)
    {
        System.out.println("Received request for " + requestEvent.getSubject());
        RecordType1Message initialMessage =
            publisher.getMessageFactory().createRecordType1Message(requestEvent.getSubject());
        publisher.publishInitialMessage(initialMessage);
    }
  2. Restart your adapter

Using Liberator Explorer, submit a request for the subject /FX/GBPUSD (see Handling a subject request above). This time Liberator Explorer receives a response and the “Callbacks” panel indicates that the subscription was successful:

In - RecordType1DataEvent [subject=/FX/GBPUSD, fields=[Type=211,SID=PricingAdapterPricingSvc1], image=false, timeReceived=1397567812085]

Notice that even though you created an empty message in your adapter, Liberator has added two fields of its own:

  • Type=211: indicates that this is a record message.
  • SID=PricingAdapterPricingSvc1: the source ID of the message. Indicates which data service provided this record message.

These metadata fields are typically not useful for display in the client application, but can be useful for monitoring and log analysis.

Adding fields to responses

In this section, you add code to publish a second message that contains a full set of fields with hard-coded values. This second message is published using the Publisher.publishToSubscribedPeers method. This method does not set the 'image' flag on the message, so it can be used to send messages with a partial set of fields if required. For the sake of simplicity, the adapter in this tutorial does not track field values and always sends the same set of fields with each update.

Follow the steps below:

  1. Create a new record and set these fields before publishing the message:
    @Override
    public void onRequest(RequestEvent requestEvent)
    {    
        System.out.println("Received request for " + requestEvent.getSubject()); 
        RecordType1Message initialMessage =
            publisher.getMessageFactory().createRecordType1Message(requestEvent.getSubject());
        publisher.publishInitialMessage(initialMessage);
    
        RecordType1Message updateMessage =
            publisher.getMessageFactory().createRecordType1Message(requestEvent.getSubject());
        updateMessage.setField("InstrumentName", requestEvent.getSubject());
        updateMessage.setField("BidPrice", "1.23");
        updateMessage.setField("AskPrice", "2.34");
        updateMessage.setField("BidPriceId", "00011");
        updateMessage.setField("AskPriceId", "00022");
        updateMessage.setField("BidGFA", "500000");
        updateMessage.setField("AskGFA", "300000");
        Date date = new Date();
        String timestamp = String.valueOf(date.getTime());
        updateMessage.setField("Timestamp", timestamp);
        publisher.publishToSubscribedPeers(updateMessage);
    }
  2. Restart your adapter

  3. Make a request from Liberator Explorer for /FX/GBP. It fails! Notice that the adapter displays a warning in the Eclipse console:

    java.lang.IllegalArgumentException: Cannot set field: No entry in field name/number mappings for name InstrumentName

    The reason for this error is that field names in Type 1 messages are encoded as numbers rather than field names, and we have yet to map our fields to numbers.

  4. Open the file blade/blade_config/fields.conf. Add the configuration below to the bottom of this file. Note that the field IDs are arbitrary; they simply have to be unique integers. Also, be sure to edit the file in blade/blade_config/fields.conf and not the file global_config/fields.conf.
    add-field InstrumentName          -10003
    add-field BidPrice                -10004
    add-field AskPrice                -10005
    add-field BidPriceId              -10006
    add-field AskPriceId              -10007
    add-field BidGFA                  -10008
    add-field AskGFA                  -10009
    add-field Timestamp               -10010
    
  5. Re-export your adapter's configuration as a config-only blade and deploy the blade to your Deployment Framework (see Export your adapter's configuration as a config-only blade and Deploy the config-only blade to your DFW)

    Note: every time you change your adapter's configuration you must re-export your adapter's configuration and deploy it to your Deployment Framework.
  6. Restart the deployment framework and try requesting /FX/GBPUSD (or any other instrument with the /FX/ prefix) from the Liberator Explorer. You will now notice that a full record comes through with the hard-coded field values that we set:

Handling requests for containers

A container is a special type of subject that contains a list of other subjects. When a client subscribes to a container, the client is also automatically subscribed to all the subjects listed in the container.

In this section you will add code to the PricingProvider.onRequest method to handle requests for a container of major currency pairs: /CONTAINER/FX/MAJOR. On receiving a request for /CONTAINER/FX/MAJOR, your adapter should publish a ContainerMessage containing the subjects listed in the container. The ContainerMessage does not reach the client directly; Liberator intercepts it and automatically subscribes the client to each of the subjects listed in the ContainerMessage.

Handling the request

Follow the steps below:

  1. Declare a new publisher for your container as a field at the top of your class:
    private Publisher containerPublisher;

    You need to create a separate publisher for the /CONTAINER/FX/MAJOR container because the container has a different prefix to the existing publisher.

  2. Instantiate the container's publisher in the PricingProvider’s initialise method:
    this.containerPublisher = dataSource.createActivePublisher(new PrefixNamespace("/CONTAINER/FX"), this);
  3. In the onRequest method, check if the request is for /CONTAINER/FX/MAJOR. If so, create a container message and add five record subjects to it. Otherwise, fall back to your existing record message code.
    @Override
    public void onRequest(RequestEvent requestEvent)
    {
        System.out.println("Received request for " + requestEvent.getSubject());
        String subject = requestEvent.getSubject();
        if (subject.equals("/CONTAINER/FX/MAJOR")) {
            ContainerMessage containerMsg =
                containerPublisher.getMessageFactory().createContainerMessage(subject);
            containerMsg.addElement("/FX/GBPUSD");
            containerMsg.addElement("/FX/GBPEUR");
            containerMsg.addElement("/FX/GBPJPY");
            containerMsg.addElement("/FX/GBPAUD");
            containerMsg.addElement("/FX/GBPCHF");
            containerPublisher.publishInitialMessage( containerMsg );
        } else {
            RecordType1Message recordType1Message =
                publisher.getMessageFactory().createRecordType1Message(requestEvent.getSubject());
            publisher.publishInitialMessage(recordType1Message);
            ⋮
        }
    }

Configuring the data service

In the project file blade/Liberator/etc/rttpd.conf you will find configuration for Liberator that defines your adapter as a data service for all subjects prefixed with /FX/. You must add a similar piece of configuration to define your adapter as a data service for subjects prefixed with /CONTAINER/FX.

To update the Liberator and Transformer configuration, follow the steps below:

  1. Find the blade’s Liberator configuration in the file blade/Liberator/etc/rttpd.conf. Add a second include-pattern like this:
    add-data-service
        ⋮
        include-pattern     "^/CONTAINER/FX/"
        ⋮
    end-data-service
  2. Do the same for the Transformer configuration in the file blade/Transformer/etc/transformer.conf.
  3. Re-export your adapter's configuration as a config-only blade and deploy the blade to your Deployment Framework (see Export your adapter's configuration as a config-only blade and Deploy the config-only blade to your DFW)

    Note: every time you change your adapter's configuration you must re-export your adapter's configuration and deploy it to your Deployment Framework.
  4. In Liberator Explorer, subscribe to /CONTAINER/FX/MAJOR. You should see all the currencies you added to the container appearing, each having the same data:

Connecting to a pricing server

So far we have only replied to requests with hard-coded data. In production, your adapter will open a subscription to a backend pricing server for each requested currency pair. Upon receiving updates, you adapter will publish an update message to all subscribed peers.

In this part of the tutorial you will connect your adapter to a mock pricing server that represents a bank's pricing server. The mock server generates prices for a set of currency pairs and provides trading functionality for spot trades. It produces prices for any pair of the following currencies: EUR, USD, GBP, JPY, DKK, ZAR, CHF and AUD.

Running the mock pricing-server

Follow the steps below to start the mock pricing-server:

  1. Download the TrainingTradeProject zip file. Extract “TrainingTradeProject-<version>.zip” and copy the file “SimpleFXServer-<version>.jar” to a location on your hard drive.
  2. In the directory in which you unzipped the file, type the following command in your command line tool to start the server:

    java -jar SimpleFXServer-<version>.jar

    The following console output indicates that the server is running:

    <timestamp> [main] INFO  quickfix.SocketAcceptor  - SessionTimer started
    <timestamp> [main] INFO  quickfix.mina.NetworkingOptions  - Socket option: SocketTcpNoDelay=true
    <timestamp> [main] INFO  quickfix.mina.NetworkingOptions  - Socket option: SocketSynchronousWrites=false
    <timestamp> [main] INFO  quickfix.mina.NetworkingOptions  - Socket option: SocketSynchronousWriteTimeout=30000
    <timestamp> [main] INFO  quickfix.SocketAcceptor  - Listening for connections at 0.0.0.0/0.0.0.0:14045
Note: To stop the mock server, press Ctrl + C.

Connecting the adapter to the mock pricing-server

Each bank server generally provides a library that provides an API for subscribing to data updates and server status. The mock server is no different: it has a client library that your adapter can use to communicate with the mock server.

Follow the steps below:

  1. Copy the file “SimpleFxMessenger-<version>.jar” (contained in the zip file which you extracted in the previous step) to lib directory in your adapter project.

    Note: this is not the same jar that you used in the previous step. There are two jars with similar names: the server jar (which you have used already) and the messenger jar which is the library used for connecting to the server.
  2. Edit the build.gradle file in the root directory of your adapter project, and add the following line to the dependencies { } stanza:

    compile (name: 'SimpleFXMessenger', version: '0+') { transitive = false }
  3. Reload the Gradle configuration for your project in your IDE. In Eclipse, right-click your root project folder and click Gradle > Refresh Gradle Project
  4. Implement the FXPricingListener interface on your PricingAdapterPricingProvider class. This interface is provided by the mock server's client library.
    public class PricingAdapterPricingProvider implements DataProvider, FXPricingListener {    
  5. When you update your class to implement FXPricingListener, Eclipse will warn you that you have not implemented the interface's methods. Click on the red error icon to the left of the line of code above, and select "Add unimplemented methods". Eclipse will automatically add the four methods that you need to implement to the bottom of your file.
  6. Update the onConnect method to print a message to the console:
    @Override
    public void onConnect()
    {
        System.out.println("Connected to mock server");
    }
    
    @Override
    public void onDisconnect()
    {
        // TODO Auto-generated method stub
    }
    
    @Override
    public void onPriceUpdate(FXQuote arg0)
    {
        // TODO Auto-generated method stub
    }
    
    @Override
    public void onSubscriptionError(String arg0, String arg1)
    {
        // TODO Auto-generated method stub
    }
  7. Declare a new instance of FxPricingMessenger as a field at the top of your class. In the constructor, instantiate this object and call the connect() method to make a connection to the MockServer. This FXPricingMessenger object is used to communicate with the mock server.
    private Publisher publisher;
    private Publisher containerPublisher;
    private FXPricingMessenger messenger;
    
    
    public PricingDataProvider(DataSource dataSource) { 
        this.dataSource = dataSource; 
    } 
    
    public void initialise() {
    
        this.publisher =
            dataSource.createActivePublisher(new PrefixNamespace("/FX/"), this);
        this.containerPublisher =
            dataSource.createActivePublisher(new PrefixNamespace("/CONTAINER/FX"), this);
    
        this.messenger =
            new FXPricingMessenger(this, FXPricingMessenger.DEFAULT_SESSION_ID_1, FXPricingMessenger.DEFAULT_PORT_1);
        this.messenger.connect();
    }
  8. Restart your adapter. In the Eclipse console you should now see the line "Connected to mock server".

Improving the start-up behaviour of your adapter

When your adapter starts up, it automatically connects to Liberator. This can create an undesirable situation where the adapter connects successfully to Liberator but the underlying bank server is down. This is problematic because Liberator will send requests to the adapter that the adapter cannot service.

It is considered best practice for the status of the adapter to reflect the status of the underlying bank server. We are now going to improve the adapter so that it only initialises the connection to Liberator once it has successfully connected to the bank server.

Follow the steps below to postpone calling dataSource.start() until the adapter has successfully achieved a connection with the mock server:

  1. Remove the dataSource.start() instruction from the main method of the PricingAdapter class.
  2. In the constructor of the PricingAdapterPricingProvider class, store the DataSource object as a field so that it can be accessed from any method in this class. Also add a boolean field that we will use to keep track of whether or not the DataSource has been started. This boolean field will be initialised to false.
    private final Publisher publisher; 
    private final Publisher containerPublisher; 
    private final FXPricingMessenger messenger;
    private final DataSource dataSource;
    
    private boolean dataSourceIsStarted = false;
    
    public PricingAdapterPricingProvider(DataSource dataSource) 
    { 
        this.publisher = dataSource.createActivePublisher(new PrefixNamespace("/FX/"), this);
        this.containerPublisher = 
            dataSource.createActivePublisher(new PrefixNamespace("/CONTAINER/FX"), this);
        this.dataSource = dataSource;
    
        this.messenger = new FXPricingMessenger(this, FXPricingMessenger.DEFAULT_SESSION_ID_1, FXPricingMessenger.DEFAULT_PORT_1);
        this.messenger.connect(); 
    }
  3. In the onConnect method, which is called when the adapter successfully connects to the mock server, start the DataSource if it has not yet been started or toggle its status to UP if it has been started before. You will use your boolean field to know if the DataSource has been started or not.
    @Override
    public void onConnect()
    {
        System.out.println("Connected to mock server");
        if (dataSourceIsStarted) {  
            dataSource.setStatusUp();
        } else {
            dataSource.start();
            dataSourceIsStarted = true;
        }
    }
  4. In the onDisconnect method, which is called when the adapter loses connection to the mock server, toggle the DataSource status to DOWN.
    @Override
    public void onDisconnect()
    {
        System.out.println( "Disconnected from mock server" );
        dataSource.setStatusDown();
    }
  5. At this point you should be able to start your adapter and see it turn blue in the CMC, indicating that it has successfuly connected to Liberator.

  6. Try stopping the mock server by pressing Ctrl+C in the command window. You should see the Pricing Adapter turn red in the CMC, indicating that it is unavailable. This is because when you stopped the mock server the onDisconnect() method would have been called, resulting in the adapter toggling its status to DOWN.

  7. Restart the mock server. The Pricing Adapter should turn blue in the CMC, indicating that it is available again. This is because the onConnect() method would have been called when connection to the mock server was re-established, resulting in the adapter toggling its status to UP.

Requesting data from the mock server

Next, we’ll change the onRequest method so that when a request is made for an FX subject, the adapter requests data from the mock server. (This is instead of returning a record with hard-coded data as we did previously.)

Subject requests are received from the platform in the form “/FX/<currency pair>” e.g. “/FX/GBPUSD”. This particular mock server, however, expects requests in the form “<currency>/<currency>” e.g. “GBP/USD”. When a request comes in you will need to convert the requested subject to a currency pair that matches the expected format.

Follow the steps below:

  1. Create a new private method that converts a Caplin Platform subject into the format expected by the mock server:

    private String convertSubjectToInstrumentName(String subject)
    {
        String[] splittedSubject = subject.split("/");
        String currencyPair = splittedSubject[2];
        return currencyPair.substring(0, 3) + "/" + currencyPair.substring(3, 6);
    }
  2. In the onRequest method, convert the requested subject into the format expected by the mock server and subscribe to the subject on the mock server.
    @Override
    public void onRequest(RequestEvent requestEvent)
    {
        String subject = requestEvent.getSubject();
        if (subject.equals("/CONTAINER/FX/MAJOR")) {
            ⋮
        }
        else {
            String instrumentName = convertSubjectToInstrumentName(subject);
            messenger.subscribe(instrumentName);
    
            RecordType1Message initialMessage = publisher.getMessageFactory().createRecordType1Message(requestEvent.getSubject()); 
            publisher.publishInitialMessage(initialMessage); 
            ⋮
        }
    }
    
  3. Re-run your adapter and request a valid currency pair using Liberator Explorer. The result should still be the hard-coded record, but now, in the console window for the mock server, you will notice some messages appearing, every 5 seconds. These correspond to the new price updates being sent to the adapter as a response.

Propagating pricing updates to subscribing DataSource peers

The final step is to receive the data updates coming from the mock server, create a record, and propagate the update to Liberator. To receive incoming messages from the Mock Server you must implement the onPriceUpdate callback method (a method defined in the FXPricingListener interface).

Follow the steps below:

  1. Create a new private method that converts currency pairs to the format expected by the Caplin Platform (e.g. convert “GBP/USD” to “/FX/GBPUSD”).
    private String convertInstrumentNameToSubject(String instrumentName)
    {
        String subject = "/FX" + "/" + instrumentName.substring(0, 3) + instrumentName.substring(4, 7);
        return subject;
    }    
  2. In the onPriceUpdate() method, create a new record message with a subject name converted using the convertInstrumentNameToSubject method. Populate the fields on the message with the data in the FXQuote, using the same field names as in the hard-coded initial response. Then publish the message to the subscribed peers.

    @Override
    public void onPriceUpdate(FXQuote fxQuote) 
    {
        String instrumentName = fxQuote.getCurrencyPair();
        String subject = convertInstrumentNameToSubject(instrumentName);
    
        RecordType1Message updateMessage =
            publisher.getMessageFactory().createRecordType1Message(subject);
        updateMessage.setField("InstrumentName", subject);
        updateMessage.setField("BidPrice", String.valueOf(fxQuote.getBidPrice()));
        updateMessage.setField("AskPrice", String.valueOf(fxQuote.getOfferPrice()));
        updateMessage.setField("BidPriceId", String.valueOf(fxQuote.getBidID()));
        updateMessage.setField("AskPriceId", String.valueOf(fxQuote.getOfferID()));
        updateMessage.setField("BidGFA", String.valueOf(fxQuote.getBidSize()));
        updateMessage.setField("AskGFA", String.valueOf(fxQuote.getOfferSize()));
        Date date = new Date();
        String timestamp = String.valueOf(date.getTime());
        updateMessage.setField("Timestamp", timestamp);
    
        publisher.publishToSubscribedPeers(updateMessage);
    }
    
  3. In the onRequest method, remove the hard-coded data update after the initial record request so that you go back to just sending a blank initial response when a request is received:
    @Override 
    public void onRequest(RequestEvent requestEvent) 
    { 
        System.out.println("Received request for " + requestEvent.getSubject()); 
        String subject = requestEvent.getSubject(); 
        if (subject.equals("/CONTAINER/FX/MAJOR")) { 
            ContainerMessage containerMsg =
                containerPublisher.getMessageFactory().createContainerMessage(subject); 
            containerMsg.addElement("/FX/GBPUSD"); 
            containerMsg.addElement("/FX/GBPEUR"); 
            containerMsg.addElement("/FX/GBPJPY"); 
            containerMsg.addElement("/FX/GBPAUD"); 
            containerMsg.addElement("/FX/GBPCHF"); 
            containerPublisher.publishInitialMessage( containerMsg ); } 
        else {
            String instrumentName = convertSubjectToInstrumentName(subject);
            messenger.subscribe(instrumentName);
     
            RecordType1Message recordType1Message =
                publisher.getMessageFactory().createRecordType1Message(requestEvent.getSubject()); 
            publisher.publishInitialMessage(recordType1Message);
        } 
    }   
  4. Restart your adapter.

  5. Open Liberator Explorer and, once again, subscribe to the /FX/GBPUSD object and the /CONTAINER/FX/MAJOR container. You should now see data updates appearing with realistic ticking updates from the mock server.

Discarding subscriptions when the client disconnects

One minute after closing the Liberator Explorer window/tab in your browser, Liberator will drop the subscription and send a discard message to your adapter. This is your cue to end the subscription with the mock server.

Follow the steps below:

  1. Convert the discarded subject to the instrument required by the mock server (e.g. “/FX/GBPUSD” to “GBP/USD”), then use the messenger to unsubscribe from the instrument.

    @Override
    public void onDiscard(DiscardEvent discardEvent)
    {
        System.out.println("Received discard for " + discardEvent.getSubject());
        String subject = discardEvent.getSubject();
        String instrumentName = convertSubjectToInstrumentName(subject);
        messenger.unsubscribe(instrumentName);
    }
  2. Make a subscription using Liberator Explorer then close the Liberator Explorer browser window. Liberator will send a discard message one minute after you closed the Explorer window. When this happens and the adapter unsubscribes from the mock server, the mock server’s output log will display:

    pricing.IntervalTaskUpdate  - CANCELING interval update

Review

In this tutorial we looked at the implementation of new Pricing Integration Adapters – how to receive requests from the Platform and respond by sending an initial response. We also looked at how to send price updates to the Platform. We used a mock server to retrieve the price updates. While the way you communicate with backend servers will differ from one backend server to another, the way you deal with object and container requests, create new records, and send initial and update records does not change.