Real-time data

Subscriptions, ticks, and current-value boards.

A Subscription is an AutoCloseable iterable of sparse Ticks, an Arrow batch stream through arrow(), a gap-aware delivery stream through deliveries(), and a polled board through latest().

WARNING

Live subscriptions need real-time entitlements for every security and field. Without them Bloomberg may send delayed data or reject fields; xbbg reports both through XbbgWarnings and diagnostic snapshots rather than hiding them.

Open a subscription

Every subscription entry point is available on both the static Blp facade (using the shared engine configured with Xbbg.configure(...)) and on an explicit Engine:

Method	Bloomberg service	Notes
subscribe(tickers, fields[, StreamOptions])	//blp/mktdata	Core market-data subscription (Blp.subscribe accepts a single String or Collection<String> for tickers and fields; Engine#subscribe takes Collection<String>).
stream(tickers, fields[, StreamOptions])	//blp/mktdata	Alias of subscribe with Collection<String> arguments.
vwap(tickers, fields[, StreamOptions])	//blp/mktvwap	Streaming VWAP; pass service parameters in StreamOptions.Builder#options.
mktbar(ticker[, StreamOptions])	//blp/mktbar	Streaming intraday bars; fields come from StreamOptions.Builder#fields.
depth(ticker[, StreamOptions])	//blp/mktdepthdata	Market depth (B-PIPE); fields come from StreamOptions.Builder#fields.
chains(ticker[, StreamOptions])	//blp/mktlist	Option and futures chains (B-PIPE); fields come from StreamOptions.Builder#fields.
subscribeWithOptions(service, tickers, fields, StreamOptions)	Any service	Passes StreamOptions#options() to Bloomberg unchanged.
subscribeAuction(securities[, AuctionStreamOptions])	//blp/mktdata	Resolves primary auction venues first, then opens one venue subscription; see Recipes.
streamAuction(securities[, AuctionStreamOptions])	//blp/mktdata	Row-delivering variant of subscribeAuction (rows must remain unset).
subscribeEtfInav(tickers[, StreamOptions])	//blp/mktdata	Resolves and validates ETF iNAV tickers first, then streams them (fields defaults to ["LAST_PRICE"]).

All entry points return CompletableFuture<Subscription>.

Read scalar ticks

A Subscription extends SubscriptionView<Tick>, which implements Iterable<Tick> and AutoCloseable. You can consume ticks in four ways:

• Enhanced for loop (iterator()): blocks for each Tick until the subscription closes.
• stream(): returns a sequential java.util.stream.Stream<Tick> whose close() closes the subscription.
• Timed blocking read (next(Duration)): waits up to timeout and returns Optional<Tick> (empty at end of stream). If TimeoutException is thrown or the thread is interrupted, the in-flight read is carried over so the next read call receives its result without losing or overtaking data.
• Asynchronous read (nextAsync()): returns CompletableFuture<Optional<Tick>>. Cancelling a pending nextAsync() future closes the entire subscription.

Tick accessors and FieldHandle

Each Tick is an immutable, sparse update carrying only the fields present in that message:

Tick method	Return type	Behavior
topic()	String	The security's consumer label (its alias, or the subscribed topic).
timestampUs()	long	Update timestamp in microseconds since the UTC epoch.
layoutVersion()	int	Version of the native field layout used to decode this update.
fieldKind(field)	Optional<String>	"bool", "i32", "i64", "f64", "str", "date32", "time64_us", "timestamp_us", or "unknown".
has(field)	boolean	true when the update carries the field, including an explicit clear (null).
isNull(field)	boolean	true when the field is present as an explicit clear (null); false when absent or valued.
get(field)	Object	Boxed value (Boolean, Integer, Long, Double, String, LocalDate), or null when cleared or absent.
getDouble(field)	OptionalDouble	Value converted to a finite double, or empty when absent/cleared/non-numeric.
getLong(field)	OptionalLong	Value converted to a long, or empty when absent/cleared/non-integral.
getString(field)	Optional<String>	Value formatted as text, or empty when absent/cleared.
getLocalDate(field)	Optional<LocalDate>	date32 value as LocalDate, or empty when absent/cleared/not a date.
toMap()	Map<String, Object>	Unmodifiable map containing timestampUs, topic, and every present field (clears map to null).

Every field accessor accepts either a String field name or a reusable FieldHandle (sub.field("BID") or FieldHandle.of("BID")), which caches the field's index in the current layout:

import io.github.underloam.xbbg.Blp;
import io.github.underloam.xbbg.Xbbg;
import io.github.underloam.xbbg.subscription.FieldHandle;
import io.github.underloam.xbbg.subscription.Subscription;
import io.github.underloam.xbbg.subscription.Tick;
import java.time.Duration;
import java.util.Arrays;
import java.util.Optional;
import java.util.concurrent.TimeoutException;

public class TickReadingExample {
  public static void main(String[] args) throws TimeoutException {
    try (Subscription sub =
        Xbbg.await(
            Blp.subscribe(
                Arrays.asList("ES1 Index"),
                Arrays.asList("BID", "ASK", "LAST_PRICE")))) {
      FieldHandle bid = sub.field("BID");

      Optional<Tick> first = sub.next(Duration.ofSeconds(10));
      if (first.isPresent()) {
        Tick tick = first.get();
        if (tick.has(bid)) {
          System.out.println(
              tick.topic() + " BID=" + (tick.isNull(bid) ? "<cleared>" : tick.getDouble(bid)));
        }
      }
    }
  }
}

Read Arrow batches (sub.arrow())

sub.arrow() returns an ArrowSubscription (which extends SubscriptionView<Table>). Each read yields one Table batch combining up to flushThreshold already-queued updates without waiting for a batch to fill:

import io.github.underloam.xbbg.Blp;
import io.github.underloam.xbbg.Xbbg;
import io.github.underloam.xbbg.subscription.ArrowSubscription;
import io.github.underloam.xbbg.subscription.StreamOptions;
import io.github.underloam.xbbg.subscription.Subscription;
import io.github.underloam.xbbg.table.Table;
import java.util.Arrays;

public class ArrowSubscriptionExample {
  public static void main(String[] args) {
    StreamOptions options = StreamOptions.builder().flushThreshold(256).build();
    try (Subscription sub =
            Xbbg.await(
                Blp.subscribe(
                    Arrays.asList("ES1 Index", "NQ1 Index"),
                    Arrays.asList("BID", "ASK", "LAST_PRICE"),
                    options));
        ArrowSubscription batches = sub.arrow()) {
      for (Table batch : batches) {
        System.out.println("Batch rows=" + batch.numRows() + " cols=" + batch.columnNames());
        break;
      }
    }
  }
}

• Columns are timestamp, topic, the data fields, and a binary __xbbg_present column whose bit i (least-significant bit first) marks schema column i + 2 as present; a set bit with a null value is an explicit clear.
• One subscription can be read in only one mode: scalar Ticks, arrow(), or deliveries(). The first read locks the mode; attempting to read through another view throws IllegalStateException.

Stream options (StreamOptions)

Build options with StreamOptions.builder() (or AuctionStreamOptions.builder() for auction streams):

