Skip to main content

Rust

The Rust client is built on tonic and Tokio. It exposes async IngestClient, OutletClient, and (in the typed wrappers) per-API methods.

  • Runtime crate: dataspine-client (depends on tonic, prost, tokio, uuid, config). Provides Config, ConfigLoader, IngestClient, OutletClient, the bidirectional stream adapters that handle chunking and reauthentication, and the apply_authentication helper that injects Authorization: Bearer metadata.
  • Per-data-product crate: emitted by the Dataspine code generator. The placeholder used throughout this page is the crate acme-orders-client, exposing the typed OrdersClient for the com.acme.orders data product owned by an organization acme.

Installation​

[dependencies]
dataspine-client = "0.1"
acme-orders-client = "0.1"
tokio = { version = "1", features = ["full"] }
uuid = { version = "1", features = ["v4"] }

The per-data-product crate has the runtime crate as a dependency; pin compatible versions of both.

Configuration​

The runtime library reads configuration from (in priority order):

  • ~/.dataspine/config and ~/.dataspine/ingest.config
  • .dataspine.config and .dataspine.ingest.config in the current working directory
  • environment variables prefixed DATASPINE_

Fields:

  • region (required)
  • client_name (optional, used for logs)
  • endpoint_url (optional, defaults to https://api.{region}.dataspine.tech)
  • Auth — exactly one of:
    • token = "..." — a static JWT.
    • token_file = "/path/to/token" — file-backed token, refreshed every 60 seconds.
    • ssh_ed25519_token_exchange_endpoint = "...", ssh_ed25519_principal_id = "...", ssh_ed25519_private_key_file = "...", ssh_ed25519_public_key_file = "..." — SSH-Ed25519 PoP token exchange.
# .dataspine.config
region = "eu-central-1"
client_name = "orders-app"
token_file = "/var/run/secrets/dataspine/token"
use dataspine_client::config::{behavior_version::BehaviorVersion, config_loader::ConfigLoader};

let config = ConfigLoader::new(BehaviorVersion::latest()).load().await?;

config carries an authentication_provider (a tokio::sync::watch::Receiver<AuthenticationStatus>) that all clients subscribe to so a token refresh is observed by every active stream.

Typed client​

The generator emits one struct per data product:

use acme_orders_client::OrdersClient;
use uuid::Uuid;

let data_product_id = Uuid::parse_str("00000000-0000-0000-0000-000000000000")?;
let mut orders = OrdersClient::new(&config, data_product_id).await?;

It builds the underlying IngestClient and OutletClient and pre-resolves the operation IDs declared by the data product manifest.

API call​

use acme_orders_client::operations::{place_order::PlaceOrder, place_order::Order};

let response = orders
    .place_order(PlaceOrder {
        customer_id: Uuid::parse_str("00000000-0000-0000-0000-000000000000")?,
        items: vec![place_order::Item { sku: "BOOK-001".into(), quantity: 1 }],
    })
    .await?;

match response {
    Ok(success) => println!("placed: {}", success.payload.id),
    Err(error) => eprintln!("failed: {} {}", error.status as i32, error.message),
}

The typed wrapper returns Result<ApiSuccessResponse<T>, ApiErrorResponse>. The error variant carries an ApiErrorCode that maps the ApiStatusCode values in the gRPC wire reference.

Ingest​

use dataspine_client::ingest_client::ingest_client::IngestClient;
use acme_orders_client::types::OrderEvent;

let mut channel = orders.open_order_event_ingest().await?;
let (mut tx, mut rx) = channel.create_batch_channel(()).await?;

tx.send(OrderEvent { /* ... */ }).await?;
tx.send(OrderEvent { /* ... */ }).await?;
tx.end().await?;

while let Some(response) = rx.recv().await {
    println!("delivery result: {:?}", response);
}

channel.join().await?;

AppendMessageChannel multiplexes many batches over one bidirectional gRPC stream. create_batch_channel returns a (BatchChannelSender<O>, BatchChannelReceiver<O>) pair where the per-batch O is opaque metadata you supply (e.g. an offset to commit on success).

Outlet​

use dataspine_client::outlet_client::outlet_client::{OffsetType, OutletClient};
use futures_util::StreamExt;

let mut outlet = orders.subscribe_to_order_events().await?;
let mut stream = outlet.poll(orders.order_events_outlet_version_id(), OffsetType::Latest).await?;

while let Some(event) = stream.next().await {
    let event: OrderEvent = event?;
    println!("{event:?}");
}

JoiningStream joins chunked frames into single logical messages. The poll honors back-pressure: it only requests the next frame after the consumer pulls the previous one.

Reauthentication​

The runtime watches Config::authentication_provider. When the watch channel ticks with a new token, every active ingest and outlet stream pushes a Reauthenticate envelope (see the gRPC wire reference) so the server can re-validate without dropping the stream. Unary API calls always pick up the latest token because apply_authentication reads from the same channel.

Shutdown​

Drop the OrdersClient (or its constituent IngestClient / OutletClient). Background tasks for reauth and the multiplexer terminate when their channels close.

REST​

Today the Rust client uses gRPC for all API operations. There is no equivalent of the TypeScript RestApiClient. If you need to invoke a supports_raw_response operation from Rust today, use reqwest (or another HTTP client) against the REST wire reference (see raw response operations). A typed wrapper for raw responses is on the roadmap.