Publishing a Binary subject from a Java Caplin DataSource

The page provides an overview of how to publish a subject of type Binary from a Java DataSource.

The instructions on this page assume you already know how to create a Java DataSource project. For more information on creating a Java DataSource, see the Caplin Platform Developer Tutorial 1.

Contents:

Requirements
Overview
Example

Requirements

To publish binary messages, you require:

A Caplin DataSource for Java 8.0.11+
Caplin Liberator 8.0.15+ (with Caplin StreamLink 8.0.7+)
[Optional] Caplin Transformer 8.0.8+

Overview

Binary messages are published via a CachingPublisher, which caches the last published value for each subject it serves. The CachingPublisher uses its cache to serve new requests for subjects it has already served, and, in the case of Binary data, to determine if it is more efficient to publish an update as a Binary patch or in full.

You provide data for Binary subjects by writing a CachingDataProvider, which you register with a DataSource instance using DataSource.createCachingPublisher(Namespace, CachingDataProvider). This call returns a CachingPublisher, which you must make available to your CachingDataProvider in order for it to publish BinaryMessage objects.

When a DataSource receives a Binary subject request, if the CachingPublisher for the subject namespace does not have a cached object for the subject, then the CachingPublisher routes the request to its associated CachingDataProvider. The CachingDataProvider subscribes to data in the backend and publishes a BinaryMessage.

Serving a request for an uncached object

When a DataSource receives a Binary subject request, if the CachingPublisher for the subject namespace has a cached object for the subject, then the CachingPublisher serves the request directly from its cache:

Serving a request for a cached object

The CachingDataProvider implementation should also publish updates resulting from its subscription to backend data events. When the CachingDataProvider publishes an update to a subject that the CachingPublisher has already cached, the CachingPublisher chooses the most efficient format in which to publish the update: as a Binary patch or as the full image.

Serving updates for a cached object

Example

In this example, we’ll create a CachingDataProvider that generates random pricing data for currency pair subjects (/CCYS/<currency_pair>).

ExampleBinaryAdapter.java

import com.caplin.datasource.DataSource;
import com.caplin.datasource.namespace.PrefixNamespace;

public class ExampleBinaryAdapter
{
    public static void main(final String[] args)
    {
        final DataSource dataSource = DataSource.fromArgs(args);

        PricingCachingDataProvider pricingCachingDataProvider =
                new PricingCachingDataProvider(); (1)

        dataSource.createCachingPublisher(
                new PrefixNamespace("/CCYS/"), pricingCachingDataProvider); (2)

        dataSource.start();
    }
}

1	Instantiate a `CachingDataProvider`. In this example we instantiate a `PricingCachingDataProvider` (see source code below).
2	Register the `PricingCachingDataProvider` with the adapter’s `DataSource` instance to create a `CachingPublisher`. The `DataSource` injects the `CachingPublisher` into the provider via its `setPublisher` method.

PricingCachingDataProvider.java

The CachingDataProvider that responds to subject requests and discards.

import com.caplin.datasource.messaging.binary.BinaryMessage;
import com.caplin.datasource.publisher.CachingDataProvider;
import com.caplin.datasource.publisher.CachingPublisher;

import java.util.Map;
import java.util.Random;
import java.util.concurrent.*;

public class PricingCachingDataProvider implements CachingDataProvider
{
    private final ScheduledExecutorService executorService =
            Executors.newSingleThreadScheduledExecutor();
    private final Random random = new Random();
    private final Map<String, ScheduledFuture<?>> subscriptions =
            new ConcurrentHashMap<>();
    private CachingPublisher cachingPublisher = null;

    @Override
    public void onRequest(String subject) { (1)
        subscriptions.put(
            subject,
            executorService.scheduleAtFixedRate(() -> {
                Price randomPrice = createRandomPrice(subject.split("/")[2]);
                BinaryMessage binaryMessage = cachingPublisher.getCachedMessageFactory()
                        .createBinaryMessage(subject, randomPrice.toBinary());
                cachingPublisher.publish(binaryMessage);
            }, 0L, 1L, TimeUnit.SECONDS)
        );
    }

    @Override
    public void onDiscard(String subject) { (2)
        ScheduledFuture<?> scheduledFuture = subscriptions.remove(subject);
        if (scheduledFuture != null) scheduledFuture.cancel(true);
    }

    @Override
    public void setPublisher(CachingPublisher cachingPublisher) { (3)
        this.cachingPublisher = cachingPublisher;
    }

    private Price createRandomPrice(String currencyPair) {
        return new Price(
            String.valueOf(random.nextInt()),
            String.valueOf(random.nextDouble()),
            String.valueOf(random.nextDouble()),
            currencyPair
        );
    }
}

1	On subject request, parses the subject and subscribe to a backend process to provide updates. This method is called by the `CachingPublisher` when it does not have cached data for a subject; thereafter, requests for the same subject are served from the `CachingPublisher` cache. In this example, we start a simulation that publishes a random price for the subject every second.
2	On subject discard, cancel the backend process that generates updates for the subject. In this example, we cancel the `ScheduledFuture` that generates a random price every second.
3	The `setPublisher` method is defined by the `CachingDataProvider` interface as a default no-op. We override it so the `DataSource` can give the provider the `CachingPublisher` it uses to publish.

Price.java

A class that represents the data to send as a Binary message.

import java.io.*;

public class Price {
    private final String id;
    private final String bid;
    private final String ask;
    private final String currencyPair;

    public Price(String id, String bid, String ask, String currencyPair) {
        this.id = id;
        this.bid = bid;
        this.ask = ask;
        this.currencyPair = currencyPair;
    }

    public String getId() {
        return id;
    }

    public String getBid() {
        return bid;
    }

    public String getAsk() {
        return ask;
    }

    public String getCurrencyPair() {
        return currencyPair;
    }

    public byte[] toBinary() {
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        DataOutputStream out = new DataOutputStream(bos);

        try {
            out.writeUTF(id);
            out.writeUTF(bid);
            out.writeUTF(ask);
            out.writeUTF(currencyPair);

            out.flush();
        } catch (IOException ignored) {}

        return bos.toByteArray();
    }
}

See also: