# Publishing a Binary subject from a Java Caplin DataSource The page provides an overview of how to publish a subject of type [Binary](datasource-binary.md) 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](../../tutorials/caplin-platform-integration-suite/index.md). ## 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`](https://docs.caplin.com/developer/api/datasource_java/latest/com/caplin/datasource/publisher/CachingPublisher.html), 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](https://www.rfc-editor.org/rfc/rfc3284.txt) or in full. You provide data for Binary subjects by writing a [`CachingDataProvider`](https://docs.caplin.com/developer/api/datasource_java/latest/com/caplin/datasource/publisher/CachingDataProvider.html), which you register with a [`DataSource`](https://docs.caplin.com/developer/api/datasource_java/latest/com/caplin/datasource/DataSource.html) instance using `DataSource.createCachingPublisher(__Namespace__, __CachingDataProvider__)`. This call returns a [`CachingPublisher`](https://docs.caplin.com/developer/api/datasource_java/latest/com/caplin/datasource/publisher/CachingPublisher.html), which you must make available to your `CachingDataProvider` in order for it to publish [`BinaryMessage`](https://docs.caplin.com/developer/api/datasource_java/latest/com/caplin/datasource/messaging/binary/BinaryMessage.html) 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`](https://docs.caplin.com/developer/api/datasource_java/latest/com/caplin/datasource/messaging/binary/BinaryMessage.html). **Serving a request for an uncached object** ```plantuml participant StreamLink participant Liberator box "Java DataSource" participant DataSource participant "CachingPublisher (/FX/...)" as CachingPublisher participant CachingDataProvider end box participant "Backend System" as BackendSystem StreamLink -> Liberator : /FX/GBPUSD Liberator -> DataSource : /FX/GBPUSD DataSource -> CachingPublisher : /FX/GBPUSD note over CachingPublisher Object not in cache end note CachingPublisher -> CachingDataProvider : CachingDataProvider.onRequest(//subject//) CachingDataProvider -> BackendSystem : Subscribe to GBPUSD CachingDataProvider <- BackendSystem CachingPublisher <- CachingDataProvider : CachingPublisher.publish(//binaryMessage//) CachingPublisher -> CachingPublisher : Cache Binary data DataSource <- CachingPublisher Liberator <- DataSource StreamLink <- Liberator ``` 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** ```plantuml participant StreamLink participant Liberator box "Java DataSource" participant DataSource participant "CachingPublisher (/FX/...)" as CachingPublisher participant CachingDataProvider end box participant "Backend System" as BackendSystem StreamLink -> Liberator : /FX/GBPUSD Liberator -> DataSource : /FX/GBPUSD DataSource -> CachingPublisher : /FX/GBPUSD note over CachingPublisher Object in cache end note DataSource <- CachingPublisher Liberator <- DataSource StreamLink <- Liberator ``` 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](https://www.rfc-editor.org/rfc/rfc3284.txt) or as the full image. **Serving updates for a cached object** ```plantuml collections "Subscribed Peers" as Peers box "Java DataSource" participant DataSource participant "CachingPublisher (/FX/...)" as CachingPublisher participant CachingDataProvider end box participant "Backend System" as BackendSystem BackendSystem -> CachingDataProvider : Update for GBPUSD CachingDataProvider -> CachingPublisher : CachingPublisher.publish(//binaryMessage//) note over CachingPublisher Compare to cached object. Publish as a Binary patch or publish in full? end note CachingPublisher -> CachingPublisher : Update cache CachingPublisher -> DataSource DataSource -> Peers ``` ## Example In this example, we’ll create a [`CachingDataProvider`](https://docs.caplin.com/developer/api/datasource_java/latest/com/caplin/datasource/publisher/CachingDataProvider.html) that generates random pricing data for currency pair subjects (`/CCYS/`). **ExampleBinaryAdapter.java** ```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`](https://docs.caplin.com/developer/api/datasource_java/latest/com/caplin/datasource/publisher/CachingDataProvider.html). In this example we instantiate a `PricingCachingDataProvider` (see source code below). 2. Register the `PricingCachingDataProvider` with the adapter’s [`DataSource`](https://docs.caplin.com/developer/api/datasource_java/latest/com/caplin/datasource/DataSource.html) instance to create a [`CachingPublisher`](https://docs.caplin.com/developer/api/datasource_java/latest/com/caplin/datasource/publisher/CachingPublisher.html). The `DataSource` injects the `CachingPublisher` into the provider via its `setPublisher` method. **PricingCachingDataProvider.java** The [`CachingDataProvider`](https://docs.caplin.com/developer/api/datasource_java/latest/com/caplin/datasource/publisher/CachingDataProvider.html) that responds to subject requests and discards. ```java 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> 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. ```java 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**: * [Binary data type](datasource-binary.md) * [Subscribing to a Binary subject](../streamlink/streamlink-subscribing-to-binary.md)