Publishing a Binary subject from a .NET Caplin DataSource
The page provides an overview of how to publish a subject of type Binary from a .NET DataSource.
Publishing a Binary subject from a .NET DataSource
The instructions on this page assume you already know how to create a .NET DataSource project. For more information on creating a .NET DataSource, see the examples in the .NET DSDK kit.
Requirements
To publish binary messages, you require:
-
A Caplin DataSource .NET 8.0.12+
-
Caplin Liberator 8.0.15+ (with Caplin StreamLink 8.0.7+)
-
[Optional] Caplin Transformer 8.0.8+
Overview
Binary messages are published via a ICachingPublisher, which caches the last published value for each subject it serves. The ICachingPublisher 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 implementing the ICachingDataProvider interface, which you then register with a DataSource instance using DataSource.CreateCachingPublisher(INamespace, ICachingDataProvider). This call returns a ICachingPublisher, which you must make available to your ICachingDataProvider in order for it to publish IBinaryMessage objects.
When a DataSource receives a Binary subject request, if the ICachingPublisher for the subject namespace does not have a cached object for the subject, then the ICachingPublisher routes the request to its associated ICachingDataProvider. The ICachingDataProvider subscribes to data in the backend and publishes a IBinaryMessage.
When a DataSource receives a Binary subject request, if the ICachingPublisher for the subject namespace has a cached object for the subject, then the ICachingPublisher serves the request directly from its cache:
The ICachingDataProvider implementation should also publish updates resulting from its subscription to backend data events. When the ICachingDataProvider publishes an update to a subject that the ICachingPublisher has already cached, the ICachingPublisher chooses the most efficient format in which to publish the update: as a Binary patch or as the full image.
Example
In this example, we’ll create a ICachingDataProvider that handles binary data.
ExampleBinaryAdapter.cs
using Caplin.DataSource;
using Caplin.DataSource.Namespace;
using Caplin.DataSource.Publisher;
using Caplin.Logging;
namespace ExampleBinaryAdapter.NET
{
public class ExampleBinaryAdapter
{
private IDataSource dataSource;
public ExampleBinaryAdapter(string[] args)
{
ILogger logger = new ConsoleLogger();
dataSource = new DataSource("demosource.conf", args, logger);
ICachingDataProvider provider = new PricingCachingDataProvider(); (1)
dataSource.CreateCachingPublisher(new PrefixNamespace("/FX"), provider);(2)
}
public void Run()
{
dataSource.Start();
}
public static void Main(string[] args) (3)
{
new ExampleBinaryAdapter(args).Run();
}
}
}| 1 | Instantiate a ICachingDataProvider. In this example we instantiate a PricingCachingDataProvider (see source code below). |
| 2 | Register the PricingCachingDataProvider with the DataSource to create a ICachingPublisher. The DataSource injects the ICachingPublisher into the provider via its SetPublisher method. |
| 3 | The program entry point: create the adapter and start it. |
PricingCachingDataProvider.cs
The ICachingDataProvider that responds to subject requests and discards.
using Caplin.DataSource.Messaging.Binary;
using Caplin.DataSource.Publisher;
using System;
using System.Collections.Concurrent;
using System.Threading;
public class PricingCachingDataProvider : ICachingDataProvider
{
private readonly Random random = new Random();
private readonly ConcurrentDictionary<string, Timer> subscriptions =
new ConcurrentDictionary<string, Timer>();
private ICachingPublisher publisher;
public void ReceiveRequest(string subject) (1)
{
// Subscribe to the back end system. In this example we simulate it by
// publishing a new random price for the subject every second.
Timer timer = new Timer(_ =>
{
Price price = CreateRandomPrice(subject.Split('/')[2]);
IBinaryMessage message = publisher.CachedMessageFactory.CreateBinaryMessage(subject, price.ToBinary());
publisher.Publish(message);
}, null, TimeSpan.Zero, TimeSpan.FromSeconds(1));
subscriptions[subject] = timer;
}
public void ReceiveDiscard(string subject) (2)
{
// Unsubscribe from the back end system.
if (subscriptions.TryRemove(subject, out Timer timer))
{
timer.Dispose();
}
}
public void SetPublisher(ICachingPublisher publisher) (3)
{
this.publisher = publisher;
}
private Price CreateRandomPrice(string currencyPair)
{
return new Price(
random.Next().ToString(),
random.NextDouble().ToString(),
random.NextDouble().ToString(),
currencyPair);
}
}| 1 | On subject request, subscribe to a backend process to provide updates. This method is called by the ICachingPublisher when it does not have cached data for a subject; thereafter, requests for the same subject are served from the ICachingPublisher 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 timer that publishes a random price every second. |
| 3 | The SetPublisher method is part of the ICachingDataProvider interface; the DataSource calls it to give the provider the ICachingPublisher it uses to publish. |
Price.cs
A class that represents the data to send as a Binary message.
using System;
using System.IO;
using System.Text;
public class Price
{
private readonly string id;
private readonly string bid;
private readonly string ask;
private readonly 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 Id => id;
public string Bid => bid;
public string Ask => ask;
public string CurrencyPair => currencyPair;
public byte[] ToBinary()
{
using (var ms = new MemoryStream())
using (var writer = new BinaryWriter(ms, Encoding.UTF8, true))
{
try
{
writer.Write(id);
writer.Write(bid);
writer.Write(ask);
writer.Write(currencyPair);
writer.Flush();
}
catch (IOException)
{
// Ignored
}
return ms.ToArray();
}
}
}See also: