// 1. Prepare the query
final IPreparedListQuery query =
    database
        .getQueryManager()
        .listQuery()
        .forTable("trades")
        .withCondition(BaseConditions.equal(FieldPath.of("city"), "Paris"))
        .withTableFields("tradeId", "trader", "value")
        .compile();
// 2. Create the listener
final IStreamListViewListener listener = new TradeStreamListener();
// 3. Register on the data streamer
final IStreamedViewListenerRegistration registration =
    database
        .getDataStreamer()
        .registerListStreamView(query, listener, RegistrationOptions.defaults());
// 4. Wait for the registration to complete (including initial view if enabled)
registration.getAttachmentProcess().toCompletableFuture().get();
// ... the listener is now receiving updates on every commit ...
// 5. Unregister when done
registration.unregister();

static class TradeStreamListener implements IStreamListViewListener {
  // Thread-safe collection: callbacks may be invoked concurrently for different partitions
  private final Queue<List<Object>> receivedRecords = new ConcurrentLinkedQueue<>();
  @Override
  public void transactionStarted(final IEpoch epoch, final ITransactionInformation info) {
    // Called at the beginning of a commit
  }
  @Override
  public void transactionCommitted(final IEpoch epoch) {
    // Called after the transaction successfully commits
  }
  @Override
  public void transactionRolledBack() {
    // Called if the transaction rolls back
  }
  @Override
  public void recordsAdded(
      final int partitionId, final IRecordBlock<? extends IRecordReader> records) {
    // Important: record data may be backed by a reused buffer
    // Clone the data if you need to keep it beyond this callback
    for (final IRecordReader r : records) {
      this.receivedRecords.add(r.toList());
    }
  }
  @Override
  public void recordsDeleted(
      final int partitionId, final IRecordBlock<? extends IRecordReader> records) {
    // Handle deleted records
  }
  @Override
  public void recordsUpdated(
      final int partitionId,
      final IRecordBlock<? extends IRecordReader> oldValues,
      final IRecordBlock<? extends IRecordReader> newValues) {
    // Handle updated records — both old and new values are provided
  }
  @Override
  public void partitionDropped(
      final int partitionId,
      final List<int[]> dropConditionsFields,
      final IRecordBlock<? extends IRecordReader> records) {
    // Handle dropped partitions
  }
  @Override
  public void truncate() {
    // Handle truncation (all data removed)
    this.receivedRecords.clear();
  }
  @Override
  public void onSelfFailure(final Throwable error) {
    // Called if any of the above callbacks throws an exception
  }
}

final RegistrationOptions options =
    RegistrationOptions.builder()
        .sendInitialView(true) // send a snapshot of current data before streaming deltas
        .branch(IEpoch.MASTER_BRANCH_NAME) // only listen to changes on the master branch
        .dictionarized(false) // receive plain values, not dictionary-encoded
        .build();