Subscription

Data can be subscribed to from the platform by registering a MessageProcessor for a data type, then calling subscribe(...) with the topics of interest.

Message Routing

Applications register MessageProcessor callbacks for data they wish to process. These must be registered before starting a session.

Messages are delivered to a MessageProcessor in the form of a KeyMessage<T>, where <T> is the type of the data being requested. The KeyMessage also provides metadata, see Message Structure.

Message Type	Register With
SBE	session.routing().sbe(...)
DTO	session.routing().dto(...)
Flexible	session.routing().flexible(...)
Custom	session.routing().custom(...)
Raw	session.routing().raw(...)

Subscribing to Data

Subscriptions can be requested before or after the session has started.

Subscribe Syntax

session.subscribe(<subscription-type>, <subscription-listener>, <topics>);

Method Argument	Description
Subscription Type	One of LIVE, CACHED or CACHED_AND_LIVE
Subscription Listener	Callback which receives subscription success or failure
Topics	One or more topics for which data is being requested. See Topics.

Details of Subscription Types:

• LIVE - live data, which is data being published on the platform at the present time
• CACHED - a snapshot of data that has been stored in the cache
• CACHED_AND_LIVE - cached and live data. The API automatically prioritises live in-flight data such that it takes precedence over an older cached image of that data, thereby ensuring that stale cached images of data are never delivered to an application.

A SubscriptionListener is called when the subscription completes or fails.

SubscriptionListener listener = new SubscriptionListener() {
  @Override
  public void onSuccess(long count) {
    log.info("Subscription complete, received {}", count);
  }

  @Override
  public void onFailure(String message) {
    log.warn("Subscription failed: {}", message);
  }
};

The count value is the number of items received when a cache interaction took place.

Subscription listeners are called on the session thread. The same listener instance can be reused across multiple subscribe(...) calls when the application wants the same completion behaviour.

Examples:

Subscribe to live data for a topic

session.subscribe(SubscriptionType.LIVE, listener, topic);

Subscribe to both cached and live data for topic1 and topic2

session.subscribe(SubscriptionType.CACHED_AND_LIVE, listener, topic1, topic2);

info

Routing and subscription are intentionally separate concepts.

The session brings data in through a single high-performance inbound pipe. Subscriptions decide what data should be received down that pipe; routing is the local dispatch table that decides which callback receives a message when it arrives.

For low-latency applications, this also means messages can be read directly from transport-backed buffers during the callback. In the case of SBE, the application receives a generated decoder view over the incoming bytes, so hot-path code can read fields without first copying the payload into an intermediate object.

This keeps the threading model simple. Message callbacks run on the session thread, without creating one thread per subscription or handing every message through extra queues. Applications get predictable callback ordering and a fast receive path.

SBE Subscription

session.routing().sbe(PriceDecoder.class, this::processPrice, Optional.empty());

Topic topic = Topics.create(PriceDecoder.class, Topic.SOURCE_WILDCARD, "price", "UST-10Y");
session.subscribe(SubscriptionType.LIVE, listener, topic);

private void processPrice(KeyMessage<PriceDecoder> message) {
  PriceDecoder price = message.payload();
  log.info("Received Price SBE: {}", price);
}

DTO Subscription

session.routing().dto(PriceDto.class, this::processPrice, Optional.empty());

Topic topic = Topics.create(PriceDto.class, Topic.SOURCE_WILDCARD, "price", "UST-10Y");
session.subscribe(SubscriptionType.LIVE, listener, topic);

private void processPrice(KeyMessage<PriceDto> message) {
  PriceDto price = message.payload();
  log.info("Received Price DTO: {}", price);
}

Flexible Subscription

session.routing().flexible(Map.class, this::processMap, Optional.empty());

Topic topic = Topics.create(Topic.SOURCE_WILDCARD, "price", "UST-10Y");
session.subscribe(SubscriptionType.LIVE, listener, topic);

private void processMap(KeyMessage<Map<String, Object>> message) {
  Map<String, Object> map = message.payload();
  log.info("Received Map: {}", map);
}

Custom Subscription

int schemaId = 1_000;
int templateId = 1;

session.routing().custom(OrderUpdate.class, this::processOrderUpdate, Optional.empty());

Topic topic = Topics.create(schemaId, templateId, Topic.SOURCE_WILDCARD, "orders", "order-1");
session.subscribe(SubscriptionType.LIVE, listener, topic);

private void processOrderUpdate(KeyMessage<OrderUpdate> message) {
  OrderUpdate update = message.payload();
  log.info("Received order update: {}", update);
}

For Custom messages, the schemaId and templateId must match the values provided by the configured Custom Codec.

Raw Subscription

int schemaId = 100;
int templateId = 1;

session.routing().raw(schemaId, templateId, this::processRaw, Optional.empty());

Topic topic = Topics.create(schemaId, templateId, Topic.SOURCE_WILDCARD, "raw", "item-1");
session.subscribe(SubscriptionType.LIVE, listener, topic);

private void processRaw(KeyMessage<RawPayload> message) {
  RawPayload payload = message.payload();
  log.info("Received raw payload with {} bytes", payload.length());
}

Message Lifetime

The MessageProcessor callback receives a message. Some message types are backed by a transport buffer, so the message is only valid while the callback is running.

Use the message in the callback. DTO and flexible messages are fully safe and can be kept directly. For SBE, Custom and Raw messages, copy the message before the callback returns if it needs to be retained, queued, edited or published later.

Message Type	Safe to Keep Message	Notes
SBE	❌	Message is a view over the transport buffer.
DTO	✔️	Fully safe to retain.
Flexible	✔️	Fully safe to retain.
Custom	❌	Message is a view over the transport buffer.
Raw	❌	Message is a view over the transport buffer.

Only the message types that are not safe to keep need to be copied:

Copy an SBE message

KeyMessage<PriceDecoder> message = ...;

SbeMessage<PriceEncoder, PriceDecoder> copy = Messages.sbe(PriceEncoder.class);
copy.copyFrom(message);

Copy a Custom message

KeyMessage<SampleCustomPayload> message = ...;

CustomMessage<SampleCustomPayload> copy = Messages.custom(SampleCustomPayload.class);
// note that this only performs a shallow copy, the message payload itself must be manually copied
copy.copyFrom(message);

Copy a raw message

KeyMessage<RawPayload> message = ...;

RawMessage copy = Messages.raw();
copy.copyFrom(message);

Unsubscribing to Data

Use unsubscribe(...) to stop receiving data for one or more topics.

Unsubscribe Syntax

session.unsubscribe(<unsubscription-listener>, <topics>);

Method Argument	Description
Unsubscription Listener	Callback which receives unsubscription success or failure
Topics	One or more topics to remove from the session's active subscriptions

The topics should match the topics used when subscribing. Unsubscription can only be used for for LIVE and CACHED_AND_LIVE subscriptions, where the session has an ongoing live subscription to remove.

Provide an UnsubscriptionListener to handle completion or failure:

UnsubscriptionListener listener = new UnsubscriptionListener() {
  @Override
  public void onSuccess() {
    log.info("Unsubscription complete");
  }

  @Override
  public void onFailure(String message) {
    log.warn("Unsubscription failed: {}", message);
  }
};

session.unsubscribe(listener, topic);

Unsubscription listeners are called on the session thread. The same listener instance can be reused across multiple unsubscribe(...) calls when the application wants the same completion behaviour.

Voided Messages

Every routing method accepts a MessageProcessor and an optional voided MessageProcessor:

session.routing().dto(PriceDto.class, this::processPrice,
  Optional.of(this::processVoidedPrice));

The second argument receives normal messages. The third argument receives messages where message.voided() is true.

Voided messages are used when a publisher invalidates a previously published message. Pass Optional.empty() when the application does not need special handling, the session will ignore voided messages for that route.