StreamOptions.Builder method	Default	Meaning
delivery(SubscriptionDeliveryMode)	DELTAS	DELTAS (sparse updates) or STATE (conflated current images; requires sub.deliveries()).
options(Collection<String>)	empty	Bloomberg subscription options such as "interval=5".
conflate(boolean)	false	Request Bloomberg upstream quote conflation (//blp/mktdata only; cannot be combined with interval).
flushThreshold(int)	Engine subscriptionFlushThreshold (16)	Maximum already-queued updates combined into one read batch.
overflowPolicy(OverflowPolicy)	Engine overflowPolicy (DROP_NEWEST)	DROP_NEWEST, BLOCK, or DROP_OLDEST (DROP_OLDEST requires sub.deliveries()).
streamCapacity(int)	Engine subscriptionStreamCapacity (256)	Per-consumer queue capacity.
allFields(boolean)	false	Expose every top-level scalar field Bloomberg sends on the feed.
fields(Collection<String>)	empty	Field list for mktbar, depth, chains, subscribeAuction, and subscribeEtfInav.
aliases(Map<String, String>)	empty	Map from upstream topic to consumer label used in ticks, status maps, and remove().
onDelayed(DelayedPolicy)	DelayedPolicy.WARN	WARN, RAISE (fails only the delayed topic), or IGNORE.
onFieldError(FieldErrorPolicy)	FieldErrorPolicy.WARN	WARN, RAISE (fails only topics with a rejected explicit field), or IGNORE.
isolated(boolean)	false	true gives this subscription its own private feed and session instead of sharing.
rows(boolean)	true	false disables row queueing and keeps only the current-value image for latest().
zeroAsNull(Collection<String>)	empty	Fields whose numeric zero (0 or -0.0) becomes null in latest() and STATE deliveries.

Shared feeds and dynamic membership

Within one Engine, //blp/mktdata subscriptions share one upstream Bloomberg feed when the service, trimmed topic, and normalized options match. You can add or remove topics and add fields on a live subscription:

import io.github.underloam.xbbg.Engine;
import io.github.underloam.xbbg.Xbbg;
import io.github.underloam.xbbg.subscription.StreamOptions;
import io.github.underloam.xbbg.subscription.Subscription;
import java.util.Collections;

public class DynamicMembershipExample {
  public static void main(String[] args) {
    try (Engine engine = Engine.open();
        Subscription sub =
            Xbbg.await(
                engine.subscribe(
                    Collections.singletonList("IBM UN Equity"),
                    Collections.singletonList("BID"),
                    StreamOptions.builder()
                        .aliases(Collections.singletonMap("IBM UN Equity", "IBM US Equity"))
                        .build()))) {
      Xbbg.await(sub.addFields(Collections.singletonList("ASK")));
      Xbbg.await(
          sub.add(
              Collections.singletonList("MSFT UW Equity"),
              Collections.singletonMap("MSFT UW Equity", "MSFT US Equity")));

      System.out.println("Labels: " + sub.tickers() + ", Fields: " + sub.fields());
      System.out.println("Feeds: " + engine.subscriptionFeeds());

      // Remove by consumer label
      Xbbg.await(sub.remove(Collections.singletonList("IBM US Equity")));
    }
  }
}

Current-value boards (latest() and rows(false))

sub.latest() (or sub.latest(Backend<T>)) returns the last known value of every field, one row per active topic, without consuming queued updates. Opening a subscription with rows(false) creates an image-only board that never queues rows and cannot overflow:

import io.github.underloam.xbbg.Backend;
import io.github.underloam.xbbg.Blp;
import io.github.underloam.xbbg.Xbbg;
import io.github.underloam.xbbg.subscription.StreamOptions;
import io.github.underloam.xbbg.subscription.Subscription;
import io.github.underloam.xbbg.table.RowsResult;
import io.github.underloam.xbbg.table.Table;
import java.util.Arrays;

public class LatestBoardExample {
  public static void main(String[] args) throws InterruptedException {
    StreamOptions boardOptions =
        StreamOptions.builder()
            .rows(false)
            .zeroAsNull(Arrays.asList("BID", "ASK"))
            .build();

    try (Subscription board =
        Xbbg.await(
            Blp.subscribe(
                Arrays.asList("IBM US Equity", "MSFT US Equity"),
                Arrays.asList("BID", "ASK", "LAST_PRICE"),
                boardOptions))) {
      Thread.sleep(1_000L);
      Table snapshot = board.latest();
      RowsResult rows = board.latest(Backend.ROWS);
      System.out.println("Board columns: " + snapshot.columnNames() + ", rows: " + rows.size());
    }
  }
}

Columns of latest() are topic, last_update, live, delayed, followed by the data fields.Boards opened with rows(false) and delivery(SubscriptionDeliveryMode.STATE) subscriptions recover automatically from Bloomberg DATALOSS events by resubscribing the shared feed and rebuilding the image.

Gap-aware deliveries (deliveries())

sub.deliveries() (or sub.deliveries(Backend<T>)) returns a DeliveryStream<T> yielding SubscriptionDelivery<T> envelopes:

• delivery.kind(): SubscriptionDeliveryMode.DELTAS or SubscriptionDeliveryMode.STATE.
• delivery.data(): the batch in the backend's representation (Table by default, or RowsResult, byte[], VectorSchemaRoot, etc.).
• delivery.gapBefore(): Optional<SubscriptionGap> with cause() ("local_overflow") and droppedUpdates() (exact count of source updates evicted before this batch under OverflowPolicy.DROP_OLDEST).

Two subscription configurations require deliveries():

1. delivery(SubscriptionDeliveryMode.STATE): conflates updates into current-state images for topics that changed (topic, last_update, live, delayed, valid, then the fields).
2. overflowPolicy(OverflowPolicy.DROP_OLDEST): keeps newest sparse deltas when the consumer queue is full and reports exact eviction counts in gapBefore().

import io.github.underloam.xbbg.Backend;
import io.github.underloam.xbbg.Blp;
import io.github.underloam.xbbg.OverflowPolicy;
import io.github.underloam.xbbg.SubscriptionDeliveryMode;
import io.github.underloam.xbbg.Xbbg;
import io.github.underloam.xbbg.subscription.DeliveryStream;
import io.github.underloam.xbbg.subscription.StreamOptions;
import io.github.underloam.xbbg.subscription.Subscription;
import io.github.underloam.xbbg.subscription.SubscriptionDelivery;
import io.github.underloam.xbbg.table.RowsResult;
import io.github.underloam.xbbg.table.Table;
import java.time.Duration;
import java.util.Arrays;
import java.util.Optional;
import java.util.concurrent.TimeoutException;

public class DeliveriesExample {
  public static void main(String[] args) throws TimeoutException {
    // 1. Conflated current-state images
    StreamOptions stateOpts =
        StreamOptions.builder().delivery(SubscriptionDeliveryMode.STATE).build();
    try (Subscription stateSub =
            Xbbg.await(
                Blp.subscribe(
                    Arrays.asList("IBM US Equity", "MSFT US Equity"),
                    Arrays.asList("BID", "ASK", "LAST_PRICE"),
                    stateOpts));
        DeliveryStream<RowsResult> stream = stateSub.deliveries(Backend.ROWS)) {
      Optional<SubscriptionDelivery<RowsResult>> next = stream.next(Duration.ofSeconds(10));
      next.ifPresent(d -> System.out.println(d.kind() + ": " + d.data()));
    }

    // 2. Drop-oldest deltas with exact gap counts
    StreamOptions dropOldestOpts =
        StreamOptions.builder().overflowPolicy(OverflowPolicy.DROP_OLDEST).build();
    try (Subscription deltaSub =
            Xbbg.await(Blp.subscribe("ES1 Index", "LAST_PRICE", dropOldestOpts));
        DeliveryStream<Table> deliveries = deltaSub.deliveries()) {
      Optional<SubscriptionDelivery<Table>> next = deliveries.next(Duration.ofSeconds(10));
      next.ifPresent(
          d -> {
            d.gapBefore()
                .ifPresent(
                    gap ->
                        System.out.println(
                            "Dropped "
                                + Long.toUnsignedString(gap.droppedUpdates())
                                + " updates ("
                                + gap.cause()
                                + ")"));
            System.out.println("Delivered rows: " + d.data().numRows());
          });
    }
  }
}

Java 17+ Reactive Streams and sealed types (xbbg-jdk17)

In io.github.underloam:xbbg-jdk17:

• XbbgFlow adapts subscription views to demand-driven single-subscriber java.util.concurrent.Flow.Publishers:
  • XbbgFlow.ticks(Subscription) -> Flow.Publisher<Tick>
  • XbbgFlow.arrow(ArrowSubscription) -> Flow.Publisher<Table>
  • XbbgFlow.deliveries(DeliveryStream<T>) -> Flow.Publisher<SubscriptionDelivery<T>>
• DeliveryView<T> (DeliveryView.of(SubscriptionDelivery<T>)) is a sealed interface with DeliveryView.Deltas<T>(T data, Optional<SubscriptionGap> gapBefore) and DeliveryView.State<T>(T data).
• TickValue and TickValues.get(Tick, field) return Optional<TickValue> over the sealed records BoolValue, IntValue, LongValue, DoubleValue, StringValue, DateValue, TimeValue (toLocalTime()), TimestampValue (toInstant()), and Cleared.
• VirtualThreads.useForCallbacks() installs a virtual-thread-per-task callback executor on Java 21+ (VirtualThreads.isSupported()).

import io.github.underloam.xbbg.Blp;
import io.github.underloam.xbbg.SubscriptionDeliveryMode;
import io.github.underloam.xbbg.Xbbg;
import io.github.underloam.xbbg.jdk17.DeliveryView;
import io.github.underloam.xbbg.jdk17.VirtualThreads;
import io.github.underloam.xbbg.jdk17.XbbgFlow;
import io.github.underloam.xbbg.subscription.StreamOptions;
import io.github.underloam.xbbg.subscription.Subscription;
import io.github.underloam.xbbg.subscription.SubscriptionDelivery;
import io.github.underloam.xbbg.table.Table;
import java.time.Duration;
import java.util.Optional;
import java.util.concurrent.Flow;
import java.util.concurrent.TimeoutException;

public class Jdk17StreamingExample {
  public static void main(String[] args) throws TimeoutException {
    VirtualThreads.useForCallbacks();

    StreamOptions opts =
        StreamOptions.builder().delivery(SubscriptionDeliveryMode.STATE).build();
    try (Subscription sub = Xbbg.await(Blp.subscribe("AAPL US Equity", "LAST_PRICE", opts))) {
      Optional<SubscriptionDelivery<Table>> delivery =
          sub.deliveries().next(Duration.ofSeconds(10));
      if (delivery.isPresent()) {
        DeliveryView<Table> view = DeliveryView.of(delivery.get());
        if (view instanceof DeliveryView.Deltas<Table> deltas) {
          System.out.println("Deltas rows=" + deltas.data().numRows() + " gap=" + deltas.gapBefore());
        } else if (view instanceof DeliveryView.State<Table> state) {
          System.out.println("State rows=" + state.data().numRows());
        }
      }

      // Or expose as a Reactive Streams Flow.Publisher
      Flow.Publisher<SubscriptionDelivery<Table>> publisher =
          XbbgFlow.deliveries(sub.deliveries());
      System.out.println("Created publisher: " + publisher);
    }
  }
}

Diagnostics and cleanup

Diagnostics

Every Subscription exposes live health and lifecycle snapshots:

Method	Return type	Contents
status()	SubscriptionStatus	Combined snapshot: events(), failures(), failedTickers(), topicStates(), fieldErrors(), session(), services(), and admin().
stats()	SubscriptionStats	messagesReceived(), droppedBatches(), batchesSent(), slowConsumer().
events()	List<SubscriptionEvent>	Bounded history (newest 128) with atUs(), category() (SubscriptionEventCategory), level() (SubscriptionEventLevel), messageType(), topic(), detail().
failures() / failedTickers()	List<SubscriptionFailure> / List<String>	Failed or terminated topics with topic(), reason(), kind() (SubscriptionFailureKind), atUs().
topicStates()	Map<String, TopicState>	Per label: topic(), feedTopic(), delayed(), state() (TopicLifecycleState), lastChangeUs(), streamsActive(), streamsChangedUs().
fieldErrors()	Map<String, Map<String, String>>	Label -> field -> Bloomberg error category.
sessionStatus()	SessionStatus	state() (SessionState), lastChangeUs(), disconnectCount(), reconnectCount().
serviceStatus()	Map<String, ServiceStatus>	Service URI -> ServiceStatus (service(), up(), lastChangeUs()).
adminStatus()	AdminStatus	slowConsumerWarningActive(), slowConsumerWarningCount(), slowConsumerClearedCount(), dataLossCount(), lastWarningUs(), lastClearedUs(), lastDataLossUs().
isActive(), tickers(), fields(), delivery()	boolean, List<String>, List<String>, SubscriptionDeliveryMode	Whether the subscription is active, its consumer labels, current field projection, and delivery mode.

Cleanup and draining

• sub.close() (called automatically by try-with-resources) closes the subscription and waits for cleanup, discarding unread updates.
• sub.unsubscribe() returns CompletableFuture<List<Tick>> (empty list) after closing without draining.
• sub.unsubscribe(true) closes the subscription and resolves to the buffered updates remaining in the view's format (List<Tick>, List<Table> on ArrowSubscription, or List<SubscriptionDelivery<T>> on DeliveryStream<T>).

Related pages

Configuration

Engine-wide stream capacity, overflow policy, flush threshold, and session limits.

Output Formats

Use Backend.ROWS or ArrowBackends.vectorSchemaRoot with latest() and deliveries().

Recipes

Auction and ETF iNAV subscriptions with venue and relationship preflight.

API Surface

Every class in io.github.underloam.xbbg.subscription and io.github.underloam.xbbg.jdk17.