Java

The Java client is built on grpc-java and Project Reactor. Streams are exposed as Flux<T> and typed responses use Java 21 sealed records so callers can pattern-match results exhaustively.

• Runtime library: com.dataspine:data-product-client. Contains Configuration, RawIngestClient, RawOutletClient, the typed Batch<T> / DataContainer<T> helpers, and the response records.
• Per-data-product client: generated by Dataspine from your published manifest. The placeholder used throughout this page is the artifact com.acme.orders:orders-client, which exposes a typed class OrdersClient for the com.acme.orders data product owned by an organization acme.

Installation

Maven:

<dependency>
  <groupId>com.dataspine</groupId>
  <artifactId>data-product-client</artifactId>
  <version>${dataspine.version}</version>
</dependency>
<dependency>
  <groupId>com.acme.orders</groupId>
  <artifactId>orders-client</artifactId>
  <version>${orders.version}</version>
</dependency>

Gradle (Kotlin DSL):

implementation("com.dataspine:data-product-client:$dataspineVersion")
implementation("com.acme.orders:orders-client:$ordersVersion")

Configuration

The runtime library uses Lightbend Config (HOCON). Drop a reference.conf (library defaults) or application.conf (application overrides) on your classpath:

client.name = "orders-app"

ingest {
  target = "https://orders.{{region}}.acme.example"
  use-secure-channel = true
}

outlet {
  target = "https://orders.{{region}}.acme.example"
  use-secure-channel = true
}

api {
  target = "https://orders.{{region}}.acme.example"
  use-secure-channel = true
}

sts.endpoint = "https://sts.acme.example"

auth {
  type = "static-token"
  token = ${?DATASPINE_TOKEN}
}

{{region}} and {{data_product_id}} are placeholders the runtime substitutes when each typed client is built. use-secure-channel = true selects TlsChannelCredentials; false falls back to InsecureChannelCredentials (intended for local development against a test data plane).

You can also build a Configuration directly with Configuration.builder() if you do not want HOCON.

import com.dataspine.client.config.Configuration;

Configuration configuration = new Configuration();

Typed client

The generator emits a class per data product that takes a Configuration, a region, and the data product UUID:

import com.acme.orders.OrdersClient;

OrdersClient ordersClient = new OrdersClient(
    configuration,
    "eu-central-1",
    UUID.fromString("00000000-0000-0000-0000-000000000000"));

It internally instantiates the underlying AbstractIngestClient, AbstractOutletClient, and the API client, and exposes one method per declared API operation, ingest, and outlet.

API call

import com.acme.orders.types.Order;
import com.acme.orders.types.PlaceOrder;

PlaceOrder request = new PlaceOrder(
    UUID.fromString("00000000-0000-0000-0000-000000000000"),
    List.of(new PlaceOrder.Item("BOOK-001", 1)));

ordersClient.placeOrder(request)
    .doOnNext(response -> {
        switch (response) {
            case ApiResponse.Success<Order> success -> log.info("placed: {}", success.payload().id());
            case ApiResponse.Error error -> log.error("failed: {}", error.message());
        }
    })
    .subscribe();

ApiResponse<T> is a sealed interface in the per-data-product client, so the switch is exhaustive at compile time. The error variant carries an ApiErrorCode mapping the ApiStatusCode values in the gRPC wire reference.

Ingest

The typed ingest API works in terms of DataContainer<T> (a Reactor Publisher-shaped wrapper that emits ingest commands). The most common implementation is Batch<T>:

import com.dataspine.client.ingest.typed.Batch;
import com.acme.orders.types.OrderEvent;
import reactor.core.publisher.Flux;

Flux<OrderEvent> orderEvents = orderEventSource();

ordersClient.ingestOrderEvents(Batch.of(orderEvents))
    .doOnNext(response -> {
        switch (response) {
            case BatchDeliveryResult br -> log.info("batch delivered: clientHandle={}, errors={}",
                br.clientHandle(), br.errors().size());
            case TransactionDeliveryResult tr -> log.info("transaction {} {}", tr.transactionId(), tr.responseType());
            case MessageDeliveryResult mr -> log.info("message {} delivered", mr.messageId());
            default -> log.info("response: {}", response);
        }
    })
    .blockLast();

For transactional producers, use Transaction.of(...) (or a similar typed wrapper if the generator produced one for your operation). The runtime library multiplexes a single bidirectional stream for all batches/transactions you submit through one client instance.

Outlet

The outlet poll returns a Flux<OutletMessage<T>>:

import com.dataspine.client.outlet.OutletMessage;

ordersClient.subscribeToOrderEvents(Offset.latest())
    .doOnNext(message -> {
        OrderEvent event = message.data();
        log.info("offset={} ts={} event={}", message.offset(), message.timestamp(), event);
    })
    .blockLast();

Offset.earliest(), Offset.latest(), and Offset.at(long offset) map to the Earliest / End / offset start values of OutletService.FetchMessages in the wire reference. Use at only with an offset returned in a previous OutletMessage for that outlet (or restored from your store of those values), not to seek to an arbitrary value—see Outlets: offsets and consumer position.

Backpressure is propagated end-to-end: requesting n items from the Flux causes the runtime to request n frames from the gRPC stream. Message chunking is handled inside RawOutletClient; subscribers always see the joined logical message.

Reauthentication

When the configured TokenProvider rotates a token, the runtime injects the new token via gRPC CallCredentials for fresh calls. Long-lived ingest and outlet streams use the Reauthenticate envelope from the gRPC wire reference so the call does not need to be torn down.

Shutdown

ordersClient.shutdown();

shutdown waits for in-flight calls to complete (it eventually delegates to ManagedChannel.shutdown); shutdownNow cancels them. In a Spring context, wiring shutdown into your application's lifecycle is sufficient.